diff options
| -rw-r--r-- | meta/classes/testimage.bbclass | 39 | ||||
| -rw-r--r-- | meta/lib/oeqa/core/target/qemu.py | 7 | ||||
| -rw-r--r-- | meta/lib/oeqa/utils/qemurunner.py | 31 |
3 files changed, 70 insertions, 7 deletions
diff --git a/meta/classes/testimage.bbclass b/meta/classes/testimage.bbclass index 0d55c3247e..75f0f2c3e3 100644 --- a/meta/classes/testimage.bbclass +++ b/meta/classes/testimage.bbclass @@ -34,6 +34,17 @@ TESTIMAGE_AUTO ??= "0" # TEST_QEMUPARAMS can be used to pass extra parameters to qemu, e.g. "-m 1024" for setting the amount of ram to 1 GB. # TEST_RUNQEMUPARAMS can be used to pass extra parameters to runqemu, e.g. "gl" to enable OpenGL acceleration. +# TESTIMAGE_BOOT_PATTERNS can be used to override certain patterns used to communicate with the target when booting, +# if a pattern is not specifically present on this variable a default will be used when booting the target. +# TESTIMAGE_BOOT_PATTERNS[<flag>] overrides the pattern used for that specific flag, where flag comes from a list of accepted flags +# e.g. normally the system boots and waits for a login prompt (login:), after that it sends the command: "root\n" to log as the root user +# if we wanted to log in as the hypothetical "webserver" user for example we could set the following: +# TESTIMAGE_BOOT_PATTERNS = "send_login_user search_login_succeeded" +# TESTIMAGE_BOOT_PATTERNS[send_login_user] = "webserver\n" +# TESTIMAGE_BOOT_PATTERNS[search_login_succeeded] = "webserver@[a-zA-Z0-9\-]+:~#" +# The accepted flags are the following: search_reached_prompt, send_login_user, search_login_succeeded, search_cmd_finished. +# They are prefixed with either search/send, to differentiate if the pattern is meant to be sent or searched to/from the target terminal + TEST_LOG_DIR ?= "${WORKDIR}/testimage" TEST_EXPORT_DIR ?= "${TMPDIR}/testimage/${PN}" @@ -68,6 +79,8 @@ TEST_TARGET ?= "qemu" TEST_QEMUPARAMS ?= "" TEST_RUNQEMUPARAMS ?= "" +TESTIMAGE_BOOT_PATTERNS ?= "" + TESTIMAGEDEPENDS = "" TESTIMAGEDEPENDS_append_qemuall = " qemu-native:do_populate_sysroot qemu-helper-native:do_populate_sysroot qemu-helper-native:do_addto_recipe_sysroot" TESTIMAGEDEPENDS += "${@bb.utils.contains('IMAGE_PKGTYPE', 'rpm', 'cpio-native:do_populate_sysroot', '', d)}" @@ -150,6 +163,29 @@ def get_testimage_json_result_dir(d): def get_testimage_result_id(configuration): return '%s_%s_%s_%s' % (configuration['TEST_TYPE'], configuration['IMAGE_BASENAME'], configuration['MACHINE'], configuration['STARTTIME']) +def get_testimage_boot_patterns(d): + from collections import defaultdict + boot_patterns = defaultdict(str) + # Only accept certain values + accepted_patterns = ['search_reached_prompt', 'send_login_user', 'search_login_succeeded', 'search_cmd_finished'] + # Not all patterns need to be overriden, e.g. perhaps we only want to change the user + boot_patterns_flags = d.getVarFlags('TESTIMAGE_BOOT_PATTERNS') or {} + if boot_patterns_flags: + patterns_set = [p for p in boot_patterns_flags.items() if p[0] in d.getVar('TESTIMAGE_BOOT_PATTERNS').split()] + for flag, flagval in patterns_set: + if flag not in accepted_patterns: + bb.fatal('Testimage: The only accepted boot patterns are: search_reached_prompt,send_login_user, \ + search_login_succeeded,search_cmd_finished\n Make sure your TESTIMAGE_BOOT_PATTERNS=%s \ + contains an accepted flag.' % d.getVar('TESTIMAGE_BOOT_PATTERNS')) + return + # We know boot prompt is searched through in binary format, others might be expressions + if flag == 'search_reached_prompt': + boot_patterns[flag] = flagval.encode() + else: + boot_patterns[flag] = flagval.encode().decode('unicode-escape') + return boot_patterns + + def testimage_main(d): import os import json @@ -259,6 +295,9 @@ def testimage_main(d): 'serial_ports': len(d.getVar("SERIAL_CONSOLES").split()), } + if d.getVar("TESTIMAGE_BOOT_PATTERNS"): + target_kwargs['boot_patterns'] = get_testimage_boot_patterns(d) + # TODO: Currently BBPATH is needed for custom loading of targets. # It would be better to find these modules using instrospection. target_kwargs['target_modules_path'] = d.getVar('BBPATH') diff --git a/meta/lib/oeqa/core/target/qemu.py b/meta/lib/oeqa/core/target/qemu.py index 008a9f03ce..059106e915 100644 --- a/meta/lib/oeqa/core/target/qemu.py +++ b/meta/lib/oeqa/core/target/qemu.py @@ -8,6 +8,7 @@ import os import sys import signal import time +from collections import defaultdict from .ssh import OESSHTarget from oeqa.utils.qemurunner import QemuRunner @@ -18,7 +19,8 @@ class OEQemuTarget(OESSHTarget): def __init__(self, logger, server_ip, timeout=300, user='root', port=None, machine='', rootfs='', kernel='', kvm=False, slirp=False, dump_dir='', dump_host_cmds='', display='', bootlog='', - tmpdir='', dir_image='', boottime=60, serial_ports=2, **kwargs): + tmpdir='', dir_image='', boottime=60, serial_ports=2, + boot_patterns = defaultdict(str), **kwargs): super(OEQemuTarget, self).__init__(logger, None, server_ip, timeout, user, port) @@ -30,13 +32,14 @@ class OEQemuTarget(OESSHTarget): self.kernel = kernel self.kvm = kvm self.use_slirp = slirp + self.boot_patterns = boot_patterns self.runner = QemuRunner(machine=machine, rootfs=rootfs, tmpdir=tmpdir, deploy_dir_image=dir_image, display=display, logfile=bootlog, boottime=boottime, use_kvm=kvm, use_slirp=slirp, dump_dir=dump_dir, dump_host_cmds=dump_host_cmds, logger=logger, - serial_ports=serial_ports) + serial_ports=serial_ports, boot_patterns = boot_patterns) def start(self, params=None, extra_bootparams=None, runqemuparams=''): if self.use_slirp and not self.server_ip: diff --git a/meta/lib/oeqa/utils/qemurunner.py b/meta/lib/oeqa/utils/qemurunner.py index 4704422211..ed74ea8fad 100644 --- a/meta/lib/oeqa/utils/qemurunner.py +++ b/meta/lib/oeqa/utils/qemurunner.py @@ -21,6 +21,7 @@ import threading import codecs import logging from oeqa.utils.dump import HostDumper +from collections import defaultdict # Get Unicode non printable control chars control_range = list(range(0,32))+list(range(127,160)) @@ -31,7 +32,7 @@ re_control_char = re.compile('[%s]' % re.escape("".join(control_chars))) class QemuRunner: def __init__(self, machine, rootfs, display, tmpdir, deploy_dir_image, logfile, boottime, dump_dir, dump_host_cmds, - use_kvm, logger, use_slirp=False, serial_ports=2): + use_kvm, logger, use_slirp=False, serial_ports=2, boot_patterns = defaultdict(str)): # Popen object for runqemu self.runqemu = None @@ -57,6 +58,7 @@ class QemuRunner: self.use_slirp = use_slirp self.serial_ports = serial_ports self.msg = '' + self.boot_patterns = boot_patterns self.runqemutime = 120 self.qemu_pidfile = 'pidfile_'+str(os.getpid()) @@ -65,6 +67,25 @@ class QemuRunner: self.logger = logger + # Enable testing other OS's + # Set commands for target communication, and default to Linux ALWAYS + # Other OS's or baremetal applications need to provide their + # own implementation passing it through QemuRunner's constructor + # or by passing them through TESTIMAGE_BOOT_PATTERNS[flag] + # provided variables, where <flag> is one of the mentioned below. + accepted_patterns = ['search_reached_prompt', 'send_login_user', 'search_login_succeeded', 'search_cmd_finished'] + default_boot_patterns = defaultdict(str) + # Default to the usual paterns used to communicate with the target + default_boot_patterns['search_reached_prompt'] = b' login:' + default_boot_patterns['send_login_user'] = 'root\n' + default_boot_patterns['search_login_succeeded'] = r"root@[a-zA-Z0-9\-]+:~#" + default_boot_patterns['search_cmd_finished'] = r"[a-zA-Z0-9]+@[a-zA-Z0-9\-]+:~#" + + # Only override patterns that were set e.g. login user TESTIMAGE_BOOT_PATTERNS[send_login_user] = "webserver\n" + for pattern in accepted_patterns: + if not self.boot_patterns[pattern]: + self.boot_patterns[pattern] = default_boot_patterns[pattern] + def create_socket(self): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -321,7 +342,7 @@ class QemuRunner: self.log(data) data = b'' - if b' login:' in bootlog: + if self.boot_patterns['search_reached_prompt'] in bootlog: self.server_socket = qemusock stopread = True reachedlogin = True @@ -353,8 +374,8 @@ class QemuRunner: # If we are not able to login the tests can continue try: - (status, output) = self.run_serial("root\n", raw=True) - if re.search(r"root@[a-zA-Z0-9\-]+:~#", output): + (status, output) = self.run_serial(self.boot_patterns['send_login_user'], raw=True) + if re.search(self.boot_patterns['search_login_succeeded'], output): self.logged = True self.logger.debug("Logged as root in serial console") if netconf: @@ -478,7 +499,7 @@ class QemuRunner: if answer: data += answer.decode('utf-8') # Search the prompt to stop - if re.search(r"[a-zA-Z0-9]+@[a-zA-Z0-9\-]+:~#", data): + if re.search(self.boot_patterns['search_cmd_finished'], data): break else: raise Exception("No data on serial console socket") |