qvm_volume.py 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU Lesser General Public License as published by
  10. # the Free Software Foundation; either version 2.1 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. # GNU Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. import qubesadmin.tests
  21. import qubesadmin.tests.tools
  22. import qubesadmin.tools.qvm_volume
  23. class TC_00_qvm_volume(qubesadmin.tests.QubesTestCase):
  24. def setup_expected_calls_for_list(self, vms=('vm1', 'sys-net')):
  25. self.app.expected_calls[
  26. ('dom0', 'mgmt.vm.List', None, None)] = \
  27. b'0\x00vm1 class=AppVM state=Running\n' \
  28. b'sys-net class=AppVM state=Running\n'
  29. for vm in vms:
  30. for vol in ('root', 'private'):
  31. self.app.expected_calls[
  32. (vm, 'mgmt.vm.volume.Info', vol, None)] = \
  33. b'0\x00' + \
  34. (b'pool=pool-file\n' if vol == 'root' else
  35. b'pool=other-pool\n') + \
  36. b'vid=' + vm.encode() + b'-' + vol.encode() + b'\n' \
  37. b'internal=True\n' \
  38. b'size=10737418240\n'
  39. self.app.expected_calls[
  40. (vm, 'mgmt.vm.volume.ListSnapshots', vol, None)] = \
  41. b'0\x00snap1\n' if vol == 'private' else b'0\x00'
  42. self.app.expected_calls[
  43. (vm, 'mgmt.vm.volume.List', None, None)] = \
  44. b'0\x00root\nprivate\n'
  45. def test_000_list(self):
  46. self.setup_expected_calls_for_list()
  47. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  48. self.assertEqual(0,
  49. qubesadmin.tools.qvm_volume.main(['ls', '-i'], app=self.app))
  50. self.assertEqual(stdout.getvalue(),
  51. 'POOL:VOLUME VMNAME VOLUME_NAME '
  52. 'REVERT_POSSIBLE\n'
  53. 'other-pool:sys-net-private sys-net private Yes\n'
  54. 'other-pool:vm1-private vm1 private Yes\n'
  55. 'pool-file:sys-net-root sys-net root No\n'
  56. 'pool-file:vm1-root vm1 root No\n'
  57. )
  58. self.assertAllCalled()
  59. def test_001_list_domain(self):
  60. self.setup_expected_calls_for_list(vms=('vm1',))
  61. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  62. self.assertEqual(0,
  63. qubesadmin.tools.qvm_volume.main(['ls', '-i', 'vm1'],
  64. app=self.app))
  65. self.assertEqual(stdout.getvalue(),
  66. 'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
  67. 'other-pool:vm1-private vm1 private Yes\n'
  68. 'pool-file:vm1-root vm1 root No\n'
  69. )
  70. self.assertAllCalled()
  71. def test_002_list_domain_pool(self):
  72. self.setup_expected_calls_for_list(vms=('vm1',))
  73. self.app.expected_calls[('dom0', 'mgmt.pool.List', None, None)] = \
  74. b'0\x00pool-file\nother-pool\n'
  75. del self.app.expected_calls[
  76. ('vm1', 'mgmt.vm.volume.ListSnapshots', 'private', None)]
  77. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  78. self.assertEqual(0,
  79. qubesadmin.tools.qvm_volume.main(
  80. ['ls', '-i', '-p', 'pool-file', 'vm1'],
  81. app=self.app))
  82. self.assertEqual(stdout.getvalue(),
  83. 'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
  84. 'pool-file:vm1-root vm1 root No\n'
  85. )
  86. self.assertAllCalled()
  87. def test_003_list_pool(self):
  88. self.setup_expected_calls_for_list()
  89. self.app.expected_calls[('dom0', 'mgmt.pool.List', None, None)] = \
  90. b'0\x00pool-file\nother-pool\n'
  91. del self.app.expected_calls[
  92. ('vm1', 'mgmt.vm.volume.ListSnapshots', 'private', None)]
  93. del self.app.expected_calls[
  94. ('sys-net', 'mgmt.vm.volume.ListSnapshots', 'private', None)]
  95. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  96. self.assertEqual(0,
  97. qubesadmin.tools.qvm_volume.main(
  98. ['ls', '-i', '-p', 'pool-file'],
  99. app=self.app))
  100. self.assertEqual(stdout.getvalue(),
  101. 'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
  102. 'pool-file:sys-net-root sys-net root No\n'
  103. 'pool-file:vm1-root vm1 root No\n'
  104. )
  105. self.assertAllCalled()
  106. def test_004_list_multiple_domains(self):
  107. self.setup_expected_calls_for_list(vms=('vm1', 'vm2'))
  108. self.app.expected_calls[
  109. ('dom0', 'mgmt.vm.List', None, None)] = \
  110. b'0\x00vm1 class=AppVM state=Running\n' \
  111. b'vm2 class=AppVM state=Running\n' \
  112. b'vm3 class=AppVM state=Running\n'
  113. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  114. self.assertEqual(0,
  115. qubesadmin.tools.qvm_volume.main(
  116. ['ls', '-i', 'vm1', 'vm2'], app=self.app))
  117. self.assertEqual(stdout.getvalue(),
  118. 'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
  119. 'other-pool:vm1-private vm1 private Yes\n'
  120. 'other-pool:vm2-private vm2 private Yes\n'
  121. 'pool-file:vm1-root vm1 root No\n'
  122. 'pool-file:vm2-root vm2 root No\n'
  123. )
  124. self.assertAllCalled()
  125. def test_010_extend(self):
  126. self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \
  127. b'0\x00testvm class=AppVM state=Running\n'
  128. self.app.expected_calls[
  129. ('testvm', 'mgmt.vm.volume.List', None, None)] = \
  130. b'0\x00root\nprivate\n'
  131. self.app.expected_calls[
  132. ('testvm', 'mgmt.vm.volume.Resize', 'private', b'10737418240')] = \
  133. b'0\x00'
  134. self.assertEqual(0,
  135. qubesadmin.tools.qvm_volume.main(
  136. ['extend', 'testvm:private', '10GiB'],
  137. app=self.app))
  138. self.assertAllCalled()
  139. def test_011_extend_error(self):
  140. self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \
  141. b'0\x00testvm class=AppVM state=Running\n'
  142. self.app.expected_calls[
  143. ('testvm', 'mgmt.vm.volume.List', None, None)] = \
  144. b'0\x00root\nprivate\n'
  145. self.app.expected_calls[
  146. ('testvm', 'mgmt.vm.volume.Resize', 'private', b'1073741824')] = \
  147. b'2\x00StoragePoolException\x00\x00Failed to resize volume: ' \
  148. b'shrink not allowed\x00'
  149. with qubesadmin.tests.tools.StderrBuffer() as stderr:
  150. self.assertEqual(1,
  151. qubesadmin.tools.qvm_volume.main(
  152. ['extend', 'testvm:private', '1GiB'],
  153. app=self.app))
  154. self.assertIn('shrink not allowed', stderr.getvalue())
  155. self.assertAllCalled()