storage_file.py 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658
  1. #
  2. # The Qubes OS Project, https://www.qubes-os.org/
  3. #
  4. # Copyright (C) 2015 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
  5. #
  6. # This library is free software; you can redistribute it and/or
  7. # modify it under the terms of the GNU Lesser General Public
  8. # License as published by the Free Software Foundation; either
  9. # version 2.1 of the License, or (at your option) any later version.
  10. #
  11. # This library is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14. # Lesser General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU Lesser General Public
  17. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  18. #
  19. ''' Tests for the file storage backend '''
  20. import os
  21. import shutil
  22. import asyncio
  23. import unittest.mock
  24. import subprocess
  25. import qubes.storage
  26. import qubes.utils
  27. import qubes.tests.storage
  28. from qubes.config import defaults
  29. # :pylint: disable=invalid-name
  30. class TestApp(qubes.Qubes):
  31. ''' A Mock App object '''
  32. def __init__(self, *args, **kwargs): # pylint: disable=unused-argument
  33. super(TestApp, self).__init__('/tmp/qubes-test.xml', load=False,
  34. offline_mode=True, **kwargs)
  35. self.load_initial_values()
  36. self.pools['linux-kernel'].dir_path = '/tmp/qubes-test-kernel'
  37. dummy_kernel = os.path.join(self.pools['linux-kernel'].dir_path,
  38. 'dummy')
  39. os.makedirs(dummy_kernel, exist_ok=True)
  40. open(os.path.join(dummy_kernel, 'vmlinuz'), 'w').close()
  41. open(os.path.join(dummy_kernel, 'modules.img'), 'w').close()
  42. open(os.path.join(dummy_kernel, 'initramfs'), 'w').close()
  43. self.default_kernel = 'dummy'
  44. def cleanup(self):
  45. ''' Remove temporary directories '''
  46. shutil.rmtree(self.pools['linux-kernel'].dir_path)
  47. if os.path.exists(self.store):
  48. os.unlink(self.store)
  49. def create_dummy_template(self):
  50. ''' Initalizes a dummy TemplateVM as the `default_template` '''
  51. template = self.add_new_vm(qubes.vm.templatevm.TemplateVM,
  52. name='test-template', label='red',
  53. memory=1024, maxmem=1024)
  54. self.default_template = template
  55. class TC_00_FilePool(qubes.tests.QubesTestCase):
  56. """ This class tests some properties of the 'default' pool.
  57. This test might become obsolete if we change the driver for the default
  58. pool to something else as 'file'.
  59. """
  60. def setUp(self):
  61. super(TC_00_FilePool, self).setUp()
  62. self.app = TestApp()
  63. def tearDown(self):
  64. self.app.cleanup()
  65. self.app.close()
  66. del self.app
  67. super(TC_00_FilePool, self).tearDown()
  68. def test000_default_pool_dir(self):
  69. """ The predefined dir for the default pool should be ``/var/lib/qubes``
  70. .. sealso::
  71. Data :data:``qubes.qubes.defaults['pool_config']``.
  72. """
  73. result = self.app.get_pool("varlibqubes").dir_path
  74. expected = '/var/lib/qubes'
  75. self.assertEqual(result, expected)
  76. def test001_default_storage_class(self):
  77. """ Check when using default pool the Storage is
  78. ``qubes.storage.Storage``. """
  79. result = self._init_app_vm().storage
  80. self.assertIsInstance(result, qubes.storage.Storage)
  81. def _init_app_vm(self):
  82. """ Return initalised, but not created, AppVm. """
  83. vmname = self.make_vm_name('appvm')
  84. self.app.create_dummy_template()
  85. return self.app.add_new_vm(qubes.vm.appvm.AppVM, name=vmname,
  86. template=self.app.default_template,
  87. label='red')
  88. class TC_01_FileVolumes(qubes.tests.QubesTestCase):
  89. ''' Test correct handling of different types of volumes '''
  90. POOL_DIR = '/tmp/test-pool'
  91. POOL_NAME = 'test-pool'
  92. POOL_CONF = {'driver': 'file', 'dir_path': POOL_DIR, 'name': POOL_NAME}
  93. def setUp(self):
  94. """ Add a test file based storage pool """
  95. super(TC_01_FileVolumes, self).setUp()
  96. self.app = TestApp()
  97. self.loop.run_until_complete(self.app.add_pool(**self.POOL_CONF))
  98. self.app.default_pool = self.app.get_pool(self.POOL_NAME)
  99. self.app.create_dummy_template()
  100. def tearDown(self):
  101. """ Remove the file based storage pool after testing """
  102. for vm in list(self.app.domains):
  103. if vm.name.startswith(qubes.tests.VMPREFIX):
  104. del self.app.domains[vm]
  105. self.app.default_template = None
  106. del self.app.domains['test-template']
  107. self.app.default_pool = 'varlibqubes'
  108. self.loop.run_until_complete(self.app.remove_pool("test-pool"))
  109. self.app.cleanup()
  110. self.app.close()
  111. del self.app
  112. super(TC_01_FileVolumes, self).tearDown()
  113. shutil.rmtree(self.POOL_DIR, ignore_errors=True)
  114. def test_000_origin_volume(self):
  115. config = {
  116. 'name': 'root',
  117. 'pool': self.POOL_NAME,
  118. 'save_on_stop': True,
  119. 'rw': True,
  120. 'size': defaults['root_img_size'],
  121. }
  122. vm = qubes.tests.storage.TestVM(self)
  123. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  124. self.assertEqual(volume.name, 'root')
  125. self.assertEqual(volume.pool, self.POOL_NAME)
  126. self.assertEqual(volume.size, defaults['root_img_size'])
  127. self.assertFalse(volume.snap_on_start)
  128. self.assertTrue(volume.save_on_stop)
  129. self.assertTrue(volume.rw)
  130. block = volume.block_device()
  131. self.assertEqual(block.path,
  132. '{base}.img:{base}-cow.img'.format(
  133. base=self.POOL_DIR + '/appvms/' + vm.name + '/root'))
  134. self.assertEqual(block.script, 'block-origin')
  135. self.assertEqual(block.rw, True)
  136. self.assertEqual(block.name, 'root')
  137. self.assertEqual(block.devtype, 'disk')
  138. self.assertIsNone(block.domain)
  139. def test_001_snapshot_volume(self):
  140. template_vm = self.app.default_template
  141. vm = qubes.tests.storage.TestVM(self, template=template_vm)
  142. original_size = qubes.config.defaults['root_img_size']
  143. source_config = {
  144. 'name': 'root',
  145. 'pool': self.POOL_NAME,
  146. 'save_on_stop': True,
  147. 'rw': False,
  148. 'size': original_size,
  149. }
  150. source = self.app.get_pool(self.POOL_NAME).init_volume(template_vm,
  151. source_config)
  152. config = {
  153. 'name': 'root',
  154. 'pool': self.POOL_NAME,
  155. 'snap_on_start': True,
  156. 'rw': False,
  157. 'source': source,
  158. 'size': original_size,
  159. }
  160. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  161. self.assertEqual(volume.name, 'root')
  162. self.assertEqual(volume.pool, self.POOL_NAME)
  163. self.assertEqual(volume.size, original_size)
  164. self.assertTrue(volume.snap_on_start)
  165. self.assertTrue(volume.snap_on_start)
  166. self.assertFalse(volume.save_on_stop)
  167. self.assertFalse(volume.rw)
  168. self.assertEqual(volume.usage, 0)
  169. block = volume.block_device()
  170. assert isinstance(block, qubes.storage.BlockDevice)
  171. self.assertEqual(block.path,
  172. '{base}/{src}.img:{base}/{src}-cow.img:'
  173. '{base}/{dst}-cow.img'.format(
  174. base=self.POOL_DIR,
  175. src='vm-templates/' + template_vm.name + '/root',
  176. dst='appvms/' + vm.name + '/root',
  177. ))
  178. self.assertEqual(block.name, 'root')
  179. self.assertEqual(block.script, 'block-snapshot')
  180. self.assertEqual(block.rw, False)
  181. self.assertEqual(block.devtype, 'disk')
  182. def test_002_read_write_volume(self):
  183. config = {
  184. 'name': 'root',
  185. 'pool': self.POOL_NAME,
  186. 'rw': True,
  187. 'save_on_stop': True,
  188. 'size': defaults['root_img_size'],
  189. }
  190. vm = qubes.tests.storage.TestVM(self)
  191. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  192. self.assertEqual(volume.name, 'root')
  193. self.assertEqual(volume.pool, self.POOL_NAME)
  194. self.assertEqual(volume.size, defaults['root_img_size'])
  195. self.assertFalse(volume.snap_on_start)
  196. self.assertTrue(volume.save_on_stop)
  197. self.assertTrue(volume.rw)
  198. block = volume.block_device()
  199. self.assertEqual(block.name, 'root')
  200. self.assertEqual(block.path, '{base}.img:{base}-cow.img'.format(
  201. base=self.POOL_DIR + '/appvms/' + vm.name + '/root'))
  202. self.assertEqual(block.rw, True)
  203. self.assertEqual(block.script, 'block-origin')
  204. def test_003_read_only_volume(self):
  205. template = self.app.default_template
  206. vid = template.volumes['root'].vid
  207. config = {
  208. 'name': 'root',
  209. 'pool': self.POOL_NAME,
  210. 'rw': False,
  211. 'vid': vid,
  212. }
  213. vm = qubes.tests.storage.TestVM(self, template=template)
  214. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  215. self.assertEqual(volume.name, 'root')
  216. self.assertEqual(volume.pool, self.POOL_NAME)
  217. # original_size = qubes.config.defaults['root_img_size']
  218. # FIXME: self.assertEqual(volume.size, original_size)
  219. self.assertFalse(volume.snap_on_start)
  220. self.assertFalse(volume.save_on_stop)
  221. self.assertFalse(volume.rw)
  222. block = volume.block_device()
  223. self.assertEqual(block.rw, False)
  224. def test_004_volatile_volume(self):
  225. config = {
  226. 'name': 'root',
  227. 'pool': self.POOL_NAME,
  228. 'size': defaults['root_img_size'],
  229. 'rw': True,
  230. }
  231. vm = qubes.tests.storage.TestVM(self)
  232. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  233. self.assertEqual(volume.name, 'root')
  234. self.assertEqual(volume.pool, self.POOL_NAME)
  235. self.assertEqual(volume.size, defaults['root_img_size'])
  236. self.assertFalse(volume.snap_on_start)
  237. self.assertFalse(volume.save_on_stop)
  238. self.assertTrue(volume.rw)
  239. def test_005_appvm_volumes(self):
  240. ''' Check if AppVM volumes are propertly initialized '''
  241. vmname = self.make_vm_name('appvm')
  242. vm = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=vmname,
  243. template=self.app.default_template,
  244. label='red')
  245. template_dir = os.path.join(self.POOL_DIR, 'vm-templates',
  246. vm.template.name)
  247. vm_dir = os.path.join(self.POOL_DIR, 'appvms', vmname)
  248. expected = template_dir + '/root.img:' + \
  249. template_dir + '/root-cow.img:' + \
  250. vm_dir + '/root-cow.img'
  251. self.assertVolumePath(vm, 'root', expected, rw=True)
  252. expected = vm_dir + '/private.img:' + \
  253. vm_dir + '/private-cow.img'
  254. self.assertVolumePath(vm, 'private', expected, rw=True)
  255. expected = vm_dir + '/volatile.img'
  256. self.assertVolumePath(vm, 'volatile', expected, rw=True)
  257. def test_006_template_volumes(self):
  258. ''' Check if TemplateVM volumes are propertly initialized '''
  259. vmname = self.make_vm_name('appvm')
  260. vm = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM, name=vmname,
  261. label='red')
  262. vm_dir = os.path.join(self.POOL_DIR, 'vm-templates', vmname)
  263. expected = vm_dir + '/root.img:' + vm_dir + '/root-cow.img'
  264. self.assertVolumePath(vm, 'root', expected, rw=True)
  265. expected = vm_dir + '/private.img:' + \
  266. vm_dir + '/private-cow.img'
  267. self.assertVolumePath(vm, 'private', expected, rw=True)
  268. expected = vm_dir + '/volatile.img'
  269. self.assertVolumePath(vm, 'volatile', expected, rw=True)
  270. def test_010_revisions_to_keep_reject_invalid(self):
  271. ''' Check if TemplateVM volumes are propertly initialized '''
  272. config = {
  273. 'name': 'root',
  274. 'pool': self.POOL_NAME,
  275. 'save_on_stop': True,
  276. 'rw': True,
  277. 'size': defaults['root_img_size'],
  278. }
  279. vm = qubes.tests.storage.TestVM(self)
  280. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  281. self.assertEqual(volume.revisions_to_keep, 1)
  282. with self.assertRaises((NotImplementedError, ValueError)):
  283. volume.revisions_to_keep = 2
  284. self.assertEqual(volume.revisions_to_keep, 1)
  285. def test_020_import_data(self):
  286. config = {
  287. 'name': 'root',
  288. 'pool': self.POOL_NAME,
  289. 'save_on_stop': True,
  290. 'rw': True,
  291. 'size': 1024 * 1024,
  292. }
  293. vm = qubes.tests.storage.TestVM(self)
  294. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  295. volume.create()
  296. import_path = volume.import_data(volume.size)
  297. self.assertNotEqual(volume.path, import_path)
  298. with open(import_path, 'w+') as import_file:
  299. import_file.write('test')
  300. volume.import_data_end(True)
  301. self.assertFalse(os.path.exists(import_path), import_path)
  302. with open(volume.path) as volume_file:
  303. volume_data = volume_file.read().strip('\0')
  304. self.assertEqual(volume_data, 'test')
  305. def test_021_import_data_fail(self):
  306. config = {
  307. 'name': 'root',
  308. 'pool': self.POOL_NAME,
  309. 'save_on_stop': True,
  310. 'rw': True,
  311. 'size': 1024 * 1024,
  312. }
  313. vm = qubes.tests.storage.TestVM(self)
  314. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  315. volume.create()
  316. import_path = volume.import_data(volume.size)
  317. self.assertNotEqual(volume.path, import_path)
  318. with open(import_path, 'w+') as import_file:
  319. import_file.write('test')
  320. volume.import_data_end(False)
  321. self.assertFalse(os.path.exists(import_path), import_path)
  322. with open(volume.path) as volume_file:
  323. volume_data = volume_file.read().strip('\0')
  324. self.assertNotEqual(volume_data, 'test')
  325. def test_022_import_data_empty(self):
  326. config = {
  327. 'name': 'root',
  328. 'pool': self.POOL_NAME,
  329. 'save_on_stop': True,
  330. 'rw': True,
  331. 'size': 1024 * 1024,
  332. }
  333. vm = qubes.tests.storage.TestVM(self)
  334. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  335. volume.create()
  336. with open(volume.path, 'w') as vol_file:
  337. vol_file.write('test data')
  338. import_path = volume.import_data(volume.size)
  339. self.assertNotEqual(volume.path, import_path)
  340. with open(import_path, 'w+'):
  341. pass
  342. volume.import_data_end(True)
  343. self.assertFalse(os.path.exists(import_path), import_path)
  344. with open(volume.path) as volume_file:
  345. volume_data = volume_file.read().strip('\0')
  346. self.assertNotEqual(volume_data, 'test data')
  347. def test_023_resize(self):
  348. config = {
  349. 'name': 'root',
  350. 'pool': self.POOL_NAME,
  351. 'rw': True,
  352. 'save_on_stop': True,
  353. 'size': 32 * 1024**2,
  354. }
  355. vm = qubes.tests.storage.TestVM(self)
  356. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  357. qubes.utils.void_coros_maybe([volume.create()])
  358. new_size = 64 * 1024 ** 2
  359. qubes.utils.void_coros_maybe([volume.resize(new_size)])
  360. self.assertEqual(os.path.getsize(volume.path), new_size)
  361. self.assertEqual(volume.size, new_size)
  362. def test_024_import_data_with_new_size(self):
  363. config = {
  364. 'name': 'root',
  365. 'pool': self.POOL_NAME,
  366. 'save_on_stop': True,
  367. 'rw': True,
  368. 'size': 1024 * 1024,
  369. }
  370. new_size = 2 * 1024 * 1024
  371. vm = qubes.tests.storage.TestVM(self)
  372. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  373. volume.create()
  374. import_path = volume.import_data(new_size)
  375. self.assertNotEqual(volume.path, import_path)
  376. with open(import_path, 'r+b') as import_file:
  377. import_file.write(b'test')
  378. volume.import_data_end(True)
  379. self.assertFalse(os.path.exists(import_path), import_path)
  380. with open(volume.path, 'rb') as volume_file:
  381. volume_data = volume_file.read()
  382. self.assertEqual(volume_data.strip(b'\0'), b'test')
  383. self.assertEqual(len(volume_data), new_size)
  384. def _get_loop_size(self, path):
  385. sudo = [] if os.getuid() == 0 else ['sudo']
  386. try:
  387. loop_name = subprocess.check_output(
  388. sudo + ['losetup', '--associated', path]).decode().split(':')[0]
  389. if os.getuid() != 0:
  390. return int(
  391. subprocess.check_output(
  392. ['sudo', 'blockdev', '--getsize64', loop_name]))
  393. fd = os.open(loop_name, os.O_RDONLY)
  394. try:
  395. return os.lseek(fd, 0, os.SEEK_END)
  396. finally:
  397. os.close(fd)
  398. except subprocess.CalledProcessError:
  399. return None
  400. def _setup_loop(self, path):
  401. sudo = [] if os.getuid() == 0 else ['sudo']
  402. loop_name = subprocess.check_output(
  403. sudo + ['losetup', '--show', '--find', path]).decode().strip()
  404. self.addCleanup(subprocess.call, sudo + ['losetup', '-d', loop_name])
  405. def test_007_resize_running(self):
  406. old_size = 32 * 1024**2
  407. config = {
  408. 'name': 'root',
  409. 'pool': self.POOL_NAME,
  410. 'rw': True,
  411. 'save_on_stop': True,
  412. 'size': old_size,
  413. }
  414. vm = qubes.tests.storage.TestVM(self)
  415. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  416. qubes.utils.void_coros_maybe([volume.create()])
  417. qubes.utils.void_coros_maybe([volume.start()])
  418. self._setup_loop(volume.path)
  419. new_size = 64 * 1024 ** 2
  420. orig_check_call = subprocess.check_call
  421. with unittest.mock.patch('subprocess.check_call') as mock_subprocess:
  422. sudo = [] if os.getuid() == 0 else ['sudo']
  423. mock_subprocess.side_effect = (lambda *args, **kwargs:
  424. orig_check_call(sudo + args[0], *args[1:], **kwargs))
  425. qubes.utils.void_coros_maybe([volume.resize(new_size)])
  426. self.assertEqual(os.path.getsize(volume.path), new_size)
  427. self.assertEqual(self._get_loop_size(volume.path), new_size)
  428. self.assertEqual(volume.size, new_size)
  429. qubes.utils.void_coros_maybe([volume.stop()])
  430. self.assertEqual(os.path.getsize(volume.path), new_size)
  431. self.assertEqual(volume.size, new_size)
  432. def assertVolumePath(self, vm, dev_name, expected, rw=True):
  433. # :pylint: disable=invalid-name
  434. volumes = vm.volumes
  435. b_dev = volumes[dev_name].block_device()
  436. self.assertEqual(b_dev.rw, rw)
  437. self.assertEqual(b_dev.path, expected)
  438. class TC_03_FilePool(qubes.tests.QubesTestCase):
  439. """ Test the paths for the default file based pool (``FilePool``).
  440. """
  441. POOL_DIR = '/tmp/test-pool'
  442. APPVMS_DIR = '/tmp/test-pool/appvms'
  443. TEMPLATES_DIR = '/tmp/test-pool/vm-templates'
  444. SERVICE_DIR = '/tmp/test-pool/servicevms'
  445. POOL_NAME = 'test-pool'
  446. POOL_CONFIG = {'driver': 'file', 'dir_path': POOL_DIR, 'name': POOL_NAME}
  447. def setUp(self):
  448. """ Add a test file based storage pool """
  449. super(TC_03_FilePool, self).setUp()
  450. self.test_base_dir = '/tmp/qubes-test-dir'
  451. self.base_dir_patch = unittest.mock.patch.dict(qubes.config.system_path,
  452. {'qubes_base_dir': self.test_base_dir})
  453. self.base_dir_patch2 = unittest.mock.patch(
  454. 'qubes.config.qubes_base_dir', self.test_base_dir)
  455. self.base_dir_patch3 = unittest.mock.patch.dict(
  456. qubes.config.defaults['pool_configs']['varlibqubes'],
  457. {'dir_path': self.test_base_dir})
  458. self.base_dir_patch.start()
  459. self.base_dir_patch2.start()
  460. self.base_dir_patch3.start()
  461. self.app = TestApp()
  462. self.loop.run_until_complete(self.app.add_pool(**self.POOL_CONFIG))
  463. self.app.create_dummy_template()
  464. def tearDown(self):
  465. """ Remove the file based storage pool after testing """
  466. for vm in list(self.app.domains):
  467. if vm.name.startswith(qubes.tests.VMPREFIX):
  468. del self.app.domains[vm]
  469. self.app.default_template = None
  470. del self.app.domains['test-template']
  471. self.loop.run_until_complete(self.app.remove_pool("test-pool"))
  472. self.app.cleanup()
  473. self.app.close()
  474. del self.app
  475. self.base_dir_patch3.stop()
  476. self.base_dir_patch2.stop()
  477. self.base_dir_patch.stop()
  478. super(TC_03_FilePool, self).tearDown()
  479. shutil.rmtree(self.POOL_DIR, ignore_errors=True)
  480. if os.path.exists('/tmp/qubes-test-dir'):
  481. shutil.rmtree('/tmp/qubes-test-dir')
  482. def test_001_pool_exists(self):
  483. """ Check if the storage pool was added to the storage pool config """
  484. self.assertIn('test-pool', self.app.pools.keys())
  485. def test_002_pool_dir_create(self):
  486. """ Check if the storage pool dir and subdirs were created """
  487. # The dir should not exists before
  488. pool_name = 'foo'
  489. pool_dir = '/tmp/foo'
  490. appvms_dir = '/tmp/foo/appvms'
  491. templates_dir = '/tmp/foo/vm-templates'
  492. self.assertFalse(os.path.exists(pool_dir))
  493. self.loop.run_until_complete(
  494. self.app.add_pool(name=pool_name, dir_path=pool_dir, driver='file'))
  495. self.assertTrue(os.path.exists(pool_dir))
  496. self.assertTrue(os.path.exists(appvms_dir))
  497. self.assertTrue(os.path.exists(templates_dir))
  498. shutil.rmtree(pool_dir, ignore_errors=True)
  499. def test_003_size(self):
  500. pool = self.app.get_pool(self.POOL_NAME)
  501. with self.assertNotRaises(NotImplementedError):
  502. size = pool.size
  503. statvfs = os.statvfs(self.POOL_DIR)
  504. self.assertEqual(size, statvfs.f_blocks * statvfs.f_frsize)
  505. def test_004_usage(self):
  506. pool = self.app.get_pool(self.POOL_NAME)
  507. with self.assertNotRaises(NotImplementedError):
  508. usage = pool.usage
  509. statvfs = os.statvfs(self.POOL_DIR)
  510. self.assertEqual(usage,
  511. statvfs.f_frsize * (statvfs.f_blocks - statvfs.f_bfree))
  512. def test_005_revisions_to_keep(self):
  513. pool = self.app.get_pool(self.POOL_NAME)
  514. self.assertEqual(pool.revisions_to_keep, 1)
  515. with self.assertRaises((NotImplementedError, ValueError)):
  516. pool.revisions_to_keep = 2
  517. self.assertEqual(pool.revisions_to_keep, 1)
  518. def test_011_appvm_file_images(self):
  519. """ Check if all the needed image files are created for an AppVm"""
  520. vmname = self.make_vm_name('appvm')
  521. vm = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=vmname,
  522. template=self.app.default_template,
  523. volume_config={
  524. 'private': {
  525. 'pool': 'test-pool'
  526. },
  527. 'volatile': {
  528. 'pool': 'test-pool'
  529. }
  530. }, label='red')
  531. loop = asyncio.get_event_loop()
  532. loop.run_until_complete(vm.create_on_disk())
  533. expected_vmdir = os.path.join(self.APPVMS_DIR, vm.name)
  534. expected_private_path = os.path.join(expected_vmdir, 'private.img')
  535. self.assertEqual(vm.volumes['private'].path, expected_private_path)
  536. expected_volatile_path = os.path.join(expected_vmdir, 'volatile.img')
  537. vm.volumes['volatile'].reset()
  538. self.assertEqualAndExists(vm.volumes['volatile'].path,
  539. expected_volatile_path)
  540. def test_013_template_file_images(self):
  541. """ Check if root.img, private.img, volatile.img and root-cow.img are
  542. created propertly by the storage system
  543. """
  544. vmname = self.make_vm_name('tmvm')
  545. vm = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM, name=vmname,
  546. volume_config={
  547. 'root': {
  548. 'pool': 'test-pool'
  549. },
  550. 'private': {
  551. 'pool': 'test-pool'
  552. },
  553. 'volatile': {
  554. 'pool': 'test-pool'
  555. }
  556. }, label='red')
  557. loop = asyncio.get_event_loop()
  558. loop.run_until_complete(vm.create_on_disk())
  559. expected_vmdir = os.path.join(self.TEMPLATES_DIR, vm.name)
  560. expected_root_origin_path = os.path.join(expected_vmdir, 'root.img')
  561. expected_root_cow_path = os.path.join(expected_vmdir, 'root-cow.img')
  562. expected_root_path = '%s:%s' % (expected_root_origin_path,
  563. expected_root_cow_path)
  564. self.assertEqual(vm.volumes['root'].block_device().path,
  565. expected_root_path)
  566. self.assertExist(vm.volumes['root'].path)
  567. expected_private_path = os.path.join(expected_vmdir, 'private.img')
  568. self.assertEqualAndExists(vm.volumes['private'].path,
  569. expected_private_path)
  570. expected_rootcow_path = os.path.join(expected_vmdir, 'root-cow.img')
  571. self.assertEqual(vm.volumes['root'].path_cow, expected_rootcow_path)
  572. def assertEqualAndExists(self, result_path, expected_path):
  573. """ Check if the ``result_path``, matches ``expected_path`` and exists.
  574. See also: :meth:``assertExist``
  575. """
  576. # :pylint: disable=invalid-name
  577. self.assertEqual(result_path, expected_path)
  578. self.assertExist(result_path)
  579. def assertExist(self, path):
  580. """ Assert that the given path exists. """
  581. # :pylint: disable=invalid-name
  582. self.assertTrue(
  583. os.path.exists(path), "Path {!s} does not exist".format(path))