diff options
Diffstat (limited to 'external/meta-updater/lib')
7 files changed, 851 insertions, 0 deletions
diff --git a/external/meta-updater/lib/oeqa/selftest/cases/qemucommand.py b/external/meta-updater/lib/oeqa/selftest/cases/qemucommand.py new file mode 120000 index 00000000..075cdb8f --- /dev/null +++ b/external/meta-updater/lib/oeqa/selftest/cases/qemucommand.py @@ -0,0 +1 @@ +../../../../scripts/qemucommand.py
\ No newline at end of file diff --git a/external/meta-updater/lib/oeqa/selftest/cases/testutils.py b/external/meta-updater/lib/oeqa/selftest/cases/testutils.py new file mode 100644 index 00000000..8d618a68 --- /dev/null +++ b/external/meta-updater/lib/oeqa/selftest/cases/testutils.py @@ -0,0 +1,143 @@ +import os +import oe.path +import logging +import re +import subprocess +from time import sleep + +from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_bb_vars +from qemucommand import QemuCommand + +logger = logging.getLogger("selftest") + + +def qemu_launch(efi=False, machine=None, imagename='core-image-minimal', **kwargs): + qemu_bake_image(imagename) + return qemu_boot_image(efi=efi, machine=machine, imagename=imagename, **kwargs) + + +def qemu_terminate(s): + try: + s.terminate() + s.wait(timeout=10) + except KeyboardInterrupt: + pass + + +def qemu_boot_image(imagename, **kwargs): + # Create empty object. + args = type('', (), {})() + args.imagename = imagename + args.mac = kwargs.get('mac', None) + # Could use DEPLOY_DIR_IMAGE here but it's already in the machine + # subdirectory. + args.dir = 'tmp/deploy/images' + args.efi = kwargs.get('efi', False) + args.machine = kwargs.get('machine', None) + args.mem = kwargs.get('mem', '128M') + qemu_use_kvm = get_bb_var("QEMU_USE_KVM") + if qemu_use_kvm and \ + (qemu_use_kvm == 'True' and 'x86' in args.machine or + get_bb_var('MACHINE') in qemu_use_kvm.split()): + args.kvm = True + else: + args.kvm = None # Autodetect + args.no_gui = kwargs.get('no_gui', True) + args.gdb = kwargs.get('gdb', False) + args.pcap = kwargs.get('pcap', None) + args.overlay = kwargs.get('overlay', None) + args.dry_run = kwargs.get('dry_run', False) + args.secondary_network = kwargs.get('secondary_network', False) + + qemu = QemuCommand(args) + cmdline = qemu.command_line() + print('Booting image with run-qemu-ota...') + s = subprocess.Popen(cmdline) + sleep(kwargs.get('wait_for_boot_time', 10)) + return qemu, s + + +def qemu_bake_image(imagename): + logger.info('Running bitbake to build {}'.format(imagename)) + bitbake(imagename) + + +def qemu_send_command(port, command, timeout=120): + command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' + + str(port) + ' "' + command + '"'] + s2 = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + stdout, stderr = s2.communicate(timeout=timeout) + return stdout, stderr, s2.returncode + + +def metadir(): + # Assume the directory layout for finding other layers. We could also + # make assumptions by using 'show-layers', but either way, if the + # layers we need aren't where we expect them, we are out of luck. + path = os.path.abspath(os.path.dirname(__file__)) + metadir = path + "/../../../../../" + + return metadir + + +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 + # + # Requirements in base test class (setUpClass for example): + # bitbake aktualizr-native + # bitbake build-sysroots -c build_native_sysroot + # + # (technique found in poky/meta/lib/oeqa/selftest/cases/package.py) + bb_vars = get_bb_vars(['STAGING_DIR', 'BUILD_ARCH']) + sysroot = oe.path.join(bb_vars['STAGING_DIR'], bb_vars['BUILD_ARCH']) + + result = runCmd(cmd, native_sysroot=sysroot, ignore_status=True, **kwargs) + testInst.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + + +def verifyNotProvisioned(testInst, machine): + print('Checking output of aktualizr-info:') + ran_ok = False + for delay in [5, 5, 5, 5, 10, 10, 10, 10]: + stdout, stderr, retcode = testInst.qemu_command('aktualizr-info') + if retcode == 0 and stderr == b'': + ran_ok = True + break + sleep(delay) + testInst.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) + + # Verify that device has NOT yet provisioned. + testInst.assertIn(b'Couldn\'t load device ID', stdout, + 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) + testInst.assertIn(b'Couldn\'t load ECU serials', stdout, + 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) + testInst.assertIn(b'Provisioned on server: no', stdout, + 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) + testInst.assertIn(b'Fetched metadata: no', stdout, + 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) + + +def verifyProvisioned(testInst, machine): + # Verify that device HAS provisioned. + ran_ok = False + for delay in [5, 5, 5, 5, 10, 10, 10, 10]: + stdout, stderr, retcode = testInst.qemu_command('aktualizr-info') + if retcode == 0 and stderr == b'' and stdout.decode().find('Fetched metadata: yes') >= 0: + ran_ok = True + break + sleep(delay) + testInst.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) + + testInst.assertIn(b'Device ID: ', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode()) + testInst.assertIn(b'Primary ecu hardware ID: ' + machine.encode(), stdout, + 'Provisioning failed: ' + stderr.decode() + stdout.decode()) + testInst.assertIn(b'Fetched metadata: yes', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode()) + p = re.compile(r'Device ID: ([a-z0-9-]*)\n') + m = p.search(stdout.decode()) + testInst.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) + testInst.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) + logger.info('Device successfully provisioned with ID: ' + m.group(1)) + +# vim:set ts=4 sw=4 sts=4 expandtab: diff --git a/external/meta-updater/lib/oeqa/selftest/cases/updater_minnowboard.py b/external/meta-updater/lib/oeqa/selftest/cases/updater_minnowboard.py new file mode 100644 index 00000000..f4da3605 --- /dev/null +++ b/external/meta-updater/lib/oeqa/selftest/cases/updater_minnowboard.py @@ -0,0 +1,54 @@ +import re + +from oeqa.selftest.case import OESelftestTestCase +from oeqa.utils.commands import runCmd, get_bb_var +from testutils import metadir, qemu_launch, qemu_send_command, qemu_terminate, verifyProvisioned + + +class MinnowTests(OESelftestTestCase): + + def setUpLocal(self): + layer_intel = "meta-intel" + layer_minnow = "meta-updater-minnowboard" + result = runCmd('bitbake-layers show-layers') + if re.search(layer_intel, result.output) is None: + self.meta_intel = metadir() + layer_intel + runCmd('bitbake-layers add-layer "%s"' % self.meta_intel) + else: + self.meta_intel = None + if re.search(layer_minnow, result.output) is None: + self.meta_minnow = metadir() + layer_minnow + runCmd('bitbake-layers add-layer "%s"' % self.meta_minnow) + else: + self.meta_minnow = None + self.append_config('MACHINE = "intel-corei7-64"') + self.append_config('OSTREE_BOOTLOADER = "grub"') + self.append_config('SOTA_CLIENT_PROV = " aktualizr-shared-prov "') + self.qemu, self.s = qemu_launch(efi=True, machine='intel-corei7-64', mem='512M') + + def tearDownLocal(self): + qemu_terminate(self.s) + if self.meta_intel: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_intel, ignore_status=True) + if self.meta_minnow: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_minnow, ignore_status=True) + + def qemu_command(self, command): + return qemu_send_command(self.qemu.ssh_port, command) + + def test_provisioning(self): + 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?") + machine = get_bb_var('MACHINE', 'core-image-minimal') + self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) + # Strip off line ending. + value = stdout.decode()[:-1] + self.assertEqual(value, machine, + 'MACHINE does not match hostname: ' + machine + ', ' + value + + '\nIs TianoCore ovmf installed on your host machine?') + + verifyProvisioned(self, machine) + +# vim:set ts=4 sw=4 sts=4 expandtab: diff --git a/external/meta-updater/lib/oeqa/selftest/cases/updater_native.py b/external/meta-updater/lib/oeqa/selftest/cases/updater_native.py new file mode 100644 index 00000000..d2bf341d --- /dev/null +++ b/external/meta-updater/lib/oeqa/selftest/cases/updater_native.py @@ -0,0 +1,48 @@ +# pylint: disable=C0111,C0325 +import logging + +from oeqa.selftest.case import OESelftestTestCase +from oeqa.utils.commands import runCmd, bitbake, get_bb_var +from testutils import akt_native_run + + +class SotaToolsTests(OESelftestTestCase): + + @classmethod + def setUpClass(cls): + super(SotaToolsTests, cls).setUpClass() + logger = logging.getLogger("selftest") + logger.info('Running bitbake to build aktualizr-native tools') + bitbake('aktualizr-native') + bitbake('build-sysroots -c build_native_sysroot') + + def test_push_help(self): + akt_native_run(self, 'garage-push --help') + + def test_deploy_help(self): + akt_native_run(self, 'garage-deploy --help') + + def test_garagesign_help(self): + akt_native_run(self, 'garage-sign --help') + + +class GeneralTests(OESelftestTestCase): + + def test_feature_sota(self): + result = get_bb_var('DISTRO_FEATURES').find('sota') + self.assertNotEqual(result, -1, 'Feature "sota" not set at DISTRO_FEATURES') + + def test_feature_usrmerge(self): + result = get_bb_var('DISTRO_FEATURES').find('usrmerge') + self.assertNotEqual(result, -1, 'Feature "sota" not set at DISTRO_FEATURES') + + def test_feature_systemd(self): + result = get_bb_var('DISTRO_FEATURES').find('systemd') + self.assertNotEqual(result, -1, 'Feature "systemd" not set at DISTRO_FEATURES') + + def test_java(self): + result = runCmd('which java', ignore_status=True) + self.assertEqual(result.status, 0, + "Java not found. Do you have a JDK installed on your host machine?") + +# vim:set ts=4 sw=4 sts=4 expandtab: diff --git a/external/meta-updater/lib/oeqa/selftest/cases/updater_qemux86_64.py b/external/meta-updater/lib/oeqa/selftest/cases/updater_qemux86_64.py new file mode 100644 index 00000000..2b4726cb --- /dev/null +++ b/external/meta-updater/lib/oeqa/selftest/cases/updater_qemux86_64.py @@ -0,0 +1,476 @@ +# pylint: disable=C0111,C0325 +import os +import logging +import re +import subprocess +import unittest +from time import sleep +from uuid import uuid4 + +from oeqa.selftest.case import OESelftestTestCase +from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_bb_vars +from testutils import qemu_launch, qemu_send_command, qemu_terminate, \ + metadir, akt_native_run, verifyNotProvisioned, verifyProvisioned, \ + qemu_bake_image, qemu_boot_image + + +class GeneralTests(OESelftestTestCase): + def test_credentials(self): + logger = logging.getLogger("selftest") + logger.info('Running bitbake to build core-image-minimal') + self.append_config('SOTA_CLIENT_PROV = "aktualizr-shared-prov"') + + # note: this also tests ostreepush/garagesign/garagecheck which are + # omitted from other test cases + bitbake('core-image-minimal') + credentials = get_bb_var('SOTA_PACKED_CREDENTIALS') + # skip the test if the variable SOTA_PACKED_CREDENTIALS is not set + if credentials is None: + raise unittest.SkipTest("Variable 'SOTA_PACKED_CREDENTIALS' not set.") + # Check if the file exists + self.assertTrue(os.path.isfile(credentials), "File %s does not exist" % credentials) + deploydir = get_bb_var('DEPLOY_DIR_IMAGE') + imagename = get_bb_var('IMAGE_LINK_NAME', 'core-image-minimal') + # Check if the credentials are included in the output image + result = runCmd('tar -jtvf %s/%s.tar.bz2 | grep sota_provisioning_credentials.zip' % + (deploydir, imagename), ignore_status=True) + self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + + +class AktualizrToolsTests(OESelftestTestCase): + + @classmethod + def setUpClass(cls): + super(AktualizrToolsTests, cls).setUpClass() + logger = logging.getLogger("selftest") + logger.info('Running bitbake to build aktualizr-native tools') + bitbake('aktualizr-native aktualizr-device-prov') + bitbake('build-sysroots -c build_native_sysroot') + + def test_cert_provider_help(self): + akt_native_run(self, 'aktualizr-cert-provider --help') + + def test_cert_provider_local_output(self): + 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(['WORKDIR', 'libdir'], 'aktualizr-device-prov') + config = bb_vars_prov['WORKDIR'] + '/sysroot-destdir' + bb_vars_prov['libdir'] + '/sota/conf.d/20-sota-device-cred.toml' + + 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 + '/var/sota/import/client.pem' + self.assertTrue(os.path.isfile(cert_path), "Client certificate not found at %s." % cert_path) + self.assertTrue(os.path.getsize(cert_path) > 0, "Client certificate at %s is empty." % cert_path) + pkey_path = temp_dir + '/var/sota/import/pkey.pem' + self.assertTrue(os.path.isfile(pkey_path), "Private key not found at %s." % pkey_path) + self.assertTrue(os.path.getsize(pkey_path) > 0, "Private key at %s is empty." % pkey_path) + ca_path = temp_dir + '/var/sota/import/root.crt' + self.assertTrue(os.path.isfile(ca_path), "Client certificate not found at %s." % ca_path) + self.assertTrue(os.path.getsize(ca_path) > 0, "Client certificate at %s is empty." % ca_path) + + +class SharedCredProvTests(OESelftestTestCase): + + def setUpLocal(self): + layer = "meta-updater-qemux86-64" + result = runCmd('bitbake-layers show-layers') + if re.search(layer, result.output) is None: + self.meta_qemu = metadir() + layer + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) + else: + self.meta_qemu = None + self.append_config('MACHINE = "qemux86-64"') + self.append_config('SOTA_CLIENT_PROV = " aktualizr-shared-prov "') + self.append_config('IMAGE_FSTYPES_remove = "ostreepush garagesign garagecheck"') + self.qemu, self.s = qemu_launch(machine='qemux86-64') + + def tearDownLocal(self): + qemu_terminate(self.s) + if self.meta_qemu: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + + def qemu_command(self, command): + return qemu_send_command(self.qemu.ssh_port, command) + + def test_provisioning(self): + 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?") + machine = get_bb_var('MACHINE', 'core-image-minimal') + self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) + # Strip off line ending. + value = stdout.decode()[:-1] + self.assertEqual(value, machine, + 'MACHINE does not match hostname: ' + machine + ', ' + value) + + verifyProvisioned(self, machine) + + +class ManualControlTests(OESelftestTestCase): + + def setUpLocal(self): + layer = "meta-updater-qemux86-64" + result = runCmd('bitbake-layers show-layers') + if re.search(layer, result.output) is None: + self.meta_qemu = metadir() + layer + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) + else: + self.meta_qemu = None + self.append_config('MACHINE = "qemux86-64"') + self.append_config('SOTA_CLIENT_PROV = " aktualizr-shared-prov "') + self.append_config('SYSTEMD_AUTO_ENABLE_aktualizr = "disable"') + self.append_config('IMAGE_FSTYPES_remove = "ostreepush garagesign garagecheck"') + self.qemu, self.s = qemu_launch(machine='qemux86-64') + + def tearDownLocal(self): + qemu_terminate(self.s) + if self.meta_qemu: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + + def qemu_command(self, command): + return qemu_send_command(self.qemu.ssh_port, command) + + def test_manual_run_mode_once(self): + """ + Disable the systemd service then run aktualizr manually + """ + sleep(20) + stdout, stderr, retcode = self.qemu_command('aktualizr-info') + self.assertIn(b'Can\'t open database', stderr, + 'Aktualizr should not have run yet' + stderr.decode() + stdout.decode()) + + stdout, stderr, retcode = self.qemu_command('aktualizr once') + + stdout, stderr, retcode = self.qemu_command('aktualizr-info') + self.assertIn(b'Fetched metadata: yes', stdout, + 'Aktualizr should have run' + stderr.decode() + stdout.decode()) + + +class DeviceCredProvTests(OESelftestTestCase): + + def setUpLocal(self): + layer = "meta-updater-qemux86-64" + result = runCmd('bitbake-layers show-layers') + if re.search(layer, result.output) is None: + self.meta_qemu = metadir() + layer + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) + else: + self.meta_qemu = None + self.append_config('MACHINE = "qemux86-64"') + self.append_config('SOTA_CLIENT_PROV = " aktualizr-device-prov "') + self.append_config('SOTA_DEPLOY_CREDENTIALS = "0"') + self.append_config('IMAGE_FSTYPES_remove = "ostreepush garagesign garagecheck"') + self.qemu, self.s = qemu_launch(machine='qemux86-64') + bitbake('build-sysroots -c build_native_sysroot') + + def tearDownLocal(self): + qemu_terminate(self.s) + if self.meta_qemu: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + + def qemu_command(self, command): + return qemu_send_command(self.qemu.ssh_port, command) + + def test_provisioning(self): + 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?") + machine = get_bb_var('MACHINE', 'core-image-minimal') + self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) + # Strip off line ending. + value = stdout.decode()[:-1] + self.assertEqual(value, machine, + 'MACHINE does not match hostname: ' + machine + ', ' + value) + + verifyNotProvisioned(self, machine) + + # Run aktualizr-cert-provider. + bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') + creds = bb_vars['SOTA_PACKED_CREDENTIALS'] + bb_vars_prov = get_bb_vars(['WORKDIR', 'libdir'], 'aktualizr-device-prov') + config = bb_vars_prov['WORKDIR'] + '/sysroot-destdir' + bb_vars_prov['libdir'] + '/sota/conf.d/20-sota-device-cred.toml' + + print('Provisining at root@localhost:%d' % self.qemu.ssh_port) + akt_native_run(self, 'aktualizr-cert-provider -c {creds} -t root@localhost -p {port} -s -u -r -g {config}' + .format(creds=creds, port=self.qemu.ssh_port, config=config)) + + verifyProvisioned(self, machine) + + +class DeviceCredProvHsmTests(OESelftestTestCase): + + def setUpLocal(self): + layer = "meta-updater-qemux86-64" + result = runCmd('bitbake-layers show-layers') + if re.search(layer, result.output) is None: + self.meta_qemu = metadir() + layer + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) + else: + self.meta_qemu = None + self.append_config('MACHINE = "qemux86-64"') + self.append_config('SOTA_CLIENT_PROV = "aktualizr-device-prov-hsm"') + self.append_config('SOTA_DEPLOY_CREDENTIALS = "0"') + self.append_config('SOTA_CLIENT_FEATURES = "hsm"') + self.append_config('IMAGE_INSTALL_append = " softhsm-testtoken"') + self.append_config('IMAGE_FSTYPES_remove = "ostreepush garagesign garagecheck"') + self.qemu, self.s = qemu_launch(machine='qemux86-64') + bitbake('build-sysroots -c build_native_sysroot') + + def tearDownLocal(self): + qemu_terminate(self.s) + if self.meta_qemu: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + + def qemu_command(self, command): + return qemu_send_command(self.qemu.ssh_port, command) + + def test_provisioning(self): + 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?") + machine = get_bb_var('MACHINE', 'core-image-minimal') + self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) + # Strip off line ending. + value = stdout.decode()[:-1] + self.assertEqual(value, machine, + 'MACHINE does not match hostname: ' + machine + ', ' + value) + + verifyNotProvisioned(self, machine) + + # Verify that HSM is not yet initialized. + 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()) + softhsm2_command = 'softhsm2-util --show-slots' + stdout, stderr, retcode = self.qemu_command(softhsm2_command) + self.assertNotEqual(retcode, 0, 'softhsm2-tool succeeded before initialization: ' + + stdout.decode() + stderr.decode()) + + # Run aktualizr-cert-provider. + bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') + creds = bb_vars['SOTA_PACKED_CREDENTIALS'] + bb_vars_prov = get_bb_vars(['WORKDIR', 'libdir'], 'aktualizr-device-prov-hsm') + config = bb_vars_prov['WORKDIR'] + '/sysroot-destdir' + bb_vars_prov['libdir'] + '/sota/conf.d/20-sota-device-cred-hsm.toml' + + akt_native_run(self, 'aktualizr-cert-provider -c {creds} -t root@localhost -p {port} -r -s -u -g {config}' + .format(creds=creds, port=self.qemu.ssh_port, config=config)) + + # Verify that HSM is able to initialize. + for delay in [5, 5, 5, 5, 10]: + sleep(delay) + p11_out, p11_err, p11_ret = self.qemu_command(pkcs11_command) + hsm_out, hsm_err, hsm_ret = self.qemu_command(softhsm2_command) + if (p11_ret == 0 and hsm_ret == 0 and hsm_err == b'' and + b'X.509 cert' in p11_out and b'present token' in p11_err): + break + else: + self.fail('pkcs11-tool or softhsm2-tool failed: ' + p11_err.decode() + + p11_out.decode() + hsm_err.decode() + hsm_out.decode()) + + self.assertIn(b'Initialized: yes', hsm_out, 'softhsm2-tool failed: ' + + hsm_err.decode() + hsm_out.decode()) + self.assertIn(b'User PIN init.: yes', hsm_out, 'softhsm2-tool failed: ' + + hsm_err.decode() + hsm_out.decode()) + + # Check that pkcs11 output matches sofhsm output. + p11_p = re.compile(r'Using slot [0-9] with a present token \((0x[0-9a-f]*)\)\s') + p11_m = p11_p.search(p11_err.decode()) + self.assertTrue(p11_m, 'Slot number not found with pkcs11-tool: ' + p11_err.decode() + p11_out.decode()) + self.assertGreater(p11_m.lastindex, 0, 'Slot number not found with pkcs11-tool: ' + + p11_err.decode() + p11_out.decode()) + hsm_p = re.compile(r'Description:\s*SoftHSM slot ID (0x[0-9a-f]*)\s') + hsm_m = hsm_p.search(hsm_out.decode()) + self.assertTrue(hsm_m, 'Slot number not found with softhsm2-tool: ' + hsm_err.decode() + hsm_out.decode()) + self.assertGreater(hsm_m.lastindex, 0, 'Slot number not found with softhsm2-tool: ' + + hsm_err.decode() + hsm_out.decode()) + self.assertEqual(p11_m.group(1), hsm_m.group(1), 'Slot number does not match: ' + + p11_err.decode() + p11_out.decode() + hsm_err.decode() + hsm_out.decode()) + + verifyProvisioned(self, machine) + + +class IpSecondaryTests(OESelftestTestCase): + + class Image: + def __init__(self, imagename, binaryname, machine='qemux86-64', bake=True, **kwargs): + self.machine = machine + self.imagename = imagename + self.boot_kwargs = kwargs + self.binaryname = binaryname + self.stdout = '' + self.stderr = '' + self.retcode = 0 + if bake: + self.bake() + + def bake(self): + self.configure() + qemu_bake_image(self.imagename) + + def send_command(self, cmd, timeout=60): + stdout, stderr, retcode = qemu_send_command(self.qemu.ssh_port, cmd, timeout=timeout) + return str(stdout), str(stderr), retcode + + def __enter__(self): + self.qemu, self.process = qemu_boot_image(machine=self.machine, imagename=self.imagename, + wait_for_boot_time=1, **self.boot_kwargs) + # wait until the VM is booted and is SSHable + self.wait_till_sshable() + + def __exit__(self, exc_type, exc_val, exc_tb): + qemu_terminate(self.process) + + def wait_till_sshable(self): + # qemu_send_command tries to ssh into the qemu VM and blocks until it gets there or timeout happens + # so it helps us to block q control flow until the VM is booted and a target binary/daemon is running there + self.stdout, self.stderr, self.retcode = self.send_command(self.binaryname + ' --help', timeout=300) + + def was_successfully_booted(self): + return self.retcode == 0 + + class Secondary(Image): + def __init__(self, test_ctx): + self._test_ctx = test_ctx + self.sndry_serial = str(uuid4()) + self.sndry_hw_id = 'qemux86-64-oeselftest-sndry' + self.id = (self.sndry_hw_id, self.sndry_serial) + super(IpSecondaryTests.Secondary, self).__init__('secondary-image', 'aktualizr-secondary', + secondary_network=True) + + def configure(self): + self._test_ctx.append_config('SECONDARY_SERIAL_ID = "{}"'.format(self.sndry_serial)) + self._test_ctx.append_config('SECONDARY_HARDWARE_ID = "{}"'.format(self.sndry_hw_id)) + + class Primary(Image): + def __init__(self, test_ctx): + self._test_ctx = test_ctx + super(IpSecondaryTests.Primary, self).__init__('primary-image', 'aktualizr', secondary_network=True) + + def configure(self): + self._test_ctx.append_config('MACHINE = "qemux86-64"') + self._test_ctx.append_config('SOTA_CLIENT_PROV = " aktualizr-shared-prov "') + + def is_ecu_registered(self, ecu_id): + max_number_of_tries = 40 + try_counter = 0 + + # aktualizr-info is not always able to load ECU serials from DB + # so, let's run it a few times until it actually succeeds + while try_counter < max_number_of_tries: + device_status = self.get_info() + try_counter += 1 + if device_status.find("load ECU serials") == -1: + break + sleep(1) + + if not ((device_status.find(ecu_id[0]) != -1) and (device_status.find(ecu_id[1]) != -1)): + return False + not_registered_field = "Removed or not registered ecus:" + not_reg_start = device_status.find(not_registered_field) + return not_reg_start == -1 or (device_status.find(ecu_id[1], not_reg_start) == -1) + + def get_info(self): + stdout, stderr, retcode = self.send_command('aktualizr-info') + self._test_ctx.assertEqual(retcode, 0, 'Unable to run aktualizr-info: {}'.format(stderr)) + return stdout + + def setUpLocal(self): + layer = "meta-updater-qemux86-64" + result = runCmd('bitbake-layers show-layers') + if re.search(layer, result.output) is None: + self.meta_qemu = metadir() + layer + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) + else: + self.meta_qemu = None + + self.append_config('IMAGE_FSTYPES_remove = "ostreepush garagesign garagecheck"') + self.primary = IpSecondaryTests.Primary(self) + self.secondary = IpSecondaryTests.Secondary(self) + + def tearDownLocal(self): + if self.meta_qemu: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + + def test_ip_secondary_registration_if_secondary_starts_first(self): + with self.secondary: + self.assertTrue(self.secondary.was_successfully_booted(), + 'The secondary failed to boot: {}'.format(self.secondary.stderr)) + + with self.primary: + self.assertTrue(self.primary.was_successfully_booted(), + 'The primary failed to boot: {}'.format(self.primary.stderr)) + + self.assertTrue(self.primary.is_ecu_registered(self.secondary.id), + "The secondary wasn't registered at the primary: {}".format(self.primary.get_info())) + + def test_ip_secondary_registration_if_primary_starts_first(self): + with self.primary: + self.assertTrue(self.primary.was_successfully_booted(), + 'The primary failed to boot: {}'.format(self.primary.stderr)) + + with self.secondary: + self.assertTrue(self.secondary.was_successfully_booted(), + 'The secondary failed to boot: {}'.format(self.secondary.stderr)) + + self.assertTrue(self.primary.is_ecu_registered(self.secondary.id), + "The secondary wasn't registered at the primary: {}".format(self.primary.get_info())) + + +class ResourceControlTests(OESelftestTestCase): + def setUpLocal(self): + layer = "meta-updater-qemux86-64" + result = runCmd('bitbake-layers show-layers') + if re.search(layer, result.output) is None: + self.meta_qemu = metadir() + layer + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) + else: + self.meta_qemu = None + self.append_config('MACHINE = "qemux86-64"') + self.append_config('SOTA_CLIENT_PROV = " aktualizr-shared-prov "') + self.append_config('IMAGE_FSTYPES_remove = "ostreepush garagesign garagecheck"') + self.append_config('IMAGE_INSTALL_append += " aktualizr-resource-control "') + self.append_config('RESOURCE_CPU_WEIGHT_pn-aktualizr = "1000"') + self.append_config('RESOURCE_MEMORY_HIGH_pn-aktualizr = "50M"') + self.append_config('RESOURCE_MEMORY_MAX_pn-aktualizr = "1M"') + self.qemu, self.s = qemu_launch(machine='qemux86-64') + + def tearDownLocal(self): + qemu_terminate(self.s) + if self.meta_qemu: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + + def qemu_command(self, command): + return qemu_send_command(self.qemu.ssh_port, command) + + def test_aktualizr_resource_control(self): + print('Checking aktualizr was killed') + ran_ok = False + for delay in [5, 5, 5, 5]: + sleep(delay) + try: + stdout, stderr, retcode = self.qemu_command('systemctl --no-pager show aktualizr') + if retcode == 0 and b'ExecMainStatus=9' in stdout: + ran_ok = True + break + except subprocess.TimeoutExpired: + pass + self.assertTrue(ran_ok, 'Aktualizr was not killed') + + self.assertIn(b'CPUWeight=1000', stdout, 'CPUWeight was not set correctly') + self.assertIn(b'MemoryHigh=52428800', stdout, 'MemoryHigh was not set correctly') + self.assertIn(b'MemoryMax=1048576', stdout, 'MemoryMax was not set correctly') + + self.qemu_command('systemctl --runtime set-property aktualizr MemoryMax=') + self.qemu_command('systemctl restart aktualizr') + + stdout, stderr, retcode = self.qemu_command('systemctl --no-pager show --property=ExecMainStatus aktualizr') + self.assertIn(b'ExecMainStatus=0', stdout, 'Aktualizr did not restart') + +# vim:set ts=4 sw=4 sts=4 expandtab: diff --git a/external/meta-updater/lib/oeqa/selftest/cases/updater_qemux86_64_ptest.py b/external/meta-updater/lib/oeqa/selftest/cases/updater_qemux86_64_ptest.py new file mode 100644 index 00000000..d20a9f0e --- /dev/null +++ b/external/meta-updater/lib/oeqa/selftest/cases/updater_qemux86_64_ptest.py @@ -0,0 +1,47 @@ +# pylint: disable=C0111,C0325 +import re + +from oeqa.selftest.case import OESelftestTestCase +from oeqa.utils.commands import runCmd +from testutils import metadir, qemu_launch, qemu_send_command, qemu_terminate + + +class PtestTests(OESelftestTestCase): + + def setUpLocal(self): + layer = "meta-updater-qemux86-64" + result = runCmd('bitbake-layers show-layers') + if re.search(layer, result.output) is None: + self.meta_qemu = metadir() + layer + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) + else: + self.meta_qemu = None + self.append_config('MACHINE = "qemux86-64"') + self.append_config('SYSTEMD_AUTO_ENABLE_aktualizr = "disable"') + self.append_config('PTEST_ENABLED_pn-aktualizr = "1"') + self.append_config('IMAGE_INSTALL_append += "aktualizr-ptest ptest-runner "') + self.append_config('IMAGE_FSTYPES_remove = "ostreepush garagesign garagecheck"') + self.qemu, self.s = qemu_launch(machine='qemux86-64', mem="768M") + + def tearDownLocal(self): + qemu_terminate(self.s) + if self.meta_qemu: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + + def qemu_command(self, command, timeout=60): + return qemu_send_command(self.qemu.ssh_port, command, timeout=timeout) + + def test_run_ptests(self): + # simulate a login shell, so that /usr/sbin is in $PATH (from /etc/profile) + stdout, stderr, retcode = self.qemu_command('sh -l -c ptest-runner', timeout=None) + output = stdout.decode() + print(output) + + has_failure = re.search('^FAIL', output, flags=re.MULTILINE) is not None + if has_failure: + print("Full test suite log:") + stdout, _, _ = self.qemu_command('cat /tmp/aktualizr-ptest.log || cat /tmp/aktualizr-ptest.log.tmp', timeout=None) + print(stdout.decode(errors='replace')) + + self.assertEqual(retcode, 0) + self.assertFalse(has_failure) diff --git a/external/meta-updater/lib/oeqa/selftest/cases/updater_raspberrypi.py b/external/meta-updater/lib/oeqa/selftest/cases/updater_raspberrypi.py new file mode 100644 index 00000000..26d5c4c6 --- /dev/null +++ b/external/meta-updater/lib/oeqa/selftest/cases/updater_raspberrypi.py @@ -0,0 +1,82 @@ +# pylint: disable=C0111,C0325 +import os +import logging +import re +import unittest + +from oeqa.selftest.case import OESelftestTestCase +from oeqa.utils.commands import runCmd, bitbake, get_bb_var + +from testutils import metadir + + +class RpiTests(OESelftestTestCase): + + def setUpLocal(self): + # Add layers before changing the machine type, otherwise the sanity + # checker complains loudly. + layer_python = "meta-openembedded/meta-python" + layer_rpi = "meta-raspberrypi" + layer_upd_rpi = "meta-updater-raspberrypi" + result = runCmd('bitbake-layers show-layers') + if re.search(layer_python, result.output) is None: + self.meta_python = metadir() + layer_python + runCmd('bitbake-layers add-layer "%s"' % self.meta_python) + else: + self.meta_python = None + if re.search(layer_rpi, result.output) is None: + self.meta_rpi = metadir() + layer_rpi + runCmd('bitbake-layers add-layer "%s"' % self.meta_rpi) + else: + self.meta_rpi = None + if re.search(layer_upd_rpi, result.output) is None: + self.meta_upd_rpi = metadir() + layer_upd_rpi + runCmd('bitbake-layers add-layer "%s"' % self.meta_upd_rpi) + else: + self.meta_upd_rpi = None + + # This is trickier that I would've thought. The fundamental problem is + # that the qemu layer changes the u-boot file extension to .rom, but + # raspberrypi still expects .bin. To prevent this, the qemu layer must + # be temporarily removed if it is present. It has to be removed by name + # without the complete path, but to add it back when we are done, we + # need the full path. + p = re.compile(r'meta-updater-qemux86-64\s*(\S*meta-updater-qemux86-64)\s') + m = p.search(result.output) + if m and m.lastindex > 0: + self.meta_qemu = m.group(1) + runCmd('bitbake-layers remove-layer meta-updater-qemux86-64') + else: + self.meta_qemu = None + + self.append_config('MACHINE = "raspberrypi3"') + self.append_config('SOTA_CLIENT_PROV = " aktualizr-shared-prov "') + + def tearDownLocal(self): + if self.meta_qemu: + runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu, ignore_status=True) + if self.meta_upd_rpi: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_upd_rpi, ignore_status=True) + if self.meta_rpi: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_rpi, ignore_status=True) + if self.meta_python: + runCmd('bitbake-layers remove-layer "%s"' % self.meta_python, ignore_status=True) + + def test_build(self): + logger = logging.getLogger("selftest") + logger.info('Running bitbake to build core-image-minimal') + bitbake('core-image-minimal') + credentials = get_bb_var('SOTA_PACKED_CREDENTIALS') + # Skip the test if the variable SOTA_PACKED_CREDENTIALS is not set. + if credentials is None: + raise unittest.SkipTest("Variable 'SOTA_PACKED_CREDENTIALS' not set.") + # Check if the file exists. + self.assertTrue(os.path.isfile(credentials), "File %s does not exist" % credentials) + deploydir = get_bb_var('DEPLOY_DIR_IMAGE') + imagename = get_bb_var('IMAGE_LINK_NAME', 'core-image-minimal') + # Check if the credentials are included in the output image. + result = runCmd('tar -jtvf %s/%s.tar.bz2 | grep sota_provisioning_credentials.zip' % + (deploydir, imagename), ignore_status=True) + self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) + +# vim:set ts=4 sw=4 sts=4 expandtab: |