# 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}"
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)}"
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
'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')
import sys
import signal
import time
+from collections import defaultdict
from .ssh import OESSHTarget
from oeqa.utils.qemurunner import QemuRunner
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)
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:
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))
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
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())
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)
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
# 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:
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")
