storage.py 4.4 KB

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