storage.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347
  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 subprocess
  21. import qubesadmin.tests
  22. import qubesadmin.storage
  23. class TestVMVolume(qubesadmin.tests.QubesTestCase):
  24. def setUp(self):
  25. super(TestVMVolume, self).setUp()
  26. self.vol = qubesadmin.storage.Volume(self.app, vm='test-vm',
  27. vm_name='volname')
  28. self.pool_vol = qubesadmin.storage.Volume(self.app, pool='test-pool',
  29. vid='some-id')
  30. def expect_info(self):
  31. self.app.expected_calls[
  32. ('test-vm', 'admin.vm.volume.Info', 'volname', None)] = \
  33. b'0\x00' \
  34. b'pool=test-pool\n' \
  35. b'vid=some-id\n' \
  36. b'size=1024\n' \
  37. b'usage=512\n' \
  38. b'rw=True\n' \
  39. b'snap_on_start=True\n' \
  40. b'save_on_stop=True\n' \
  41. b'source=\n' \
  42. b'revisions_to_keep=3\n'
  43. def test_000_qubesd_call(self):
  44. self.app.expected_calls[
  45. ('test-vm', 'admin.vm.volume.TestMethod', 'volname', None)] = \
  46. b'0\x00method_result'
  47. self.assertEqual(self.vol._qubesd_call('TestMethod'),
  48. b'method_result')
  49. self.assertAllCalled()
  50. def test_001_fetch_info(self):
  51. self.app.expected_calls[
  52. ('test-vm', 'admin.vm.volume.Info', 'volname', None)] = \
  53. b'0\x00prop1=val1\nprop2=val2\n'
  54. self.vol._fetch_info()
  55. self.assertEqual(self.vol._info, {'prop1': 'val1', 'prop2': 'val2'})
  56. self.assertAllCalled()
  57. def test_010_pool(self):
  58. self.expect_info()
  59. self.assertEqual(self.vol.pool, 'test-pool')
  60. self.assertAllCalled()
  61. def test_011_vid(self):
  62. self.expect_info()
  63. self.assertEqual(self.vol.vid, 'some-id')
  64. self.assertAllCalled()
  65. def test_012_size(self):
  66. self.expect_info()
  67. self.assertEqual(self.vol.size, 1024)
  68. self.assertAllCalled()
  69. def test_013_usage(self):
  70. self.expect_info()
  71. self.assertEqual(self.vol.usage, 512)
  72. self.assertAllCalled()
  73. def test_014_rw(self):
  74. self.expect_info()
  75. self.assertEqual(self.vol.rw, True)
  76. self.assertAllCalled()
  77. def test_015_snap_on_start(self):
  78. self.expect_info()
  79. self.assertEqual(self.vol.snap_on_start, True)
  80. self.assertAllCalled()
  81. def test_016_save_on_stop(self):
  82. self.expect_info()
  83. self.assertEqual(self.vol.save_on_stop, True)
  84. self.assertAllCalled()
  85. def test_017_source_none(self):
  86. self.expect_info()
  87. self.assertEqual(self.vol.source, None)
  88. self.assertAllCalled()
  89. def test_018_source(self):
  90. self.expect_info()
  91. call_key = list(self.app.expected_calls)[0]
  92. self.app.expected_calls[call_key] = self.app.expected_calls[
  93. call_key].replace(b'source=\n', b'source=test-pool:other-id\n')
  94. self.assertEqual(self.vol.source, 'test-pool:other-id')
  95. self.assertAllCalled()
  96. def test_020_revisions_to_keep(self):
  97. self.expect_info()
  98. self.assertEqual(self.vol.revisions_to_keep, 3)
  99. self.assertAllCalled()
  100. def test_021_revisions(self):
  101. self.app.expected_calls[
  102. ('test-vm', 'admin.vm.volume.ListSnapshots', 'volname', None)] = \
  103. b'0\x00' \
  104. b'snapid1\n' \
  105. b'snapid2\n' \
  106. b'snapid3\n'
  107. self.assertEqual(self.vol.revisions,
  108. ['snapid1', 'snapid2', 'snapid3'])
  109. self.assertAllCalled()
  110. def test_022_revisions_empty(self):
  111. self.app.expected_calls[
  112. ('test-vm', 'admin.vm.volume.ListSnapshots', 'volname', None)] = \
  113. b'0\x00'
  114. self.assertEqual(self.vol.revisions, [])
  115. self.assertAllCalled()
  116. def test_030_resize(self):
  117. self.app.expected_calls[
  118. ('test-vm', 'admin.vm.volume.Resize', 'volname', b'2048')] = b'0\x00'
  119. self.vol.resize(2048)
  120. self.assertAllCalled()
  121. def test_031_revert(self):
  122. self.app.expected_calls[
  123. ('test-vm', 'admin.vm.volume.Revert', 'volname', b'snapid1')] = \
  124. b'0\x00'
  125. self.vol.revert('snapid1')
  126. self.assertAllCalled()
  127. def test_040_import_data(self):
  128. self.app.expected_calls[
  129. ('test-vm', 'admin.vm.volume.Import', 'volname', b'some-data')] = \
  130. b'0\x00'
  131. input_proc = subprocess.Popen(['echo', '-n', 'some-data'],
  132. stdout=subprocess.PIPE)
  133. self.vol.import_data(input_proc.stdout)
  134. input_proc.stdout.close()
  135. self.assertAllCalled()
  136. def test_050_clone(self):
  137. self.app.expected_calls[
  138. ('source-vm', 'admin.vm.volume.CloneFrom', 'volname', None)] = \
  139. b'0\x00abcdef'
  140. self.app.expected_calls[
  141. ('test-vm', 'admin.vm.volume.CloneTo', 'volname', b'abcdef')] = \
  142. b'0\x00'
  143. self.app.expected_calls[
  144. ('dom0', 'admin.vm.List', None, None)] = \
  145. b'0\x00source-vm class=AppVM state=Halted\n'
  146. self.app.expected_calls[
  147. ('source-vm', 'admin.vm.volume.List', None, None)] = \
  148. b'0\x00volname\nother\n'
  149. self.vol.clone(self.app.domains['source-vm'].volumes['volname'])
  150. self.assertAllCalled()
  151. class TestPoolVolume(TestVMVolume):
  152. def setUp(self):
  153. super(TestPoolVolume, self).setUp()
  154. self.vol = qubesadmin.storage.Volume(self.app, pool='test-pool',
  155. vid='some-id')
  156. def test_000_qubesd_call(self):
  157. self.app.expected_calls[
  158. ('dom0', 'admin.pool.volume.TestMethod',
  159. 'test-pool', b'some-id')] = \
  160. b'0\x00method_result'
  161. self.assertEqual(self.vol._qubesd_call('TestMethod'),
  162. b'method_result')
  163. self.assertAllCalled()
  164. def expect_info(self):
  165. self.app.expected_calls[
  166. ('dom0', 'admin.pool.volume.Info', 'test-pool', b'some-id')] = \
  167. b'0\x00' \
  168. b'pool=test-pool\n' \
  169. b'vid=some-id\n' \
  170. b'size=1024\n' \
  171. b'usage=512\n' \
  172. b'rw=True\n' \
  173. b'snap_on_start=True\n' \
  174. b'save_on_stop=True\n' \
  175. b'source=\n' \
  176. b'revisions_to_keep=3\n'
  177. def test_001_fetch_info(self):
  178. self.app.expected_calls[
  179. ('dom0', 'admin.pool.volume.Info', 'test-pool',
  180. b'some-id')] = \
  181. b'0\x00prop1=val1\nprop2=val2\n'
  182. self.vol._fetch_info()
  183. self.assertEqual(self.vol._info, {'prop1': 'val1', 'prop2': 'val2'})
  184. self.assertAllCalled()
  185. def test_010_pool(self):
  186. # this should _not_ produce any api call, as pool is already known
  187. self.assertEqual(self.vol.pool, 'test-pool')
  188. self.assertAllCalled()
  189. def test_011_vid(self):
  190. # this should _not_ produce any api call, as vid is already known
  191. self.assertEqual(self.vol.vid, 'some-id')
  192. self.assertAllCalled()
  193. def test_021_revisions(self):
  194. self.app.expected_calls[
  195. ('dom0', 'admin.pool.volume.ListSnapshots',
  196. 'test-pool', b'some-id')] = \
  197. b'0\x00' \
  198. b'snapid1\n' \
  199. b'snapid2\n' \
  200. b'snapid3\n'
  201. self.assertEqual(self.vol.revisions,
  202. ['snapid1', 'snapid2', 'snapid3'])
  203. self.assertAllCalled()
  204. def test_022_revisions_empty(self):
  205. self.app.expected_calls[
  206. ('dom0', 'admin.pool.volume.ListSnapshots',
  207. 'test-pool', b'some-id')] = b'0\x00'
  208. self.assertEqual(self.vol.revisions, [])
  209. self.assertAllCalled()
  210. def test_030_resize(self):
  211. self.app.expected_calls[
  212. ('dom0', 'admin.pool.volume.Resize',
  213. 'test-pool', b'some-id 2048')] = b'0\x00'
  214. self.vol.resize(2048)
  215. self.assertAllCalled()
  216. def test_031_revert(self):
  217. self.app.expected_calls[
  218. ('dom0', 'admin.pool.volume.Revert', 'test-pool',
  219. b'some-id snapid1')] = b'0\x00'
  220. self.vol.revert('snapid1')
  221. self.assertAllCalled()
  222. def test_040_import_data(self):
  223. self.skipTest('admin.pool.volume.Import not supported')
  224. def test_050_clone(self):
  225. self.app.expected_calls[
  226. ('dom0', 'admin.pool.volume.CloneFrom', 'test-pool',
  227. b'volid')] = b'0\x00abcdef'
  228. self.app.expected_calls[
  229. ('dom0', 'admin.pool.volume.CloneTo', 'test-pool',
  230. b'some-id abcdef')] = b'0\x00'
  231. source_vol = qubesadmin.storage.Volume(self.app, pool='test-pool',
  232. vid='volid')
  233. self.vol.clone(source_vol)
  234. self.assertAllCalled()
  235. def test_050_clone_wrong_volume(self):
  236. self.skipTest('admin.pool.volume.Clone not supported')
  237. class TestPool(qubesadmin.tests.QubesTestCase):
  238. def test_000_list(self):
  239. self.app.expected_calls[('dom0', 'admin.pool.List', None, None)] = \
  240. b'0\x00file\nlvm\n'
  241. seen = set()
  242. for pool in self.app.pools:
  243. self.assertIsInstance(pool, qubesadmin.storage.Pool)
  244. self.assertIn(pool.name, ('file', 'lvm'))
  245. self.assertNotIn(pool.name, seen)
  246. seen.add(pool.name)
  247. self.assertEqual(seen, set(['file', 'lvm']))
  248. self.assertAllCalled()
  249. def test_010_config(self):
  250. self.app.expected_calls[('dom0', 'admin.pool.List', None, None)] = \
  251. b'0\x00file\nlvm\n'
  252. self.app.expected_calls[('dom0', 'admin.pool.Info', 'file', None)] = \
  253. b'0\x00driver=file\n' \
  254. b'dir_path=/var/lib/qubes\n' \
  255. b'name=file\n' \
  256. b'revisions_to_keep=3\n'
  257. pool = self.app.pools['file']
  258. self.assertEqual(pool.config, {
  259. 'driver': 'file',
  260. 'dir_path': '/var/lib/qubes',
  261. 'name': 'file',
  262. 'revisions_to_keep': '3',
  263. })
  264. self.assertAllCalled()
  265. def test_020_volumes(self):
  266. self.app.expected_calls[('dom0', 'admin.pool.List', None, None)] = \
  267. b'0\x00file\nlvm\n'
  268. self.app.expected_calls[
  269. ('dom0', 'admin.pool.volume.List', 'file', None)] = \
  270. b'0\x00vol1\n' \
  271. b'vol2\n'
  272. pool = self.app.pools['file']
  273. seen = set()
  274. for volume in pool.volumes:
  275. self.assertIsInstance(volume, qubesadmin.storage.Volume)
  276. self.assertIn(volume.vid, ('vol1', 'vol2'))
  277. self.assertEqual(volume.pool, 'file')
  278. self.assertNotIn(volume.vid, seen)
  279. seen.add(volume.vid)
  280. self.assertEqual(seen, set(['vol1', 'vol2']))
  281. self.assertAllCalled()
  282. def test_030_pool_drivers(self):
  283. self.app.expected_calls[
  284. ('dom0', 'admin.pool.ListDrivers', None, None)] = \
  285. b'0\x00file dir_path revisions_to_keep\n' \
  286. b'lvm volume_group thin_pool revisions_to_keep\n'
  287. self.assertEqual(set(self.app.pool_drivers), set(['file', 'lvm']))
  288. self.assertEqual(set(self.app.pool_driver_parameters('file')),
  289. set(['dir_path', 'revisions_to_keep']))
  290. self.assertAllCalled()
  291. def test_040_add(self):
  292. self.app.expected_calls[
  293. ('dom0', 'admin.pool.Add', 'some-driver',
  294. b'name=test-pool\nparam1=value1\nparam2=123\n')] = b'0\x00'
  295. self.app.add_pool('test-pool', driver='some-driver',
  296. param1='value1', param2=123)
  297. self.assertAllCalled()
  298. def test_050_remove(self):
  299. self.app.expected_calls[
  300. ('dom0', 'admin.pool.Remove', 'test-pool', None)] = b'0\x00'
  301. self.app.remove_pool('test-pool')
  302. self.assertAllCalled()