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

tests/lvm & callback: remove explicit class references

Browse Source
This commit is contained in:
3hhh 2020-08-05 16:56:02 +02:00
parent b95339ea27
commit 9e26bfd432
No known key found for this signature in database
GPG Key ID: EB03A691DB2F0833
2 changed files with 30 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
qubes/tests/storage_callback.py
View File

@ -92,10 +92,10 @@ class CallbackBase:
@classmethod @classmethod
def setUpClass(cls, conf_id='utest-callback-01'): def setUpClass(cls, conf_id='utest-callback-01'):
conf = {'name': CallbackBase.pool_name, conf = {'name': cls.pool_name,
'driver': 'callback', 'driver': 'callback',
'conf_id': conf_id} 'conf_id': conf_id}
CallbackBase.conf_id = conf_id cls.conf_id = conf_id
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
@ -155,12 +155,12 @@ for arg in "$@" ; do
done done
exit 0 exit 0
""" """
script = script.replace('LOG_OUT', LoggingCallbackBase.test_log) script = script.replace('LOG_OUT', cls.test_log)
with open(LOG_BIN, 'w') as f: with open(LOG_BIN, 'w') as f:
f.write(script) f.write(script)
os.chmod(LOG_BIN, 0o775) os.chmod(LOG_BIN, 0o775)
LoggingCallbackBase.test_log_expected = log_expected cls.test_log_expected = log_expected
super().setUpClass(conf_id=conf_id) super().setUpClass(conf_id=conf_id)
@classmethod @classmethod
@ -169,7 +169,7 @@ exit 0
os.remove(LOG_BIN) os.remove(LOG_BIN)
def setUp(self, init_pool=False): def setUp(self, init_pool=False):
assert not(os.path.exists(LoggingCallbackBase.test_log)), '%s must NOT exist. Please delete it, if you do not need it.' % LoggingCallbackBase.test_log assert not(os.path.exists(self.test_log)), '%s must NOT exist. Please delete it, if you do not need it.' % self.test_log
self.maxDiff = None self.maxDiff = None
xml = """ xml = """
@ -199,16 +199,16 @@ exit 0
</domains> </domains>
</qubes> </qubes>
""" """
xml = xml.replace('CONF_ID', CallbackBase.conf_id) xml = xml.replace('CONF_ID', self.conf_id)
xml = xml.replace('POOL_NAME', CallbackBase.pool_name) xml = xml.replace('POOL_NAME', self.pool_name)
with open(LoggingCallbackBase.xml_path, 'w') as f: with open(self.xml_path, 'w') as f:
f.write(xml) f.write(xml)
self.app = qubes.Qubes(LoggingCallbackBase.xml_path, self.app = qubes.Qubes(self.xml_path,
clockvm=None, clockvm=None,
updatevm=None, updatevm=None,
offline_mode=True, offline_mode=True,
) )
os.environ['QUBES_XML_PATH'] = LoggingCallbackBase.xml_path os.environ['QUBES_XML_PATH'] = self.xml_path
super().setUp(init_pool=init_pool) super().setUp(init_pool=init_pool)
def tearDown(self): def tearDown(self):
@ -220,18 +220,18 @@ exit 0
if isinstance(getattr(self, attr), qubes.vm.BaseVM): if isinstance(getattr(self, attr), qubes.vm.BaseVM):
delattr(self, attr) delattr(self, attr)
if os.path.exists(LoggingCallbackBase.test_log): if os.path.exists(self.test_log):
os.remove(LoggingCallbackBase.test_log) os.remove(self.test_log)
if os.path.exists(LoggingCallbackBase.xml_path): if os.path.exists(self.xml_path):
os.remove(LoggingCallbackBase.xml_path) os.remove(self.xml_path)
def assertLogContent(self, expected): def assertLogContent(self, expected):
''' Assert that the log matches the given string. ''' Assert that the log matches the given string.
:param expected: Expected content of the log file (String). :param expected: Expected content of the log file (String).
''' '''
try: try:
with open(LoggingCallbackBase.test_log, 'r') as f: with open(self.test_log, 'r') as f:
found = f.read() found = f.read()
except FileNotFoundError: except FileNotFoundError:
found = '' found = ''
@ -244,7 +244,7 @@ exit 0
:param test_name: Name of the test. :param test_name: Name of the test.
:param ind: Index inside `test_log_expected` to check against (Integer starting at 0). :param ind: Index inside `test_log_expected` to check against (Integer starting at 0).
''' '''
d = LoggingCallbackBase.test_log_expected[str(self.__class__) + test_name] d = self.test_log_expected[str(self.__class__) + test_name]
expected = [] expected = []
for i in range(ind+1): for i in range(ind+1):
expected = expected + [d[i]] expected = expected + [d[i]]
@ -254,8 +254,8 @@ exit 0
def test_001_callbacks(self): def test_001_callbacks(self):
''' create a lvm pool with additional callbacks ''' ''' create a lvm pool with additional callbacks '''
config = { config = {
'name': LoggingCallbackBase.volume_name, 'name': self.volume_name,
'pool': CallbackBase.pool_name, 'pool': self.pool_name,
'save_on_stop': True, 'save_on_stop': True,
'rw': True, 'rw': True,
'revisions_to_keep': 2, 'revisions_to_keep': 2,
@ -291,10 +291,10 @@ 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 = CallbackBase.pool_name name = cls.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 = cls.volume_name
vid = '{0}/vm-test-inst-appvm-{1}'.format(qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0], vname) vid = '{0}/vm-test-inst-appvm-{1}'.format(qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0], vname)
vsize = 2 * qubes.config.defaults['root_img_size'] vsize = 2 * qubes.config.defaults['root_img_size']
log_expected = \ log_expected = \
@ -334,8 +334,8 @@ class TC_92_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBa
def test_002_failing_callback(self): def test_002_failing_callback(self):
''' 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': self.volume_name,
'pool': CallbackBase.pool_name, 'pool': self.pool_name,
'save_on_stop': True, 'save_on_stop': True,
'rw': True, 'rw': True,
'revisions_to_keep': 2, 'revisions_to_keep': 2,

16
qubes/tests/storage_lvm.py
View File

@ -76,9 +76,9 @@ class ThinPoolBase(qubes.tests.QubesTestCase):
'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 cls.pool_class = pool_class
ThinPoolBase.volume_class = volume_class cls.volume_class = volume_class
ThinPoolBase.pool_conf = conf cls.pool_conf = conf
def setUp(self, init_pool=True): def setUp(self, init_pool=True):
super(ThinPoolBase, self).setUp() super(ThinPoolBase, self).setUp()
@ -113,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(**ThinPoolBase.pool_conf)) self.app.add_pool(**self.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):
@ -121,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__, ThinPoolBase.pool_class)] if issubclass(p.__class__, self.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:
@ -169,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, ThinPoolBase.volume_class) self.assertIsInstance(volume, self.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'])
@ -189,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, ThinPoolBase.volume_class) self.assertIsInstance(volume, self.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'])
@ -966,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, ThinPoolBase.volume_class) self.assertIsInstance(volume, self.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'])