storage.py 4.0 KB

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