storage.py 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  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. import shutil
  20. import unittest.mock
  21. import qubes.log
  22. import qubes.storage
  23. from qubes.exc import QubesException
  24. from qubes.storage import pool_drivers
  25. from qubes.storage.file import FilePool
  26. from qubes.storage.reflink import ReflinkPool
  27. from qubes.tests import SystemTestCase, QubesTestCase
  28. # :pylint: disable=invalid-name
  29. class TestPool(unittest.mock.Mock):
  30. def __init__(self, *args, **kwargs):
  31. super(TestPool, self).__init__(*args, spec=qubes.storage.Pool, **kwargs)
  32. try:
  33. self.name = kwargs['name']
  34. except KeyError:
  35. pass
  36. def __str__(self):
  37. return 'test'
  38. def init_volume(self, vm, volume_config):
  39. vol = unittest.mock.Mock(spec=qubes.storage.Volume)
  40. vol.configure_mock(**volume_config)
  41. vol.pool = self
  42. vol.import_data.return_value = '/tmp/test-' + vm.name
  43. return vol
  44. class TestVM(object):
  45. def __init__(self, test, template=None):
  46. self.app = test.app
  47. self.name = test.make_vm_name('appvm')
  48. self.dir_path_prefix = 'appvms'
  49. self.dir_path = '/var/lib/qubes/appvms/' + self.name
  50. self.log = qubes.log.get_vm_logger(self.name)
  51. if template:
  52. self.template = template
  53. def is_template(self):
  54. # :pylint: disable=no-self-use
  55. return False
  56. def is_disposablevm(self):
  57. # :pylint: disable=no-self-use
  58. return False
  59. class TestTemplateVM(TestVM):
  60. dir_path_prefix = qubes.config.system_path['qubes_templates_dir']
  61. def __init__(self, test, template=None):
  62. super(TestTemplateVM, self).__init__(test, template)
  63. self.dir_path = '/var/lib/qubes/vm-templates/' + self.name
  64. def is_template(self):
  65. return True
  66. class TestDisposableVM(TestVM):
  67. def is_disposablevm(self):
  68. return True
  69. class TestApp(qubes.Qubes):
  70. def __init__(self, *args, **kwargs): # pylint: disable=unused-argument
  71. super(TestApp, self).__init__('/tmp/qubes-test.xml',
  72. load=False, offline_mode=True, **kwargs)
  73. self.load_initial_values()
  74. self.default_pool = self.pools['varlibqubes']
  75. class TC_00_Pool(QubesTestCase):
  76. """ This class tests the utility methods from :mod:``qubes.storage`` """
  77. def setUp(self):
  78. super(TC_00_Pool, self).setUp()
  79. self.basedir_patch = unittest.mock.patch('qubes.config.qubes_base_dir',
  80. '/tmp/qubes-test-basedir')
  81. self.basedir_patch.start()
  82. self.app = TestApp()
  83. def tearDown(self):
  84. self.basedir_patch.stop()
  85. self.app.close()
  86. del self.app
  87. shutil.rmtree('/tmp/qubes-test-basedir', ignore_errors=True)
  88. super().tearDown()
  89. def test_000_unknown_pool_driver(self):
  90. # :pylint: disable=protected-access
  91. """ Expect an exception when unknown pool is requested"""
  92. with self.assertRaises(QubesException):
  93. self.app.get_pool('foo-bar')
  94. def test_001_all_pool_drivers(self):
  95. """ Expect all our pool drivers (and only them) """
  96. self.assertCountEqual(
  97. ['linux-kernel', 'lvm_thin', 'file', 'file-reflink'],
  98. pool_drivers())
  99. def test_002_get_pool_klass(self):
  100. """ Expect the default pool to be `FilePool` or `ReflinkPool` """
  101. # :pylint: disable=protected-access
  102. result = self.app.get_pool('varlibqubes')
  103. self.assertTrue(isinstance(result, FilePool)
  104. or isinstance(result, ReflinkPool))
  105. def test_003_pool_exists_default(self):
  106. """ Expect the default pool to exists """
  107. self.assertPoolExists('varlibqubes')
  108. def test_004_add_remove_pool(self):
  109. """ Tries to adding and removing a pool. """
  110. pool_name = 'asdjhrp89132'
  111. # make sure it's really does not exist
  112. self.loop.run_until_complete(self.app.remove_pool(pool_name))
  113. self.assertFalse(self.assertPoolExists(pool_name))
  114. self.loop.run_until_complete(
  115. self.app.add_pool(name=pool_name,
  116. driver='file',
  117. dir_path='/tmp/asdjhrp89132'))
  118. self.assertTrue(self.assertPoolExists(pool_name))
  119. self.loop.run_until_complete(self.app.remove_pool(pool_name))
  120. self.assertFalse(self.assertPoolExists(pool_name))
  121. def assertPoolExists(self, pool):
  122. """ Check if specified pool exists """
  123. return pool in self.app.pools.keys()
  124. def test_005_remove_used(self):
  125. pool_name = 'test-pool-asdf'
  126. dir_path = '/tmp/{}'.format(pool_name)
  127. pool = self.loop.run_until_complete(
  128. self.app.add_pool(name=pool_name,
  129. driver='file',
  130. dir_path=dir_path))
  131. self.addCleanup(shutil.rmtree, dir_path)
  132. vm = self.app.add_new_vm('StandaloneVM', label='red',
  133. name=self.make_vm_name('vm'))
  134. self.loop.run_until_complete(vm.create_on_disk(pool=pool))
  135. with self.assertRaises(qubes.exc.QubesPoolInUseError):
  136. self.loop.run_until_complete(self.app.remove_pool(pool_name))