From fc84de73eb3f04f1e7600a1c86061af7f856e8d4 Mon Sep 17 00:00:00 2001 From: Laurent Bonnans Date: Tue, 20 Feb 2018 17:43:52 +0100 Subject: Refactor oe selftests Factorize the call of aktualizr-native commands in each testing class --- lib/oeqa/selftest/updater.py | 121 +++++++++++++++++++++++++------------------ 1 file changed, 70 insertions(+), 51 deletions(-) diff --git a/lib/oeqa/selftest/updater.py b/lib/oeqa/selftest/updater.py index 703df91..8563e3b 100644 --- a/lib/oeqa/selftest/updater.py +++ b/lib/oeqa/selftest/updater.py @@ -16,32 +16,32 @@ class SotaToolsTests(oeSelfTest): @classmethod def setUpClass(cls): logger = logging.getLogger("selftest") + bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'], + 'aktualizr-native') + cls.sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix'] + cls.sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + cls.libdir = bb_vars['libdir'] + logger.info('Running bitbake to build aktualizr-native tools') bitbake('aktualizr-native') - def test_push_help(self): - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native') - l = bb_vars['libdir'] - p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/garage-push" - self.assertTrue(os.path.isfile(p), msg = "No garage-push found (%s)" % p) - result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True) + def runNativeCmd(self, cmd, **kwargs): + program, *_ = cmd.split(' ') + p = '{}/{}'.format(self.sysrootbin, program) + self.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p)) + env = dict(os.environ) + env['LD_LIBRARY_PATH'] = self.libdir + result = runCmd(cmd, env=env, native_sysroot=self.sysroot, ignore_status=True, **kwargs) self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + def test_push_help(self): + self.runNativeCmd('garage-push --help') + def test_deploy_help(self): - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native') - l = bb_vars['libdir'] - p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/garage-deploy" - self.assertTrue(os.path.isfile(p), msg = "No garage-deploy found (%s)" % p) - result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True) - self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + self.runNativeCmd('garage-deploy --help') def test_garagesign_help(self): - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native') - l = bb_vars['libdir'] - p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/garage-sign" - self.assertTrue(os.path.isfile(p), msg = "No garage-sign found (%s)" % p) - result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True) - self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + self.runNativeCmd('garage-sign --help') class GeneralTests(oeSelfTest): @@ -113,24 +113,29 @@ class AktualizrToolsTests(oeSelfTest): @classmethod def setUpClass(cls): logger = logging.getLogger("selftest") + bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'], + 'aktualizr-native') + cls.sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix'] + cls.sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + cls.libdir = bb_vars['libdir'] + logger.info('Running bitbake to build aktualizr-native tools') bitbake('aktualizr-native') - def test_implicit_writer_help(self): - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native') - l = bb_vars['libdir'] - p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_implicit_writer" - self.assertTrue(os.path.isfile(p), msg = "No aktualizr_implicit_writer found (%s)" % p) - result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True) + def runNativeCmd(self, cmd, **kwargs): + program, *_ = cmd.split(' ') + p = '{}/{}'.format(self.sysrootbin, program) + self.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p)) + env = dict(os.environ) + env['LD_LIBRARY_PATH'] = self.libdir + result = runCmd(cmd, env=env, native_sysroot=self.sysroot, ignore_status=True, **kwargs) self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + def test_implicit_writer_help(self): + self.runNativeCmd('aktualizr_implicit_writer --help') + def test_cert_provider_help(self): - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native') - l = bb_vars['libdir'] - p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_cert_provider" - self.assertTrue(os.path.isfile(p), msg = "No aktualizr_cert_provider found (%s)" % p) - result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True) - self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + self.runNativeCmd('aktualizr_cert_provider --help') def test_cert_provider_local_output(self): logger = logging.getLogger("selftest") @@ -138,17 +143,14 @@ class AktualizrToolsTests(oeSelfTest): bitbake('aktualizr-implicit-prov') bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir', 'SOTA_PACKED_CREDENTIALS', 'T'], 'aktualizr-native') - l = bb_vars['libdir'] - p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_cert_provider" creds = bb_vars['SOTA_PACKED_CREDENTIALS'] temp_dir = bb_vars['T'] bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-implicit-prov') config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov.toml' - self.assertTrue(os.path.isfile(p), msg = "No aktualizr_cert_provider found (%s)" % p) - command = 'LD_LIBRARY_PATH=' + l + ' ' + p + ' -c ' + creds + ' -r -l ' + temp_dir + ' -g ' + config - # logger.info('Checking output of: ' + command) - result = runCmd(command, ignore_status=True) - self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + + self.runNativeCmd('aktualizr_cert_provider -c {creds} -r -l {temp} -g {config}' + .format(creds=creds, temp=temp_dir, config=config)) + # Might be nice if these names weren't hardcoded. cert_path = temp_dir + '/client.pem' self.assertTrue(os.path.isfile(cert_path), "Client certificate not found at %s." % cert_path) @@ -178,7 +180,7 @@ class QemuTests(oeSelfTest): print('Checking machine name (hostname) of device:') stdout, stderr, retcode = self.qemu_command('hostname') self.assertEqual(retcode, 0, "Unable to check hostname. " + - "Is an ssh daemon (such as dropbear or openssh) installed on the device?") + "Is an ssh daemon (such as dropbear or openssh) installed on the device?") machine = get_bb_var('MACHINE', 'core-image-minimal') self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) # Strip off line ending. @@ -224,7 +226,7 @@ class GrubTests(oeSelfTest): print('Checking machine name (hostname) of device:') stdout, stderr, retcode = self.qemu_command('hostname') self.assertEqual(retcode, 0, "Unable to check hostname. " + - "Is an ssh daemon (such as dropbear or openssh) installed on the device?") + "Is an ssh daemon (such as dropbear or openssh) installed on the device?") machine = get_bb_var('MACHINE', 'core-image-minimal') self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) # Strip off line ending. @@ -246,6 +248,18 @@ class GrubTests(oeSelfTest): class HsmTests(oeSelfTest): + @classmethod + def setUpClass(cls): + logger = logging.getLogger("selftest") + bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'], + 'aktualizr-native') + cls.sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix'] + cls.sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + cls.libdir = bb_vars['libdir'] + + logger.info('Running bitbake to build aktualizr-native tools') + bitbake('aktualizr-native') + def setUpLocal(self): self.append_config('SOTA_CLIENT_PROV = "aktualizr-hsm-prov"') self.append_config('SOTA_CLIENT_FEATURES = "hsm"') @@ -254,6 +268,15 @@ class HsmTests(oeSelfTest): def tearDownLocal(self): qemu_terminate(self.s) + def runNativeCmd(self, cmd, **kwargs): + program, *_ = cmd.split(' ') + p = '{}/{}'.format(self.sysrootbin, program) + self.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p)) + env = dict(os.environ) + env['LD_LIBRARY_PATH'] = self.libdir + result = runCmd(cmd, env=env, native_sysroot=self.sysroot, ignore_status=True, **kwargs) + self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + def qemu_command(self, command): return qemu_send_command(self.qemu.ssh_port, command) @@ -261,7 +284,7 @@ class HsmTests(oeSelfTest): print('Checking machine name (hostname) of device:') stdout, stderr, retcode = self.qemu_command('hostname') self.assertEqual(retcode, 0, "Unable to check hostname. " + - "Is an ssh daemon (such as dropbear or openssh) installed on the device?") + "Is an ssh daemon (such as dropbear or openssh) installed on the device?") machine = get_bb_var('MACHINE', 'core-image-minimal') self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) # Strip off line ending. @@ -291,7 +314,7 @@ class HsmTests(oeSelfTest): pkcs11_command = 'pkcs11-tool --module=/usr/lib/softhsm/libsofthsm2.so -O' stdout, stderr, retcode = self.qemu_command(pkcs11_command) self.assertNotEqual(retcode, 0, 'pkcs11-tool succeeded before initialization: ' + - stdout.decode() + stderr.decode()) + stdout.decode() + stderr.decode()) softhsm2_command = 'softhsm2-util --show-slots' stdout, stderr, retcode = self.qemu_command(softhsm2_command) self.assertNotEqual(retcode, 0, 'softhsm2-tool succeeded before initialization: ' + @@ -300,18 +323,12 @@ class HsmTests(oeSelfTest): # Run cert_provider. bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir', 'SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') - l = bb_vars['libdir'] - p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_cert_provider" creds = bb_vars['SOTA_PACKED_CREDENTIALS'] bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-hsm-prov') config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov.toml' - self.assertTrue(os.path.isfile(p), msg = "No aktualizr_cert_provider found (%s)" % p) - command = ('LD_LIBRARY_PATH=' + l + ' ' + p + ' -c ' + creds + ' -t root@localhost -p ' + - str(self.qemu.ssh_port) + ' -r -s -g ' + config) - logger = logging.getLogger("selftest") - # logger.info('Checking output of: ' + command) - result = runCmd(command, ignore_status=True) - self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + + self.runNativeCmd('aktualizr_cert_provider -c {creds} -t root@localhost -p {port} -r -s -g {config}' + .format(creds=creds, port=self.qemu.ssh_port, config=config)) # Verify that HSM is able to initialize. ran_ok = False @@ -361,7 +378,7 @@ class HsmTests(oeSelfTest): m = p.search(stdout.decode()) self.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) self.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) - logger.info('Device successfully provisioned with ID: ' + m.group(1)) + self.logger.info('Device successfully provisioned with ID: ' + m.group(1)) def qemu_launch(efi=False, machine=None): @@ -391,12 +408,14 @@ def qemu_launch(efi=False, machine=None): sleep(10) return qemu, s + def qemu_terminate(s): try: s.terminate() except KeyboardInterrupt: pass + def qemu_send_command(port, command): command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' + str(port) + ' "' + command + '"'] -- cgit v1.2.3-54-g00ecf