storage_file.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381
  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 program is free software; you can redistribute it and/or modify
  7. # it under the terms of the GNU General Public License as published by
  8. # the Free Software Foundation; either version 2 of the License, or
  9. # (at your option) any later version.
  10. #
  11. # This program 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
  14. # GNU General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU General Public License along
  17. # with this program; if not, write to the Free Software Foundation, Inc.,
  18. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  19. #
  20. ''' Tests for the file storage backend '''
  21. import os
  22. import shutil
  23. import asyncio
  24. import qubes.storage
  25. import qubes.tests.storage
  26. from qubes.config import defaults
  27. # :pylint: disable=invalid-name
  28. class TestApp(qubes.Qubes):
  29. ''' A Mock App object '''
  30. def __init__(self, *args, **kwargs): # pylint: disable=unused-argument
  31. super(TestApp, self).__init__('/tmp/qubes-test.xml', load=False,
  32. offline_mode=True, **kwargs)
  33. self.load_initial_values()
  34. self.pools['linux-kernel'].dir_path = '/tmp/qubes-test-kernel'
  35. dummy_kernel = os.path.join(self.pools['linux-kernel'].dir_path,
  36. 'dummy')
  37. os.makedirs(dummy_kernel)
  38. open(os.path.join(dummy_kernel, 'vmlinuz'), 'w').close()
  39. open(os.path.join(dummy_kernel, 'modules.img'), 'w').close()
  40. open(os.path.join(dummy_kernel, 'initramfs'), 'w').close()
  41. self.default_kernel = 'dummy'
  42. def cleanup(self):
  43. ''' Remove temporary directories '''
  44. shutil.rmtree(self.pools['linux-kernel'].dir_path)
  45. def create_dummy_template(self):
  46. ''' Initalizes a dummy TemplateVM as the `default_template` '''
  47. template = self.add_new_vm(qubes.vm.templatevm.TemplateVM,
  48. name='test-template', label='red',
  49. memory=1024, maxmem=1024)
  50. self.default_template = template
  51. class TC_00_FilePool(qubes.tests.QubesTestCase):
  52. """ This class tests some properties of the 'default' pool.
  53. This test might become obsolete if we change the driver for the default
  54. pool to something else as 'file'.
  55. """
  56. def setUp(self):
  57. super(TC_00_FilePool, self).setUp()
  58. self.app = TestApp()
  59. def tearDown(self):
  60. self.app.cleanup()
  61. super(TC_00_FilePool, self).tearDown()
  62. def test000_default_pool_dir(self):
  63. """ The predefined dir for the default pool should be ``/var/lib/qubes``
  64. .. sealso::
  65. Data :data:``qubes.qubes.defaults['pool_config']``.
  66. """
  67. result = self.app.get_pool("default").dir_path
  68. expected = '/var/lib/qubes'
  69. self.assertEquals(result, expected)
  70. def test001_default_storage_class(self):
  71. """ Check when using default pool the Storage is
  72. ``qubes.storage.Storage``. """
  73. result = self._init_app_vm().storage
  74. self.assertIsInstance(result, qubes.storage.Storage)
  75. def _init_app_vm(self):
  76. """ Return initalised, but not created, AppVm. """
  77. vmname = self.make_vm_name('appvm')
  78. self.app.create_dummy_template()
  79. return self.app.add_new_vm(qubes.vm.appvm.AppVM, name=vmname,
  80. template=self.app.default_template,
  81. label='red')
  82. class TC_01_FileVolumes(qubes.tests.QubesTestCase):
  83. ''' Test correct handling of different types of volumes '''
  84. POOL_DIR = '/tmp/test-pool'
  85. POOL_NAME = 'test-pool'
  86. POOL_CONF = {'driver': 'file', 'dir_path': POOL_DIR, 'name': POOL_NAME}
  87. def setUp(self):
  88. """ Add a test file based storage pool """
  89. super(TC_01_FileVolumes, self).setUp()
  90. self.app = TestApp()
  91. self.app.create_dummy_template()
  92. self.app.add_pool(**self.POOL_CONF)
  93. def tearDown(self):
  94. """ Remove the file based storage pool after testing """
  95. self.app.remove_pool("test-pool")
  96. self.app.cleanup()
  97. super(TC_01_FileVolumes, self).tearDown()
  98. shutil.rmtree(self.POOL_DIR, ignore_errors=True)
  99. def test_000_origin_volume(self):
  100. config = {
  101. 'name': 'root',
  102. 'pool': self.POOL_NAME,
  103. 'save_on_stop': True,
  104. 'rw': True,
  105. 'size': defaults['root_img_size'],
  106. }
  107. vm = qubes.tests.storage.TestVM(self)
  108. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  109. self.assertEqual(volume.name, 'root')
  110. self.assertEqual(volume.pool, self.POOL_NAME)
  111. self.assertEqual(volume.size, defaults['root_img_size'])
  112. self.assertFalse(volume.snap_on_start)
  113. self.assertTrue(volume.save_on_stop)
  114. self.assertTrue(volume.rw)
  115. def test_001_snapshot_volume(self):
  116. source = 'vm-templates/fedora-23/root'
  117. original_size = qubes.config.defaults['root_img_size']
  118. config = {
  119. 'name': 'root',
  120. 'pool': 'default',
  121. 'snap_on_start': True,
  122. 'rw': False,
  123. 'source': source,
  124. 'size': original_size,
  125. }
  126. template_vm = self.app.default_template
  127. vm = qubes.tests.storage.TestVM(self, template=template_vm)
  128. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  129. self.assertEqual(volume.name, 'root')
  130. self.assertEqual(volume.pool, 'default')
  131. self.assertEqual(volume.size, original_size)
  132. self.assertTrue(volume.snap_on_start)
  133. self.assertTrue(volume.snap_on_start)
  134. self.assertFalse(volume.save_on_stop)
  135. self.assertFalse(volume.rw)
  136. self.assertEqual(volume.usage, 0)
  137. def test_002_read_write_volume(self):
  138. config = {
  139. 'name': 'root',
  140. 'pool': self.POOL_NAME,
  141. 'rw': True,
  142. 'save_on_stop': True,
  143. 'size': defaults['root_img_size'],
  144. }
  145. vm = qubes.tests.storage.TestVM(self)
  146. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  147. self.assertEqual(volume.name, 'root')
  148. self.assertEqual(volume.pool, self.POOL_NAME)
  149. self.assertEqual(volume.size, defaults['root_img_size'])
  150. self.assertFalse(volume.snap_on_start)
  151. self.assertTrue(volume.save_on_stop)
  152. self.assertTrue(volume.rw)
  153. def test_003_read_only_volume(self):
  154. template = self.app.default_template
  155. vid = template.volumes['root'].vid
  156. config = {'name': 'root', 'pool': 'default', 'rw': False, 'vid': vid}
  157. vm = qubes.tests.storage.TestVM(self, template=template)
  158. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  159. self.assertEqual(volume.name, 'root')
  160. self.assertEqual(volume.pool, 'default')
  161. # original_size = qubes.config.defaults['root_img_size']
  162. # FIXME: self.assertEqual(volume.size, original_size)
  163. self.assertFalse(volume.snap_on_start)
  164. self.assertFalse(volume.save_on_stop)
  165. self.assertFalse(volume.rw)
  166. def test_004_volatile_volume(self):
  167. config = {
  168. 'name': 'root',
  169. 'pool': self.POOL_NAME,
  170. 'size': defaults['root_img_size'],
  171. 'rw': True,
  172. }
  173. vm = qubes.tests.storage.TestVM(self)
  174. volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
  175. self.assertEqual(volume.name, 'root')
  176. self.assertEqual(volume.pool, self.POOL_NAME)
  177. self.assertEqual(volume.size, defaults['root_img_size'])
  178. self.assertFalse(volume.snap_on_start)
  179. self.assertFalse(volume.save_on_stop)
  180. self.assertTrue(volume.rw)
  181. def test_005_appvm_volumes(self):
  182. ''' Check if AppVM volumes are propertly initialized '''
  183. vmname = self.make_vm_name('appvm')
  184. vm = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=vmname,
  185. template=self.app.default_template,
  186. label='red')
  187. expected = vm.template.dir_path + '/root.img:' + vm.template.dir_path \
  188. + '/root-cow.img:' + vm.dir_path + '/root-cow.img'
  189. self.assertVolumePath(vm, 'root', expected, rw=False)
  190. expected = vm.dir_path + '/private.img:' + \
  191. vm.dir_path + '/private-cow.img'
  192. self.assertVolumePath(vm, 'private', expected, rw=True)
  193. expected = vm.dir_path + '/volatile.img'
  194. self.assertVolumePath(vm, 'volatile', expected, rw=True)
  195. def test_006_template_volumes(self):
  196. ''' Check if TemplateVM volumes are propertly initialized '''
  197. vmname = self.make_vm_name('appvm')
  198. vm = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM, name=vmname,
  199. label='red')
  200. expected = vm.dir_path + '/root.img:' + vm.dir_path + '/root-cow.img'
  201. self.assertVolumePath(vm, 'root', expected, rw=True)
  202. expected = vm.dir_path + '/private.img'
  203. self.assertVolumePath(vm, 'private', expected, rw=True)
  204. expected = vm.dir_path + '/volatile.img'
  205. self.assertVolumePath(vm, 'volatile', expected, rw=True)
  206. def assertVolumePath(self, vm, dev_name, expected, rw=True):
  207. # :pylint: disable=invalid-name
  208. volumes = vm.volumes
  209. b_dev = volumes[dev_name].block_device()
  210. self.assertEqual(b_dev.rw, rw)
  211. self.assertEquals(b_dev.path, expected)
  212. class TC_03_FilePool(qubes.tests.QubesTestCase):
  213. """ Test the paths for the default file based pool (``FilePool``).
  214. """
  215. POOL_DIR = '/tmp/test-pool'
  216. APPVMS_DIR = '/tmp/test-pool/appvms'
  217. TEMPLATES_DIR = '/tmp/test-pool/vm-templates'
  218. SERVICE_DIR = '/tmp/test-pool/servicevms'
  219. POOL_NAME = 'test-pool'
  220. POOL_CONFIG = {'driver': 'file', 'dir_path': POOL_DIR, 'name': POOL_NAME}
  221. def setUp(self):
  222. """ Add a test file based storage pool """
  223. super(TC_03_FilePool, self).setUp()
  224. self._orig_qubes_base_dir = qubes.config.system_path['qubes_base_dir']
  225. qubes.config.system_path['qubes_base_dir'] = '/tmp/qubes-test'
  226. self.app = TestApp()
  227. self.app.create_dummy_template()
  228. self.app.add_pool(**self.POOL_CONFIG)
  229. def tearDown(self):
  230. """ Remove the file based storage pool after testing """
  231. self.app.remove_pool("test-pool")
  232. self.app.cleanup()
  233. super(TC_03_FilePool, self).tearDown()
  234. shutil.rmtree(self.POOL_DIR, ignore_errors=True)
  235. if os.path.exists('/tmp/qubes-test'):
  236. shutil.rmtree('/tmp/qubes-test')
  237. qubes.config.system_path['qubes_base_dir'] = self._orig_qubes_base_dir
  238. def test_001_pool_exists(self):
  239. """ Check if the storage pool was added to the storage pool config """
  240. self.assertIn('test-pool', self.app.pools.keys())
  241. def test_002_pool_dir_create(self):
  242. """ Check if the storage pool dir and subdirs were created """
  243. # The dir should not exists before
  244. pool_name = 'foo'
  245. pool_dir = '/tmp/foo'
  246. appvms_dir = '/tmp/foo/appvms'
  247. templates_dir = '/tmp/foo/vm-templates'
  248. self.assertFalse(os.path.exists(pool_dir))
  249. self.app.add_pool(name=pool_name, dir_path=pool_dir, driver='file')
  250. self.assertTrue(os.path.exists(pool_dir))
  251. self.assertTrue(os.path.exists(appvms_dir))
  252. self.assertTrue(os.path.exists(templates_dir))
  253. shutil.rmtree(pool_dir, ignore_errors=True)
  254. def test_011_appvm_file_images(self):
  255. """ Check if all the needed image files are created for an AppVm"""
  256. vmname = self.make_vm_name('appvm')
  257. vm = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=vmname,
  258. template=self.app.default_template,
  259. volume_config={
  260. 'private': {
  261. 'pool': 'test-pool'
  262. },
  263. 'volatile': {
  264. 'pool': 'test-pool'
  265. }
  266. }, label='red')
  267. loop = asyncio.get_event_loop()
  268. loop.run_until_complete(vm.create_on_disk())
  269. expected_vmdir = os.path.join(self.APPVMS_DIR, vm.name)
  270. expected_private_path = os.path.join(expected_vmdir, 'private.img')
  271. self.assertEquals(vm.volumes['private'].path, expected_private_path)
  272. expected_volatile_path = os.path.join(expected_vmdir, 'volatile.img')
  273. vm.storage.get_pool(vm.volumes['volatile'])\
  274. .reset(vm.volumes['volatile'])
  275. self.assertEqualsAndExists(vm.volumes['volatile'].path,
  276. expected_volatile_path)
  277. def test_013_template_file_images(self):
  278. """ Check if root.img, private.img, volatile.img and root-cow.img are
  279. created propertly by the storage system
  280. """
  281. vmname = self.make_vm_name('tmvm')
  282. vm = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM, name=vmname,
  283. volume_config={
  284. 'root': {
  285. 'pool': 'test-pool'
  286. },
  287. 'private': {
  288. 'pool': 'test-pool'
  289. },
  290. 'volatile': {
  291. 'pool': 'test-pool'
  292. }
  293. }, label='red')
  294. loop = asyncio.get_event_loop()
  295. loop.run_until_complete(vm.create_on_disk())
  296. expected_vmdir = os.path.join(self.TEMPLATES_DIR, vm.name)
  297. expected_root_origin_path = os.path.join(expected_vmdir, 'root.img')
  298. expected_root_cow_path = os.path.join(expected_vmdir, 'root-cow.img')
  299. expected_root_path = '%s:%s' % (expected_root_origin_path,
  300. expected_root_cow_path)
  301. self.assertEquals(vm.volumes['root'].block_device().path,
  302. expected_root_path)
  303. self.assertExist(vm.volumes['root'].path)
  304. expected_private_path = os.path.join(expected_vmdir, 'private.img')
  305. self.assertEqualsAndExists(vm.volumes['private'].path,
  306. expected_private_path)
  307. expected_rootcow_path = os.path.join(expected_vmdir, 'root-cow.img')
  308. self.assertEqualsAndExists(vm.volumes['root'].path_cow,
  309. expected_rootcow_path)
  310. def assertEqualsAndExists(self, result_path, expected_path):
  311. """ Check if the ``result_path``, matches ``expected_path`` and exists.
  312. See also: :meth:``assertExist``
  313. """
  314. # :pylint: disable=invalid-name
  315. self.assertEquals(result_path, expected_path)
  316. self.assertExist(result_path)
  317. def assertExist(self, path):
  318. """ Assert that the given path exists. """
  319. # :pylint: disable=invalid-name
  320. self.assertTrue(
  321. os.path.exists(path), "Path {!s} does not exist".format(path))