From 7b1784708cd75d5cef7b8add002320bec6556762 Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Thu, 23 May 2019 15:38:42 +0300 Subject: OTA-2541: IP Secondary tests (oe-selftest) Signed-off-by: Mike Sul Signed-off-by: Patrick Vacek --- lib/oeqa/selftest/cases/testutils.py | 53 +++++---- lib/oeqa/selftest/cases/updater_qemux86_64.py | 151 ++++++++++++++++++-------- 2 files changed, 138 insertions(+), 66 deletions(-) diff --git a/lib/oeqa/selftest/cases/testutils.py b/lib/oeqa/selftest/cases/testutils.py index 2ad99ad..f8b1904 100644 --- a/lib/oeqa/selftest/cases/testutils.py +++ b/lib/oeqa/selftest/cases/testutils.py @@ -7,49 +7,57 @@ 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=None): - logger = logging.getLogger("selftest") - if imagename is None: - imagename = 'core-image-minimal' - logger.info('Running bitbake to build {}'.format(imagename)) - bitbake(imagename) + +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 = None + 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 = efi - args.machine = machine + args.efi = kwargs.get('efi', False) + args.machine = kwargs.get('machine', None) qemu_use_kvm = get_bb_var("QEMU_USE_KVM") if qemu_use_kvm and \ - (qemu_use_kvm == 'True' and 'x86' in machine or + (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 = True - args.gdb = False - args.pcap = None - args.overlay = None - args.dry_run = False - args.secondary_network = False + 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(10) + sleep(kwargs.get('wait_for_boot_time', 10)) return qemu, s -def qemu_terminate(s): - try: - s.terminate() - except KeyboardInterrupt: - pass +def qemu_bake_image(imagename): + logger.info('Running bitbake to build {}'.format(imagename)) + bitbake(imagename) def qemu_send_command(port, command, timeout=60): @@ -122,7 +130,6 @@ def verifyProvisioned(testInst, machine): 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 = logging.getLogger("selftest") logger.info('Device successfully provisioned with ID: ' + m.group(1)) # vim:set ts=4 sw=4 sts=4 expandtab: diff --git a/lib/oeqa/selftest/cases/updater_qemux86_64.py b/lib/oeqa/selftest/cases/updater_qemux86_64.py index bad7a87..32d3ddb 100644 --- a/lib/oeqa/selftest/cases/updater_qemux86_64.py +++ b/lib/oeqa/selftest/cases/updater_qemux86_64.py @@ -4,11 +4,12 @@ import logging import re 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, \ - akt_native_run, verifyNotProvisioned, verifyProvisioned + akt_native_run, verifyNotProvisioned, verifyProvisioned, qemu_bake_image, qemu_boot_image class GeneralTests(OESelftestTestCase): @@ -309,7 +310,91 @@ class HsmTests(OESelftestTestCase): verifyProvisioned(self, machine) -class SecondaryTests(OESelftestTestCase): +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): + stdout, stderr, retcode = qemu_send_command(self.qemu.ssh_port, cmd, timeout=60) + 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') + + 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-auto-prov "') + + def is_ecu_registered(self, ecu_id): + max_number_of_tries = 20 + 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') @@ -323,57 +408,37 @@ class SecondaryTests(OESelftestTestCase): 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-auto-prov "') - self.qemu, self.s = qemu_launch(machine='qemux86-64', imagename='secondary-image') + + self.primary = IpSecondaryTests.Primary(self) + self.secondary = IpSecondaryTests.Secondary(self) 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_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)) - def test_secondary_present(self): - print('Checking aktualizr-secondary is present') - stdout, stderr, retcode = self.qemu_command('aktualizr-secondary --help') - self.assertEqual(retcode, 0, "Unable to run aktualizr-secondary --help") - self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) + 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())) -class PrimaryTests(OESelftestTestCase): - def setUpLocal(self): - layer = "meta-updater-qemux86-64" - result = runCmd('bitbake-layers show-layers') - if re.search(layer, result.output) is None: - # 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 + "/../../../../../" - 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-auto-prov "') - self.append_config('SOTA_CLIENT_FEATURES = "secondary-network"') - self.qemu, self.s = qemu_launch(machine='qemux86-64', imagename='primary-image') + 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)) - def tearDownLocal(self): - qemu_terminate(self.s) - if self.meta_qemu: - runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) + with self.secondary: + self.assertTrue(self.secondary.was_successfully_booted(), + 'The secondary failed to boot: {}'.format(self.secondary.stderr)) - def qemu_command(self, command): - return qemu_send_command(self.qemu.ssh_port, command) - - def test_aktualizr_present(self): - print('Checking aktualizr is present') - stdout, stderr, retcode = self.qemu_command('aktualizr --help') - self.assertEqual(retcode, 0, "Unable to run aktualizr --help") - self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) + 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): -- cgit v1.2.3-54-g00ecf