Qubes/core-admin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tests/lvm & callback: Refactoring

Browse Source 
Mostly to avoid re-writing storage_lvm globals in storage_callback
tests.
This commit is contained in:
3hhh 2020-07-28 18:44:16 +02:00
parent 3db5e9f8bf
commit fd3a56e0cb
No known key found for this signature in database
GPG Key ID: EB03A691DB2F0833
2 changed files with 40 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
qubes/tests/storage_callback.py
View File

@ -29,13 +29,7 @@ import qubes.tests
import qubes.tests.storage import qubes.tests.storage
import qubes.tests.storage_lvm import qubes.tests.storage_lvm
from qubes.tests.storage_lvm import skipUnlessLvmPoolExists from qubes.tests.storage_lvm import skipUnlessLvmPoolExists
import qubes.storage.callback from qubes.storage.callback import CallbackPool, CallbackVolume
POOL_CLASS = qubes.storage.callback.CallbackPool
VOLUME_CLASS = qubes.storage.callback.CallbackVolume
POOL_CONF = {'name': 'test-callback',
'driver': 'callback',
'conf_id': 'invalid'}
CB_CONF = '/etc/qubes_callback.json' CB_CONF = '/etc/qubes_callback.json'
LOG_BIN = '/tmp/testCbLogArgs' LOG_BIN = '/tmp/testCbLogArgs'
@ -93,21 +87,15 @@ CB_DATA = {'utest-callback-01': {
class CallbackBase: class CallbackBase:
''' Mixin base class for callback tests. Has no base class. ''' ''' Mixin base class for callback tests. Has no base class. '''
bak_pool_class = None
bak_volume_class = None
bak_pool_conf = None
conf_id = None conf_id = None
pool_name = 'test-callback'
@classmethod @classmethod
def setUpClass(cls, conf_id='utest-callback-01'): def setUpClass(cls, conf_id='utest-callback-01'):
CallbackBase.bak_pool_class = qubes.tests.storage_lvm.POOL_CLASS conf = {'name': CallbackBase.pool_name,
CallbackBase.bak_volume_class = qubes.tests.storage_lvm.VOLUME_CLASS 'driver': 'callback',
CallbackBase.bak_pool_conf = qubes.tests.storage_lvm.POOL_CONF 'conf_id': conf_id}
qubes.tests.storage_lvm.POOL_CLASS = POOL_CLASS
qubes.tests.storage_lvm.VOLUME_CLASS = VOLUME_CLASS
cdict = {'conf_id': conf_id}
CallbackBase.conf_id = conf_id CallbackBase.conf_id = conf_id
qubes.tests.storage_lvm.POOL_CONF = {**POOL_CONF, **cdict}
assert not(os.path.exists(CB_CONF)), '%s must NOT exist. Please delete it, if you do not need it.' % CB_CONF assert not(os.path.exists(CB_CONF)), '%s must NOT exist. Please delete it, if you do not need it.' % CB_CONF
@ -116,17 +104,11 @@ class CallbackBase:
with open(CB_CONF, 'w') as outfile: with open(CB_CONF, 'w') as outfile:
json.dump(CB_DATA, outfile) json.dump(CB_DATA, outfile)
super().setUpClass() super().setUpClass(pool_class=CallbackPool, volume_class=CallbackVolume, pool_conf=conf)
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
super().tearDownClass() super().tearDownClass()
if CallbackBase.bak_pool_class:
qubes.tests.storage_lvm.POOL_CLASS = CallbackBase.bak_pool_class
if CallbackBase.bak_volume_class:
qubes.tests.storage_lvm.VOLUME_CLASS = CallbackBase.bak_volume_class
if CallbackBase.bak_pool_conf:
qubes.tests.storage_lvm.POOL_CONF = CallbackBase.bak_pool_conf
sudo = [] if os.getuid() == 0 else ['sudo'] sudo = [] if os.getuid() == 0 else ['sudo']
subprocess.run(sudo + ['rm', '-f', CB_CONF], check=True) subprocess.run(sudo + ['rm', '-f', CB_CONF], check=True)
@ -140,7 +122,7 @@ class CallbackBase:
def test_000_000_callback_test_init(self): def test_000_000_callback_test_init(self):
''' Check whether the test init did work. ''' ''' Check whether the test init did work. '''
if hasattr(self, 'pool'): if hasattr(self, 'pool'):
self.assertIsInstance(self.pool, qubes.storage.callback.CallbackPool) self.assertIsInstance(self.pool, CallbackPool)
self.assertEqual(self.pool.backend_class, qubes.storage.lvm.ThinPool) self.assertEqual(self.pool.backend_class, qubes.storage.lvm.ThinPool)
self.assertTrue(os.path.isfile(CB_CONF)) self.assertTrue(os.path.isfile(CB_CONF))
@ -218,7 +200,7 @@ exit 0
</qubes> </qubes>
""" """
xml = xml.replace('CONF_ID', CallbackBase.conf_id) xml = xml.replace('CONF_ID', CallbackBase.conf_id)
xml = xml.replace('POOL_NAME', POOL_CONF['name']) xml = xml.replace('POOL_NAME', CallbackBase.pool_name)
with open(LoggingCallbackBase.xml_path, 'w') as f: with open(LoggingCallbackBase.xml_path, 'w') as f:
f.write(xml) f.write(xml)
self.app = qubes.Qubes(LoggingCallbackBase.xml_path, self.app = qubes.Qubes(LoggingCallbackBase.xml_path,
@ -273,7 +255,7 @@ exit 0
''' create a lvm pool with additional callbacks ''' ''' create a lvm pool with additional callbacks '''
config = { config = {
'name': LoggingCallbackBase.volume_name, 'name': LoggingCallbackBase.volume_name,
'pool': POOL_CONF['name'], 'pool': CallbackBase.pool_name,
'save_on_stop': True, 'save_on_stop': True,
'rw': True, 'rw': True,
'revisions_to_keep': 2, 'revisions_to_keep': 2,
@ -285,7 +267,7 @@ exit 0
self.assertLog(test_name, 0) self.assertLog(test_name, 0)
self.init_pool() self.init_pool()
self.assertFalse(self.created_pool) self.assertFalse(self.created_pool)
self.assertIsInstance(self.pool, qubes.storage.callback.CallbackPool) self.assertIsInstance(self.pool, CallbackPool)
self.assertLog(test_name, 1) self.assertLog(test_name, 1)
vm = qubes.tests.storage.TestVM(self) vm = qubes.tests.storage.TestVM(self)
volume = self.app.get_pool(self.pool.name).init_volume(vm, config) volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
@ -309,7 +291,7 @@ class TC_91_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBa
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
conf_id = 'utest-callback-02' conf_id = 'utest-callback-02'
name = POOL_CONF['name'] name = CallbackBase.pool_name
bdriver = (CB_DATA[conf_id])['bdriver'] bdriver = (CB_DATA[conf_id])['bdriver']
ctor_params = json.dumps(CB_DATA[conf_id], sort_keys=True, indent=2) ctor_params = json.dumps(CB_DATA[conf_id], sort_keys=True, indent=2)
vname = LoggingCallbackBase.volume_name vname = LoggingCallbackBase.volume_name
@ -353,7 +335,7 @@ class TC_92_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBa
''' Make sure that we check the exit code of executed callbacks. ''' ''' Make sure that we check the exit code of executed callbacks. '''
config = { config = {
'name': LoggingCallbackBase.volume_name, 'name': LoggingCallbackBase.volume_name,
'pool': POOL_CONF['name'], 'pool': CallbackBase.pool_name,
'save_on_stop': True, 'save_on_stop': True,
'rw': True, 'rw': True,
'revisions_to_keep': 2, 'revisions_to_keep': 2,
@ -371,26 +353,26 @@ class TC_92_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBa
''' Make sure we error out on common user & dev mistakes. ''' ''' Make sure we error out on common user & dev mistakes. '''
#missing conf_id #missing conf_id
with self.assertRaises(qubes.storage.StoragePoolException): with self.assertRaises(qubes.storage.StoragePoolException):
cb = qubes.storage.callback.CallbackPool(name='some-name', conf_id='') cb = CallbackPool(name='some-name', conf_id='')
#invalid conf_id #invalid conf_id
with self.assertRaises(qubes.storage.StoragePoolException): with self.assertRaises(qubes.storage.StoragePoolException):
cb = qubes.storage.callback.CallbackPool(name='some-name', conf_id='nonexisting-id') cb = CallbackPool(name='some-name', conf_id='nonexisting-id')
#incorrect backend driver #incorrect backend driver
with self.assertRaises(qubes.storage.StoragePoolException): with self.assertRaises(qubes.storage.StoragePoolException):
cb = qubes.storage.callback.CallbackPool(name='some-name', conf_id='testing-fail-incorrect-bdriver') cb = CallbackPool(name='some-name', conf_id='testing-fail-incorrect-bdriver')
#missing config entries #missing config entries
with self.assertRaises(qubes.storage.StoragePoolException): with self.assertRaises(qubes.storage.StoragePoolException):
cb = qubes.storage.callback.CallbackPool(name='some-name', conf_id='testing-fail-missing-all') cb = CallbackPool(name='some-name', conf_id='testing-fail-missing-all')
#missing bdriver args #missing bdriver args
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
cb = qubes.storage.callback.CallbackPool(name='some-name', conf_id='testing-fail-missing-bdriver-args') cb = CallbackPool(name='some-name', conf_id='testing-fail-missing-bdriver-args')
class TC_93_CallbackPool(qubes.tests.QubesTestCase): class TC_93_CallbackPool(qubes.tests.QubesTestCase):
def test_001_missing_conf(self): def test_001_missing_conf(self):
''' A missing config file must cause errors. ''' ''' A missing config file must cause errors. '''
with self.assertRaises(FileNotFoundError): with self.assertRaises(FileNotFoundError):
cb = qubes.storage.callback.CallbackPool(name='some-name', conf_id='nonexisting-id') cb = CallbackPool(name='some-name', conf_id='nonexisting-id')

34
qubes/tests/storage_lvm.py
View File

@ -58,19 +58,27 @@ def skipUnlessLvmPoolExists(test_item): # pylint: disable=invalid-name
msg = 'LVM thin pool {!r} does not exist'.format(DEFAULT_LVM_POOL) msg = 'LVM thin pool {!r} does not exist'.format(DEFAULT_LVM_POOL)
return unittest.skipUnless(result, msg)(test_item) return unittest.skipUnless(result, msg)(test_item)
class ThinPoolBase(qubes.tests.QubesTestCase):
''' Sanity tests for :py:class:`qubes.storage.lvm.ThinPool` '''
POOL_CLASS = ThinPool pool_class = None
VOLUME_CLASS = ThinVolume volume_class = None
POOL_CONF = {'name': 'test-lvm', pool_conf = None
created_pool = False
@classmethod
def setUpClass(cls, pool_class=ThinPool, volume_class=ThinVolume, pool_conf=None):
''' Other test classes (e.g. callback) may use this to test their own config. '''
conf = pool_conf
if not conf:
conf = {'name': 'test-lvm',
'driver': 'lvm_thin', 'driver': 'lvm_thin',
'volume_group': DEFAULT_LVM_POOL.split('/')[0], 'volume_group': DEFAULT_LVM_POOL.split('/')[0],
'thin_pool': DEFAULT_LVM_POOL.split('/')[1]} 'thin_pool': DEFAULT_LVM_POOL.split('/')[1]}
ThinPoolBase.pool_class = pool_class
class ThinPoolBase(qubes.tests.QubesTestCase): ThinPoolBase.volume_class = volume_class
''' Sanity tests for :py:class:`qubes.storage.lvm.ThinPool` ''' ThinPoolBase.pool_conf = conf
created_pool = False
def setUp(self, init_pool=True): def setUp(self, init_pool=True):
super(ThinPoolBase, self).setUp() super(ThinPoolBase, self).setUp()
@ -105,7 +113,7 @@ class ThinPoolBase(qubes.tests.QubesTestCase):
self.pool = self._find_pool(volume_group, thin_pool) self.pool = self._find_pool(volume_group, thin_pool)
if not self.pool: if not self.pool:
self.pool = self.loop.run_until_complete( self.pool = self.loop.run_until_complete(
self.app.add_pool(**POOL_CONF)) self.app.add_pool(**ThinPoolBase.pool_conf))
self.created_pool = True self.created_pool = True
def _find_pool(self, volume_group, thin_pool): def _find_pool(self, volume_group, thin_pool):
@ -113,7 +121,7 @@ class ThinPoolBase(qubes.tests.QubesTestCase):
``thin_pool``, or None. ``thin_pool``, or None.
''' '''
pools = [p for p in self.app.pools.values() pools = [p for p in self.app.pools.values()
if issubclass(p.__class__, POOL_CLASS)] if issubclass(p.__class__, ThinPoolBase.pool_class)]
for pool in pools: for pool in pools:
if pool.volume_group == volume_group \ if pool.volume_group == volume_group \
and pool.thin_pool == thin_pool: and pool.thin_pool == thin_pool:
@ -161,7 +169,7 @@ class TC_00_ThinPool(ThinPoolBase):
} }
vm = qubes.tests.storage.TestVM(self) vm = qubes.tests.storage.TestVM(self)
volume = self.app.get_pool(self.pool.name).init_volume(vm, config) volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
self.assertIsInstance(volume, VOLUME_CLASS) self.assertIsInstance(volume, ThinPoolBase.volume_class)
self.assertEqual(volume.name, 'root') self.assertEqual(volume.name, 'root')
self.assertEqual(volume.pool, self.pool.name) self.assertEqual(volume.pool, self.pool.name)
self.assertEqual(volume.size, qubes.config.defaults['root_img_size']) self.assertEqual(volume.size, qubes.config.defaults['root_img_size'])
@ -181,7 +189,7 @@ class TC_00_ThinPool(ThinPoolBase):
} }
vm = qubes.tests.storage.TestVM(self) vm = qubes.tests.storage.TestVM(self)
volume = self.app.get_pool(self.pool.name).init_volume(vm, config) volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
self.assertIsInstance(volume, VOLUME_CLASS) self.assertIsInstance(volume, ThinPoolBase.volume_class)
self.assertEqual(volume.name, 'root') self.assertEqual(volume.name, 'root')
self.assertEqual(volume.pool, self.pool.name) self.assertEqual(volume.pool, self.pool.name)
self.assertEqual(volume.size, qubes.config.defaults['root_img_size']) self.assertEqual(volume.size, qubes.config.defaults['root_img_size'])
@ -958,7 +966,7 @@ class TC_00_ThinPool(ThinPoolBase):
} }
volume = self.app.get_pool(self.pool.name).init_volume( volume = self.app.get_pool(self.pool.name).init_volume(
vm, config_snapshot) vm, config_snapshot)
self.assertIsInstance(volume, VOLUME_CLASS) self.assertIsInstance(volume, ThinPoolBase.volume_class)
self.assertEqual(volume.name, 'root2') self.assertEqual(volume.name, 'root2')
self.assertEqual(volume.pool, self.pool.name) self.assertEqual(volume.pool, self.pool.name)
self.assertEqual(volume.size, qubes.config.defaults['root_img_size']) self.assertEqual(volume.size, qubes.config.defaults['root_img_size'])