From 5c70505c127e58aa14c25599c6c7405e77e4c360 Mon Sep 17 00:00:00 2001 From: Laurent Bonnans Date: Wed, 21 Feb 2018 12:10:01 +0100 Subject: Factorize aktualizr native helper in oe-selftest --- lib/oeqa/selftest/updater.py | 118 ++++++++++++++++--------------------------- 1 file changed, 43 insertions(+), 75 deletions(-) diff --git a/lib/oeqa/selftest/updater.py b/lib/oeqa/selftest/updater.py index cad5b2a..f237364 100644 --- a/lib/oeqa/selftest/updater.py +++ b/lib/oeqa/selftest/updater.py @@ -16,32 +16,17 @@ 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 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') + akt_native_run(self, 'garage-push --help') def test_deploy_help(self): - self.runNativeCmd('garage-deploy --help') + akt_native_run(self, 'garage-deploy --help') def test_garagesign_help(self): - self.runNativeCmd('garage-sign --help') + akt_native_run(self, 'garage-sign --help') class GeneralTests(oeSelfTest): @@ -113,43 +98,27 @@ 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 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') + akt_native_run(self, 'aktualizr_implicit_writer --help') def test_cert_provider_help(self): - self.runNativeCmd('aktualizr_cert_provider --help') + akt_native_run(self, 'aktualizr_cert_provider --help') def test_cert_provider_local_output(self): logger = logging.getLogger("selftest") logger.info('Running bitbake to build aktualizr-implicit-prov') bitbake('aktualizr-implicit-prov') - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir', - 'SOTA_PACKED_CREDENTIALS', 'T'], 'aktualizr-native') + bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS', 'T'], 'aktualizr-native') 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.runNativeCmd('aktualizr_cert_provider -c {creds} -r -l {temp} -g {config}' - .format(creds=creds, temp=temp_dir, config=config)) + akt_native_run(self, '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' @@ -250,11 +219,9 @@ class ImplProvTests(oeSelfTest): @classmethod def setUpClass(cls): - 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 = logging.getLogger("selftest") + logger.info('Running bitbake to build aktualizr-native tools') + bitbake('aktualizr-native') def setUpLocal(self): self.append_config('SOTA_CLIENT_PROV = " aktualizr-implicit-prov "') @@ -263,15 +230,6 @@ class ImplProvTests(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) @@ -306,14 +264,13 @@ class ImplProvTests(oeSelfTest): 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) # Run cert_provider. - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir', - 'SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') + bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') creds = bb_vars['SOTA_PACKED_CREDENTIALS'] 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.runNativeCmd('aktualizr_cert_provider -c {creds} -t root@localhost -p {port} -s -g {config}' \ - .format(creds=creds, port=self.qemu.ssh_port, config=config)) + akt_native_run(self, 'aktualizr_cert_provider -c {creds} -t root@localhost -p {port} -s -g {config}' + .format(creds=creds, port=self.qemu.ssh_port, config=config)) # Verify that device HAS provisioned. ran_ok = False @@ -339,11 +296,9 @@ class HsmTests(oeSelfTest): @classmethod def setUpClass(cls): - 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 = logging.getLogger("selftest") + logger.info('Running bitbake to build aktualizr-native tools') + bitbake('aktualizr-native') def setUpLocal(self): self.append_config('SOTA_CLIENT_PROV = "aktualizr-hsm-prov"') @@ -353,15 +308,6 @@ 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) @@ -406,14 +352,13 @@ class HsmTests(oeSelfTest): stdout.decode() + stderr.decode()) # Run cert_provider. - bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir', - 'SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') + bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') 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.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)) + akt_native_run(self, '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 @@ -509,4 +454,27 @@ def qemu_send_command(port, command): stdout, stderr = s2.communicate() return stdout, stderr, s2.returncode + +def akt_native_run(testInst, cmd, **kwargs): + # run a command supplied by aktualizr-native and checks that: + # - the executable exists + # - the command runs without error + # NOTE: the base test class must have built aktualizr-native (in + # setUpClass, for example) + bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'], + 'aktualizr-native') + sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix'] + sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + libdir = bb_vars['libdir'] + + program, *_ = cmd.split(' ') + p = '{}/{}'.format(sysrootbin, program) + testInst.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p)) + env = dict(os.environ) + env['LD_LIBRARY_PATH'] = libdir + result = runCmd(cmd, env=env, native_sysroot=sysroot, ignore_status=True, **kwargs) + testInst.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + + + # vim:set ts=4 sw=4 sts=4 expandtab: -- cgit v1.2.3-54-g00ecf