storage.py 4.4 KB

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