From 80ccadb700605781af8d8ca942c6f2ba40a4eac4 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Sun, 18 Oct 2026 22:16:43 +0200 Subject: [PATCH 1/5] Fix run_in_background for compound commands Output redirection applied only to the last command of a compound one, so its earlier parts kept the session output open and the call blocked until they finished. Run the command in a subshell to redirect it as a whole. A simple command is executed by the subshell directly, so the returned PID is still the command's one. Signed-off-by: Robert Baldyga --- connection/base_executor.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/connection/base_executor.py b/connection/base_executor.py index 3be941c..ddad37a 100644 --- a/connection/base_executor.py +++ b/connection/base_executor.py @@ -1,6 +1,7 @@ # # Copyright(c) 2019-2021 Intel Corporation # Copyright(c) 2023-2024 Huawei Technologies Co., Ltd. +# Copyright(c) 2026 Unvertical # SPDX-License-Identifier: BSD-3-Clause # @@ -62,7 +63,10 @@ def run_in_background(self, command, stdout_redirect_path="/dev/null", stderr_redirect_path="/dev/null"): - command += f" > {stdout_redirect_path} 2> {stderr_redirect_path} &echo $!" + # Subshell makes the redirection cover compound commands too, otherwise their + # earlier parts keep the session output open until they finish. A simple command + # is executed by the subshell directly, so the PID is still the command's one. + command = f"({command}) > {stdout_redirect_path} 2> {stderr_redirect_path} &echo $!" output = self.run(command) if output is not None: From c528c6ff8d656d72100941a00272e246f1071d16 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Sun, 18 Oct 2026 20:31:11 +0200 Subject: [PATCH 2/5] Identify DUT by host only Since the ProxyJump support the connection uses 'host', and a config with only 'ip' has been blocked, drop the 'ip' field entirely and require 'host'. The DUT IP is always resolved from the host. Name per-DUT log directories after the host instead of the IP, so they match the DUT config. Signed-off-by: Robert Baldyga --- core/test_run_utils.py | 14 +++----------- log/logger.py | 4 ++-- test_utils/dut.py | 11 +++++++++-- 3 files changed, 14 insertions(+), 15 deletions(-) diff --git a/core/test_run_utils.py b/core/test_run_utils.py index bebb2d0..71dd2e5 100644 --- a/core/test_run_utils.py +++ b/core/test_run_utils.py @@ -11,7 +11,6 @@ import traceback import pytest -from IPy import IP import core.test_run from connection.local_executor import LocalExecutor @@ -142,15 +141,8 @@ def __setup_disks(cls): def __presetup(cls): if cls.config['type'] == 'ssh': - try: - IP(cls.config['ip']) - if not cls.config['host']: - cls.config['host'] = cls.config['ip'] - except ValueError: - TestRun.block("IP address from config is in invalid format.") - except KeyError: - if 'host' not in cls.config: - TestRun.block("No IP address or host defined in config") + if not cls.config.get('host'): + TestRun.block("No host defined in config") port = cls.config.get('port', 22) @@ -185,7 +177,7 @@ def __setup(cls): except Exception as ex: raise Exception(f"Failed to setup DUT instance:\n" f"{str(ex)}\n{traceback.format_exc()}") - cls.dut.ip = cls.dut.ip or cls.executor.resolve_ip_address() + cls.dut.ip = cls.executor.resolve_ip_address() cls.__setup_disks() TestRun.LOGGER.info(f"Re-seeding random number generator with seed: {cls.random_seed}") diff --git a/log/logger.py b/log/logger.py index 622c01c..0338e55 100644 --- a/log/logger.py +++ b/log/logger.py @@ -1,6 +1,7 @@ # # Copyright(c) 2019-2021 Intel Corporation # Copyright(c) 2025 Huawei Technologies Co., Ltd. +# Copyright(c) 2026 Unvertical # SPDX-License-Identifier: BSD-3-Clause # @@ -211,11 +212,10 @@ def get_additional_logs(self): TestRun.executor.run( f"awk '/{test_identifier}/,0' {messages_log} > {log_files['messages.log']}") - dut_identifier = TestRun.dut.ip if TestRun.dut.ip else TestRun.dut.config["host"] for log_name, log_source_path in log_files.items(): try: log_destination_path = os.path.join( - self.base_dir, "dut_info", dut_identifier, log_name + self.base_dir, "dut_info", TestRun.dut.name, log_name ) TestRun.executor.copy_from(log_source_path, log_destination_path) except Exception as e: diff --git a/test_utils/dut.py b/test_utils/dut.py index e1e17fb..727cbbc 100644 --- a/test_utils/dut.py +++ b/test_utils/dut.py @@ -27,10 +27,17 @@ def __init__(self, dut_info): self.spider = dut_info.get("spider") self.wps = dut_info.get("wps") self.env = dut_info.get("env") - self.ip = dut_info.get("ip") + # Resolved from host during setup + self.ip = None + + @property + def name(self): + """Identifies the DUT in logs""" + return self.config.get("host", "localhost") def __str__(self): - dut_str = f"ip: {self.ip}\n" + dut_str = f"host: {self.name}\n" + dut_str += f"ip: {self.ip}\n" dut_str += f'ipmi: {self.ipmi["ip"]}\n' if self.ipmi is not None else "" dut_str += f'spider: {self.spider["ip"]}\n' if self.spider is not None else "" dut_str += ( From 3c33dca3fd0fe5e4b6998b6c7ce126ff74ebc75a Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Sun, 18 Oct 2026 20:31:11 +0200 Subject: [PATCH 3/5] Tag every command in command log with DUT name Command log is a single timeline for all DUTs, so to distinguish between commands executed on each DUT, mark each command with the DUT host name. Signed-off-by: Robert Baldyga --- connection/base_executor.py | 8 ++++++-- connection/ssh_executor.py | 4 ++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/connection/base_executor.py b/connection/base_executor.py index ddad37a..6098cff 100644 --- a/connection/base_executor.py +++ b/connection/base_executor.py @@ -13,6 +13,11 @@ class BaseExecutor: + @property + def name(self): + """Identifies the DUT this executor runs commands on""" + return "localhost" + def _execute(self, command, timeout): raise NotImplementedError() @@ -53,8 +58,7 @@ def run(self, command, timeout: timedelta = timedelta(minutes=30)): if TestRun.dut and TestRun.dut.env: command = f"{TestRun.dut.env} && {command}" command_id = TestRun.LOGGER.get_new_command_id() - ip_info = TestRun.dut.ip if len(TestRun.duts) > 1 else "" - TestRun.LOGGER.write_command_to_command_log(command, command_id, info=ip_info) + TestRun.LOGGER.write_command_to_command_log(command, command_id, info=self.name) output = self._execute(command, timeout) TestRun.LOGGER.write_output_to_command_log(output, command_id) return output diff --git a/connection/ssh_executor.py b/connection/ssh_executor.py index 40e4ede..beaf415 100644 --- a/connection/ssh_executor.py +++ b/connection/ssh_executor.py @@ -27,6 +27,10 @@ def __init__(self, host, username, port=22): self.ssh_config = None self._check_config_for_reboot_timeout() + @property + def name(self): + return self.host + def __del__(self): self.ssh.close() From 07e62990dc861783dbcacfd671beb2ede33e4d7e Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Sun, 18 Oct 2026 20:31:11 +0200 Subject: [PATCH 4/5] serial_monitor: Add serial monitor plugin for libvirt DUTs Collect DUT serial console output into dut_info//serial.log using the libvirt console stream, the same way 'virsh console' does. Output is written as it arrives, so it survives a hung DUT or a crashed controller. Step markers are inserted into the log to correlate it with the test flow. Signed-off-by: Robert Baldyga --- .../serial_monitor_libvirt/__init__.py | 164 ++++++++++++++++++ 1 file changed, 164 insertions(+) create mode 100644 internal_plugins/serial_monitor_libvirt/__init__.py diff --git a/internal_plugins/serial_monitor_libvirt/__init__.py b/internal_plugins/serial_monitor_libvirt/__init__.py new file mode 100644 index 0000000..380375b --- /dev/null +++ b/internal_plugins/serial_monitor_libvirt/__init__.py @@ -0,0 +1,164 @@ +# +# Copyright(c) 2026 Unvertical +# SPDX-License-Identifier: BSD-3-Clause +# + +import os +import threading +from datetime import datetime + +import libvirt + +from core.test_run import TestRun + +# libvirt dispatches stream callbacks from an event loop. Its default implementation +# is process-wide, so it is registered once and serves every plugin instance. +_event_loop_lock = threading.Lock() +_event_loop_started = False + + +def _run_event_loop(): + while True: + libvirt.virEventRunDefaultImpl() + + +def _start_event_loop(): + global _event_loop_started + with _event_loop_lock: + if _event_loop_started: + return + libvirt.virEventRegisterDefaultImpl() + threading.Thread(target=_run_event_loop, name="libvirt-event-loop", daemon=True).start() + _event_loop_started = True + + +class SerialMonitorPlugin: + def __init__(self, params, config): + print("Serial Monitor LibVirt Plugin initialization") + try: + self.url = config["url"] + self.vm_name = config["vm_name"] + except KeyError: + raise Exception("Missing fields in config! ('url','vm_name' are required fields)") + # Name of the console device as in the domain XML alias (e.g. "serial0"); + # None selects the first console. + self.dev_name = config.get("dev_name") + # Take over the console even if someone else (e.g. 'virsh console') holds it + self.force = config.get("force", False) + self.reconnect_interval = config.get("reconnect_interval", 5) + + self.conn = None + self.domain = None + self.stream = None + self.reconnect_timer = None + self.log_file = None + self.lock = threading.Lock() + self.closed = False + + def pre_setup(self): + print("Serial Monitor LibVirt Plugin pre setup") + dut_dir = os.path.join(TestRun.LOGGER.base_dir, "dut_info", TestRun.config["host"]) + os.makedirs(dut_dir, exist_ok=True) + self.log_file = open(os.path.join(dut_dir, "serial.log"), "ab", buffering=0) + + _start_event_loop() + self.conn = libvirt.open(self.url) + self.domain = self.conn.lookupByName(self.vm_name) + if not self.__connect(): + self.__schedule_reconnect() + + def post_setup(self): + pass + + def teardown(self): + with self.lock: + if self.closed: + return + self.closed = True + # Callbacks check 'closed', so libvirt cleanup can run without holding the lock + if self.reconnect_timer is not None: + libvirt.virEventRemoveTimeout(self.reconnect_timer) + self.reconnect_timer = None + self.__close_stream() + if self.conn is not None: + try: + self.conn.close() + except libvirt.libvirtError: + pass + with self.lock: + self.log_file.close() + + def send_to_serial(self, message): + """Insert a marker into the collected serial log (it is not sent to the DUT)""" + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") + self.__write(f"\n[{timestamp}] {message}\n".encode()) + + def __write(self, data): + with self.lock: + if not self.log_file.closed: + self.log_file.write(data) + + def __connect(self): + flags = libvirt.VIR_DOMAIN_CONSOLE_FORCE if self.force else 0 + try: + if not self.conn.isAlive(): + self.conn = libvirt.open(self.url) + self.domain = self.conn.lookupByName(self.vm_name) + stream = self.conn.newStream(libvirt.VIR_STREAM_NONBLOCK) + self.domain.openConsole(self.dev_name, stream, flags) + stream.eventAddCallback( + libvirt.VIR_STREAM_EVENT_READABLE + | libvirt.VIR_STREAM_EVENT_ERROR + | libvirt.VIR_STREAM_EVENT_HANGUP, + self.__stream_event, + None, + ) + except libvirt.libvirtError as e: + self.send_to_serial(f"") + return False + self.stream = stream + self.send_to_serial("") + return True + + def __close_stream(self): + stream, self.stream = self.stream, None + if stream is None: + return + try: + stream.eventRemoveCallback() + stream.abort() + except libvirt.libvirtError: + pass + + def __schedule_reconnect(self): + if self.reconnect_timer is None: + self.reconnect_timer = libvirt.virEventAddTimeout( + self.reconnect_interval * 1000, self.__reconnect, None + ) + + def __reconnect(self, timer, opaque): + if self.closed or not self.__connect(): + return + libvirt.virEventRemoveTimeout(timer) + self.reconnect_timer = None + + def __stream_event(self, stream, events, opaque): + if self.closed: + return + if events & libvirt.VIR_STREAM_EVENT_READABLE: + try: + data = stream.recv(4096) + except libvirt.libvirtError: + data = b"" + if data == -2: + return + if data: + self.__write(data) + return + # EOF, error or hangup - console is gone (e.g. domain was destroyed) + self.send_to_serial("") + self.__close_stream() + self.__schedule_reconnect() + + +plugin_class = SerialMonitorPlugin From e92f837c153682c642b547c119027282da84477e Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Sun, 18 Oct 2026 22:16:43 +0200 Subject: [PATCH 5/5] serial_monitor: Detect kernel warnings and bugs Scan serial console output for kernel warnings and bugs. Warnings are reported in the test log at the next test step and the test continues. A bug interrupts the test by injecting an exception into the test thread with a signal, which also breaks blocking calls, e.g. waiting for a command hung by the bug. The exception is raised only within pytest's runtest hook call, otherwise it is deferred to the next step. Add expect() for tests triggering kernel messages on purpose. Signed-off-by: Robert Baldyga --- .../serial_monitor_libvirt/__init__.py | 180 +++++++++++++++++- 1 file changed, 173 insertions(+), 7 deletions(-) diff --git a/internal_plugins/serial_monitor_libvirt/__init__.py b/internal_plugins/serial_monitor_libvirt/__init__.py index 380375b..26ea673 100644 --- a/internal_plugins/serial_monitor_libvirt/__init__.py +++ b/internal_plugins/serial_monitor_libvirt/__init__.py @@ -4,10 +4,14 @@ # import os +import re +import signal import threading -from datetime import datetime +from contextlib import contextmanager +from datetime import datetime, timedelta import libvirt +import pytest from core.test_run import TestRun @@ -28,10 +32,60 @@ def _start_event_loop(): if _event_loop_started: return libvirt.virEventRegisterDefaultImpl() + # Console errors are logged to serial.log, don't spam stderr with them + libvirt.registerErrorHandler(lambda ctx, error: None, None) threading.Thread(target=_run_event_loop, name="libvirt-event-loop", daemon=True).start() _event_loop_started = True +DEFAULT_WARNING_PATTERNS = [ + r"WARNING:", + r"soft lockup", + r"rcu: INFO: .* detected stalls", + r"INFO: task .* blocked for more than", +] +DEFAULT_BUG_PATTERNS = [ + r"BUG:", + r"kernel BUG at", + r"\bOops:", + r"general protection fault", + r"Kernel panic", +] +# Limit of kernel messages reported to test log per test and log level, the rest is only +# in serial.log +MAX_REPORTS = 10 + +# Bugs detected by the event loop thread, to be raised in the main (test) thread +_pending_bugs = [] + + +def _raise_pending_bugs(): + bugs = [] + while _pending_bugs: + bugs.append(_pending_bugs.pop(0)) + if bugs: + pytest.fail("\n".join(bugs)) + + +def _in_runtest_hook(frame): + while frame is not None: + code = frame.f_code + if code.co_name == "from_call" and code.co_filename.endswith( + os.path.join("_pytest", "runner.py") + ): + return True + frame = frame.f_back + return False + + +def _bug_signal_handler(signum, frame): + # Pytest reports exceptions as test result only inside the runtest hook call; anywhere + # else (e.g. in report generation) it would break pytest itself. In such case the bug + # stays pending until the next test step. + if _in_runtest_hook(frame): + _raise_pending_bugs() + + class SerialMonitorPlugin: def __init__(self, params, config): print("Serial Monitor LibVirt Plugin initialization") @@ -46,6 +100,14 @@ def __init__(self, params, config): # Take over the console even if someone else (e.g. 'virsh console') holds it self.force = config.get("force", False) self.reconnect_interval = config.get("reconnect_interval", 5) + self.warning_patterns = [ + re.compile(p) for p in config.get("warning_patterns", DEFAULT_WARNING_PATTERNS) + ] + self.bug_patterns = [ + re.compile(p) for p in config.get("bug_patterns", DEFAULT_BUG_PATTERNS) + ] + # Interrupt the test on bug, otherwise just report it as an error + self.interrupt_on_bug = config.get("interrupt_on_bug", True) self.conn = None self.domain = None @@ -54,12 +116,27 @@ def __init__(self, params, config): self.log_file = None self.lock = threading.Lock() self.closed = False + self.host = None + self.test_identifier = None + self.line_buffer = b"" + # (log method, message) waiting to be logged from the main thread + self.pending_reports = [] + self.reported = {} + self.bug_found = False + self.expected_kind = None + self.expected_pattern = None + self.expected_seen = threading.Event() + self.expected_bug_found = False def pre_setup(self): print("Serial Monitor LibVirt Plugin pre setup") - dut_dir = os.path.join(TestRun.LOGGER.base_dir, "dut_info", TestRun.config["host"]) + self.host = TestRun.config["host"] + # Test identifier is printed to kernel log and may contain anything + self.test_identifier = TestRun.LOGGER.unique_test_identifier + dut_dir = os.path.join(TestRun.LOGGER.base_dir, "dut_info", self.host) os.makedirs(dut_dir, exist_ok=True) self.log_file = open(os.path.join(dut_dir, "serial.log"), "ab", buffering=0) + signal.signal(signal.SIGUSR1, _bug_signal_handler) _start_event_loop() self.conn = libvirt.open(self.url) @@ -86,13 +163,101 @@ def teardown(self): except libvirt.libvirtError: pass with self.lock: - self.log_file.close() + if self.log_file: + self.log_file.close() + # Bug detected too late to interrupt this test must not interrupt the next one + _pending_bugs.clear() def send_to_serial(self, message): - """Insert a marker into the collected serial log (it is not sent to the DUT)""" + """ + Insert a marker into the collected serial log (it is not sent to the DUT). + Called from the main thread on each test step, so also report kernel messages + detected so far. + """ + self.__mark(message) + self.__flush_reports() + _raise_pending_bugs() + + @contextmanager + def expect(self, kind, pattern, timeout: timedelta = timedelta(seconds=30)): + """ + Expect kernel messages of given kind ("warning" or "bug") within the block. + They are logged as info instead of warning or error, but bugs still interrupt + the test. Fails if no message matching the pattern arrives within the timeout + after the block. Once an expected bug occurs, all following bugs in the test are + considered its aftermath. + """ + if kind not in ("warning", "bug"): + raise ValueError(f"Unknown kernel message kind '{kind}'") + self.expected_seen.clear() + self.expected_pattern = re.compile(pattern) + self.expected_kind = kind + try: + yield + if not self.expected_seen.wait(timeout.total_seconds()): + TestRun.fail( + f"Expected kernel {kind} matching '{pattern}' not found " + f"on {self.host} serial console" + ) + finally: + self.expected_kind = None + self.__flush_reports() + + def __mark(self, message): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") self.__write(f"\n[{timestamp}] {message}\n".encode()) + def __flush_reports(self): + while self.pending_reports: + log_method, message = self.pending_reports.pop(0) + reported = self.reported.get(log_method, 0) + if reported < MAX_REPORTS: + getattr(TestRun.LOGGER, log_method)(message) + elif reported == MAX_REPORTS: + getattr(TestRun.LOGGER, log_method)( + f"More kernel messages on {self.host}, see serial.log" + ) + self.reported[log_method] = reported + 1 + + def __scan(self, data): + lines = (self.line_buffer + data).split(b"\n") + self.line_buffer = lines.pop() + if len(self.line_buffer) > 4096: + lines.append(self.line_buffer) + self.line_buffer = b"" + for raw_line in lines: + line = raw_line.decode(errors="replace").strip() + if self.test_identifier and self.test_identifier in line: + continue + if any(p.search(line) for p in self.warning_patterns): + kind, log_method = "warning", "warning" + elif any(p.search(line) for p in self.bug_patterns): + kind, log_method = "bug", "error" + else: + continue + message = f"Kernel {kind} on {self.host}: {line}" + expected = kind == self.expected_kind + if expected or (kind == "bug" and self.expected_bug_found): + log_method = "info" + message = f"Expected kernel {kind} on {self.host}: {line}" + if kind == "bug": + self.expected_bug_found = True + if expected and self.expected_pattern.search(line): + self.expected_seen.set() + # Logged also for bugs, so that a test catching the exception still fails + self.pending_reports.append((log_method, message)) + if kind == "bug": + self.__interrupt(message) + + def __interrupt(self, message): + # Only the first bug interrupts the test, the following ones are part of its aftermath + if not self.interrupt_on_bug or self.bug_found: + return + self.bug_found = True + _pending_bugs.append(message) + # Signal interrupts also blocking calls, e.g. waiting for output of a hung command + signal.pthread_kill(threading.main_thread().ident, signal.SIGUSR1) + def __write(self, data): with self.lock: if not self.log_file.closed: @@ -114,10 +279,10 @@ def __connect(self): None, ) except libvirt.libvirtError as e: - self.send_to_serial(f"") + self.__mark(f"") return False self.stream = stream - self.send_to_serial("") + self.__mark("") return True def __close_stream(self): @@ -154,9 +319,10 @@ def __stream_event(self, stream, events, opaque): return if data: self.__write(data) + self.__scan(data) return # EOF, error or hangup - console is gone (e.g. domain was destroyed) - self.send_to_serial("") + self.__mark("") self.__close_stream() self.__schedule_reconnect()