storage_lvm.py 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195
  1. #
  2. # The Qubes OS Project, http://www.qubes-os.org
  3. #
  4. # Copyright (C) 2016 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 lvm storage driver. By default tests are going to use the
  21. 'qubes_dom0/pool00'. An alternative LVM thin pool may be provided via
  22. :envvar:`DEFAULT_LVM_POOL` shell variable.
  23. Any pool variables prefixed with 'LVM_' or 'lvm_' represent a LVM
  24. 'volume_group/thin_pool' combination. Pool variables without a prefix
  25. represent a :py:class:`qubes.storage.lvm.ThinPool`.
  26. '''
  27. import os
  28. import unittest
  29. import qubes.tests
  30. from qubes.storage.lvm import ThinPool, ThinVolume
  31. if 'DEFAULT_LVM_POOL' in os.environ.keys():
  32. DEFAULT_LVM_POOL = os.environ['DEFAULT_LVM_POOL']
  33. else:
  34. DEFAULT_LVM_POOL = 'qubes_dom0/pool00'
  35. def lvm_pool_exists(volume_group, thin_pool):
  36. ''' Returns ``True`` if thin pool exists in the volume group. '''
  37. path = "/dev/mapper/{!s}-{!s}".format(volume_group, thin_pool)
  38. return os.path.exists(path)
  39. def skipUnlessLvmPoolExists(test_item): # pylint: disable=invalid-name
  40. ''' Decorator that skips LVM tests if the default pool is missing. '''
  41. volume_group, thin_pool = DEFAULT_LVM_POOL.split('/', 1)
  42. result = lvm_pool_exists(volume_group, thin_pool)
  43. msg = 'LVM thin pool {!r} does not exist'.format(DEFAULT_LVM_POOL)
  44. return unittest.skipUnless(result, msg)(test_item)
  45. POOL_CONF = {'name': 'test-lvm',
  46. 'driver': 'lvm_thin',
  47. 'volume_group': DEFAULT_LVM_POOL.split('/')[0],
  48. 'thin_pool': DEFAULT_LVM_POOL.split('/')[1]}
  49. class ThinPoolBase(qubes.tests.QubesTestCase):
  50. ''' Sanity tests for :py:class:`qubes.storage.lvm.ThinPool` '''
  51. created_pool = False
  52. def setUp(self):
  53. super(ThinPoolBase, self).setUp()
  54. volume_group, thin_pool = DEFAULT_LVM_POOL.split('/', 1)
  55. self.pool = self._find_pool(volume_group, thin_pool)
  56. if not self.pool:
  57. self.pool = self.app.add_pool(**POOL_CONF)
  58. self.created_pool = True
  59. def tearDown(self):
  60. ''' Remove the default lvm pool if it was created only for this test '''
  61. if self.created_pool:
  62. self.app.remove_pool(self.pool.name)
  63. super(ThinPoolBase, self).tearDown()
  64. def _find_pool(self, volume_group, thin_pool):
  65. ''' Returns the pool matching the specified ``volume_group`` &
  66. ``thin_pool``, or None.
  67. '''
  68. pools = [p for p in self.app.pools
  69. if issubclass(p.__class__, ThinPool)]
  70. for pool in pools:
  71. if pool.volume_group == volume_group \
  72. and pool.thin_pool == thin_pool:
  73. return pool
  74. return None
  75. @skipUnlessLvmPoolExists
  76. class TC_00_ThinPool(ThinPoolBase):
  77. ''' Sanity tests for :py:class:`qubes.storage.lvm.ThinPool` '''
  78. def setUp(self):
  79. xml_path = '/tmp/qubes-test.xml'
  80. self.app = qubes.Qubes.create_empty_store(store=xml_path,
  81. clockvm=None,
  82. updatevm=None,
  83. offline_mode=True,
  84. )
  85. os.environ['QUBES_XML_PATH'] = xml_path
  86. super(TC_00_ThinPool, self).setUp()
  87. def tearDown(self):
  88. super(TC_00_ThinPool, self).tearDown()
  89. os.unlink(self.app.store)
  90. del self.app
  91. for attr in dir(self):
  92. if isinstance(getattr(self, attr), qubes.vm.BaseVM):
  93. delattr(self, attr)
  94. def test_000_default_thin_pool(self):
  95. ''' Check whether :py:data`DEFAULT_LVM_POOL` exists. This pool is
  96. created by default, if at installation time LVM + Thin was chosen.
  97. '''
  98. msg = 'Thin pool {!r} does not exist'.format(DEFAULT_LVM_POOL)
  99. self.assertTrue(self.pool, msg)
  100. def test_001_origin_volume(self):
  101. ''' Test origin volume creation '''
  102. config = {
  103. 'name': 'root',
  104. 'pool': self.pool.name,
  105. 'save_on_stop': True,
  106. 'rw': True,
  107. 'size': qubes.config.defaults['root_img_size'],
  108. }
  109. vm = qubes.tests.storage.TestVM(self)
  110. volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
  111. self.assertIsInstance(volume, ThinVolume)
  112. self.assertEqual(volume.name, 'root')
  113. self.assertEqual(volume.pool, self.pool.name)
  114. self.assertEqual(volume.size, qubes.config.defaults['root_img_size'])
  115. volume.create()
  116. path = "/dev/%s" % volume.vid
  117. self.assertTrue(os.path.exists(path))
  118. volume.remove()
  119. def test_003_read_write_volume(self):
  120. ''' Test read-write volume creation '''
  121. config = {
  122. 'name': 'root',
  123. 'pool': self.pool.name,
  124. 'rw': True,
  125. 'save_on_stop': True,
  126. 'size': qubes.config.defaults['root_img_size'],
  127. }
  128. vm = qubes.tests.storage.TestVM(self)
  129. volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
  130. self.assertIsInstance(volume, ThinVolume)
  131. self.assertEqual(volume.name, 'root')
  132. self.assertEqual(volume.pool, self.pool.name)
  133. self.assertEqual(volume.size, qubes.config.defaults['root_img_size'])
  134. volume.create()
  135. path = "/dev/%s" % volume.vid
  136. self.assertTrue(os.path.exists(path))
  137. volume.remove()
  138. @skipUnlessLvmPoolExists
  139. class TC_01_ThinPool(ThinPoolBase, qubes.tests.SystemTestCase):
  140. ''' Sanity tests for :py:class:`qubes.storage.lvm.ThinPool` '''
  141. def setUp(self):
  142. super(TC_01_ThinPool, self).setUp()
  143. self.init_default_template()
  144. def test_004_import(self):
  145. template_vm = self.app.default_template
  146. name = self.make_vm_name('import')
  147. vm = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM, name=name,
  148. label='red')
  149. vm.clone_properties(template_vm)
  150. vm.clone_disk_files(template_vm, pool='test-lvm')
  151. for v_name, volume in vm.volumes.items():
  152. if volume.save_on_stop:
  153. expected = "/dev/{!s}/vm-{!s}-{!s}".format(
  154. DEFAULT_LVM_POOL.split('/')[0], vm.name, v_name)
  155. self.assertEqual(volume.path, expected)
  156. with self.assertNotRaises(qubes.exc.QubesException):
  157. vm.start()
  158. def test_005_create_appvm(self):
  159. vm = self.app.add_new_vm(cls=qubes.vm.appvm.AppVM,
  160. name=self.make_vm_name('appvm'), label='red')
  161. vm.create_on_disk(pool='test-lvm')
  162. for v_name, volume in vm.volumes.items():
  163. if volume.save_on_stop:
  164. expected = "/dev/{!s}/vm-{!s}-{!s}".format(
  165. DEFAULT_LVM_POOL.split('/')[0], vm.name, v_name)
  166. self.assertEqual(volume.path, expected)
  167. with self.assertNotRaises(qubes.exc.QubesException):
  168. vm.start()