# # The Qubes OS Project, http://www.qubes-os.org # # Copyright (C) 2020 David Hobach # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public # License as published by the Free Software Foundation; either # version 2.1 of the License, or (at your option) any later version. # # This library is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, see . # ''' Tests for the callback storage driver. They are mostly based upon the lvm storage driver tests. ''' # pylint: disable=line-too-long import os import json import subprocess import qubes.tests import qubes.tests.storage import qubes.tests.storage_lvm from qubes.tests.storage_lvm import skipUnlessLvmPoolExists from qubes.storage.callback import CallbackPool, CallbackVolume CB_CONF = '/etc/qubes_callback.json' LOG_BIN = '/tmp/testCbLogArgs' CB_DATA = {'utest-callback-01': { 'bdriver': 'lvm_thin', 'bdriver_args': { 'volume_group': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0], 'thin_pool': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[1] }, 'description': 'For unit testing of the callback pool driver.' }, 'utest-callback-02': { 'bdriver': 'lvm_thin', 'bdriver_args': { 'volume_group': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0], 'thin_pool': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[1] }, 'cmd': LOG_BIN, 'description': 'For unit testing of the callback pool driver.' }, 'utest-callback-03': { 'bdriver': 'lvm_thin', 'bdriver_args': { 'volume_group': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0], 'thin_pool': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[1] }, 'cmd': 'exit 1', 'pre_sinit': LOG_BIN + ' pre_sinit', 'pre_setup': LOG_BIN + ' pre_setup', 'pre_volume_create': LOG_BIN + ' pre_volume_create', 'post_volume_create': LOG_BIN + ' post_volume_create', 'pre_volume_import_data': LOG_BIN + ' pre_volume_import_data', 'post_volume_import_data_end': LOG_BIN + ' post_volume_import_data_end', 'post_volume_remove': LOG_BIN + ' post_volume_remove', 'post_destroy': '-', 'description': 'For unit testing of the callback pool driver.' }, 'testing-fail-missing-all': { }, 'testing-fail-missing-bdriver-args': { 'bdriver': 'file', 'description': 'For unit testing of the callback pool driver.' }, 'testing-fail-incorrect-bdriver': { 'bdriver': 'nonexisting-bdriver', 'bdriver_args': { 'foo': 'bar', 'bla': 'blub' }, 'cmd': 'echo foo', 'description': 'For unit testing of the callback pool driver.' }, } class CallbackBase: ''' Mixin base class for callback tests. Has no base class. ''' conf_id = None pool_name = 'test-callback' @classmethod def setUpClass(cls, conf_id='utest-callback-01'): conf = {'name': cls.pool_name, 'driver': 'callback', '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 sudo = [] if os.getuid() == 0 else ['sudo'] subprocess.run(sudo + ['install', '-m', '666', '/dev/null', CB_CONF], check=True) with open(CB_CONF, 'w') as outfile: json.dump(CB_DATA, outfile) super().setUpClass(pool_class=CallbackPool, volume_class=CallbackVolume, pool_conf=conf) @classmethod def tearDownClass(cls): super().tearDownClass() sudo = [] if os.getuid() == 0 else ['sudo'] subprocess.run(sudo + ['rm', '-f', CB_CONF], check=True) def setUp(self, init_pool=True): super().setUp(init_pool=init_pool) if init_pool: #tests from other pools will assume that they're fully initialized after calling __init__() self.loop.run_until_complete(self.pool._assert_initialized()) def test_000_000_callback_test_init(self): ''' Check whether the test init did work. ''' if hasattr(self, 'pool'): self.assertIsInstance(self.pool, CallbackPool) self.assertEqual(self.pool.backend_class, qubes.storage.lvm.ThinPool) self.assertTrue(os.path.isfile(CB_CONF)) @skipUnlessLvmPoolExists class TC_00_CallbackPool(CallbackBase, qubes.tests.storage_lvm.TC_00_ThinPool): pass @skipUnlessLvmPoolExists class TC_01_CallbackPool(CallbackBase, qubes.tests.storage_lvm.TC_01_ThinPool): pass @skipUnlessLvmPoolExists class TC_02_cb_StorageHelpers(CallbackBase, qubes.tests.storage_lvm.TC_02_StorageHelpers): pass class LoggingCallbackBase(CallbackBase): ''' Mixin base class that sets up LOG_BIN and removes `LoggingCallbackBase.test_log`, if needed. ''' test_log = '/tmp/cb_tests.log' test_log_expected = None #dict: class + test name --> test index (int, 0..x) --> expected _additional_ log content volume_name = 'volume_name' xml_path = '/tmp/qubes-test-callback.xml' @classmethod def setUpClass(cls, conf_id=None, log_expected=None): script = """#!/bin/bash i=1 for arg in "$@" ; do echo "$i: $arg" >> "LOG_OUT" (( i++)) done exit 0 """ script = script.replace('LOG_OUT', cls.test_log) with open(LOG_BIN, 'w') as f: f.write(script) os.chmod(LOG_BIN, 0o775) cls.test_log_expected = log_expected super().setUpClass(conf_id=conf_id) @classmethod def tearDownClass(cls): super().tearDownClass() os.remove(LOG_BIN) def setUp(self, init_pool=False): 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 xml = """ linux-kernel black """ xml = xml.replace('CONF_ID', self.conf_id) xml = xml.replace('POOL_NAME', self.pool_name) with open(self.xml_path, 'w') as f: f.write(xml) self.app = qubes.Qubes(self.xml_path, clockvm=None, updatevm=None, offline_mode=True, ) os.environ['QUBES_XML_PATH'] = self.xml_path super().setUp(init_pool=init_pool) def tearDown(self): super().tearDown() os.unlink(self.app.store) self.app.close() del self.app for attr in dir(self): if isinstance(getattr(self, attr), qubes.vm.BaseVM): delattr(self, attr) if os.path.exists(self.test_log): os.remove(self.test_log) if os.path.exists(self.xml_path): os.remove(self.xml_path) def assertLogContent(self, expected): ''' Assert that the log matches the given string. :param expected: Expected content of the log file (String). ''' try: with open(self.test_log, 'r') as f: found = f.read() except FileNotFoundError: found = '' if expected != '': expected = expected + '\n' self.assertEqual(found, expected) def assertLog(self, test_name, ind=0): ''' Assert that the log matches the expected status. :param test_name: Name of the test. :param ind: Index inside `test_log_expected` to check against (Integer starting at 0). ''' d = self.test_log_expected[str(self.__class__) + test_name] expected = [] for i in range(ind+1): expected = expected + [d[i]] expected = filter(None, expected) self.assertLogContent('\n'.join(expected)) def test_001_callbacks(self): ''' create a lvm pool with additional callbacks ''' config = { 'name': self.volume_name, 'pool': self.pool_name, 'save_on_stop': True, 'rw': True, 'revisions_to_keep': 2, 'size': qubes.config.defaults['root_img_size'], } new_size = 2 * qubes.config.defaults['root_img_size'] test_name = 'test_001_callbacks' self.assertLog(test_name, 0) self.init_pool() self.assertFalse(self.created_pool) self.assertIsInstance(self.pool, CallbackPool) self.assertLog(test_name, 1) vm = qubes.tests.storage.TestVM(self) volume = self.app.get_pool(self.pool.name).init_volume(vm, config) self.assertLog(test_name, 2) self.loop.run_until_complete(volume.create()) self.assertLog(test_name, 3) self.loop.run_until_complete(volume.import_data(new_size)) self.assertLog(test_name, 4) self.loop.run_until_complete(volume.import_data_end(True)) self.assertLog(test_name, 5) self.assertEqual(volume.size, new_size) self.loop.run_until_complete(volume.remove()) self.assertLog(test_name, 6) @skipUnlessLvmPoolExists class TC_91_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBase): ''' Tests for the actual callback functionality. conf_id = utest-callback-02 ''' @classmethod def setUpClass(cls): conf_id = 'utest-callback-02' name = cls.pool_name bdriver = (CB_DATA[conf_id])['bdriver'] ctor_params = json.dumps(CB_DATA[conf_id], sort_keys=True, indent=2) vname = cls.volume_name 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'] log_expected = \ {str(cls) + 'test_001_callbacks': {0: '', 1: '', 2: '', 3: '1: {0}\n2: {1}\n3: pre_sinit\n4: {2}\n5: {3}\n1: {0}\n2: {1}\n3: pre_volume_create\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None\n1: {0}\n2: {1}\n3: post_volume_create\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None'.format(name, bdriver, conf_id, ctor_params, vname, vid), 4: '1: {0}\n2: {1}\n3: pre_volume_import_data\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None\n9: {6}'.format(name, bdriver, conf_id, ctor_params, vname, vid, vsize), 5: '1: {0}\n2: {1}\n3: post_volume_import_data_end\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None\n9: {6}'.format(name, bdriver, conf_id, ctor_params, vname, vid, True), 6: '1: {0}\n2: {1}\n3: post_volume_remove\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None'.format(name, bdriver, conf_id, ctor_params, vname, vid), } } super().setUpClass(conf_id=conf_id, log_expected=log_expected) @skipUnlessLvmPoolExists class TC_92_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBase): ''' Tests for the actual callback functionality. conf_id = utest-callback-03 ''' @classmethod def setUpClass(cls): log_expected = \ {str(cls) + 'test_001_callbacks': {0: '', 1: '', 2: '', 3: '1: pre_sinit\n1: pre_volume_create\n1: post_volume_create', 4: '1: pre_volume_import_data', 5: '1: post_volume_import_data_end', 6: '1: post_volume_remove', } } super().setUpClass(conf_id='utest-callback-03', log_expected=log_expected) def test_002_failing_callback(self): ''' Make sure that we check the exit code of executed callbacks. ''' config = { 'name': self.volume_name, 'pool': self.pool_name, 'save_on_stop': True, 'rw': True, 'revisions_to_keep': 2, 'size': qubes.config.defaults['root_img_size'], } self.init_pool() vm = qubes.tests.storage.TestVM(self) volume = self.app.get_pool(self.pool.name).init_volume(vm, config) with self.assertRaises(subprocess.CalledProcessError) as cm: #should trigger the `exit 1` of `cmd` self.loop.run_until_complete(volume.start()) self.assertTrue('exit status 1' in str(cm.exception)) def test_003_errors(self): ''' Make sure we error out on common user & dev mistakes. ''' #missing conf_id with self.assertRaises(qubes.storage.StoragePoolException): cb = CallbackPool(name='some-name', conf_id='') #invalid conf_id with self.assertRaises(qubes.storage.StoragePoolException): cb = CallbackPool(name='some-name', conf_id='nonexisting-id') #incorrect backend driver with self.assertRaises(qubes.storage.StoragePoolException): cb = CallbackPool(name='some-name', conf_id='testing-fail-incorrect-bdriver') #missing config entries with self.assertRaises(qubes.storage.StoragePoolException): cb = CallbackPool(name='some-name', conf_id='testing-fail-missing-all') #missing bdriver args with self.assertRaises(TypeError): cb = CallbackPool(name='some-name', conf_id='testing-fail-missing-bdriver-args') class TC_93_CallbackPool(qubes.tests.QubesTestCase): def test_001_missing_conf(self): ''' A missing config file must cause errors. ''' with self.assertRaises(FileNotFoundError): cb = CallbackPool(name='some-name', conf_id='nonexisting-id')