tests: Add serial monitor + self-test

test/functional/config/example_dut_config.yml

@@ -1,9 +1,11 @@
 # DUT type should be either "local" or "ssh"
-# if ssh is selected, then ip and user fields are required and you have to make sure you have
-# enabled SSH key-based authentication to the machine with given ip address
+# if ssh is selected, then host field is required and you have to make sure you have
+# enabled SSH key-based authentication to the machine with given host. The host is
+# resolved through the controller's SSH config and /etc/hosts. If user is not given,
+# the one from SSH config is used.
 type: "local"
 
-# ip: "x.x.x.x"
+# host: "dut1"
 # user: "example_user"
 
 # env: "HTTPS_PROXY=myproxy.example:12345"

test/functional/requirements.txt

@@ -1,7 +1,6 @@
 pytest>=7.0,<=7.4.4
 multimethod>=1.1
 paramiko>=2.7.2
-IPy>=1.00
 aenum>=2.2.1
 packaging>=20.3
 typing>=3.7.4.1

test/functional/tests/conftest.py

@@ -78,8 +78,8 @@ def pytest_runtest_setup(item):
     # There should be dut config file added to config package and
     # pytest should be executed with option --dut-config=conf_name'.
     #
-    # 'ip' field should be filled with valid IP string to use remote ssh executor
-    # or it should be commented out when user want to execute tests on local machine
+    # 'type' should be 'ssh' with 'host' field filled to use remote ssh executor
+    # or 'local' when user want to execute tests on local machine
     #
     # User can also have own test wrapper, which runs test prepare, cleanup, etc.
     # Then it should be placed in plugins package
@@ -109,7 +109,7 @@ def pytest_runtest_setup(item):
             )
 
         dut_config["plugins_dir"] = os.path.join(os.path.dirname(__file__), "..", "lib")
-        dut_config["opt_plugins"] = {"test_wrapper": {}, "serial_log": {}, "power_control": {}}
+        dut_config["opt_plugins"] = {"test_wrapper": {}, "serial_monitor": {}, "power_control": {}}
         dut_config["extra_logs"] = {"cas": cas_log_test_part_path}
 
         try:
@@ -309,12 +309,7 @@ def pytest_runtest_teardown():
         with TestRun.use_dut(dut):
             if TestRun.executor:
                 os.makedirs(
-                    os.path.join(
-                        TestRun.LOGGER.base_dir,
-                        "dut_info",
-                        dut.ip if dut.ip is not None else dut.config.get("host"),
-                    ),
-                    exist_ok=True,
+                    os.path.join(TestRun.LOGGER.base_dir, "dut_info", dut.name), exist_ok=True
                 )
                 # opencas.log is never rotated, so collect only this test's part of it
                 test_identifier = re.escape(TestRun.LOGGER.unique_test_identifier)

test/functional/tests/self-test/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2026 Unvertical
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/self-test/test_power_control.py

@@ -0,0 +1,50 @@
+#
+# Copyright(c) 2026 Unvertical
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from core.test_run import TestRun
+from test_tools.fs_tools import check_if_file_exists, read_file, remove, write_file
+
+
+@pytest.mark.require_plugin("power_control")
+def test_power_control_power_cycle():
+    """
+    title: Power control resets DUT.
+    description: |
+      Power cycle DUT using power control plugin and check if DUT was hard reset,
+      not gracefully rebooted, and if connection to it was restored.
+    pass_criteria:
+      - DUT is reachable after power cycle
+      - DUT boot ID changed
+      - Data not written back before power cycle is lost
+    """
+    power_control = TestRun.plugin_manager.get_plugin("power_control")
+    # /tmp may be tmpfs, the file has to be on persistent storage
+    marker_path = "/var/tmp/power_control_marker"
+    marker = "power control self-test"
+
+    with TestRun.step("Read boot ID"):
+        boot_id = read_file("/proc/sys/kernel/random/boot_id").strip()
+
+    with TestRun.step("Disable periodic writeback and write file without sync"):
+        TestRun.executor.run_expect_success("sysctl -w vm.dirty_writeback_centisecs=0")
+        remove(marker_path, force=True)
+        TestRun.executor.run_expect_success("sync")
+        write_file(marker_path, marker)
+
+    with TestRun.step("Power cycle DUT"):
+        power_control.power_cycle(wait_for_connection=True)
+
+    with TestRun.step("Check if DUT was rebooted"):
+        new_boot_id = read_file("/proc/sys/kernel/random/boot_id").strip()
+        if new_boot_id == boot_id:
+            TestRun.fail(f"Boot ID did not change ({boot_id})")
+
+    with TestRun.step("Check if unsynced data was lost"):
+        # Graceful reboot would sync the page cache, power cycle must not
+        if check_if_file_exists(marker_path) and read_file(marker_path).strip() == marker:
+            TestRun.fail("Unsynced file survived power cycle, DUT was not hard reset")
+        remove(marker_path, force=True)

test/functional/tests/self-test/test_serial_monitor.py

@@ -0,0 +1,140 @@
+#
+# Copyright(c) 2026 Unvertical
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import posixpath
+import time
+import pytest
+
+from core.test_run import TestRun
+from test_tools.fs_tools import check_if_directory_exists, create_directory, write_file
+
+# Module init triggers the kernel message selected at build time and fails,
+# so the module never stays loaded (unless it dies in init).
+trigger_source = """
+#include <linux/module.h>
+
+#if defined(TRIGGER_NULL)
+static int *volatile null_ptr;
+#endif
+
+static int __init trigger_init(void)
+{
+#if defined(TRIGGER_WARN)
+\tWARN(1, "serial monitor test warning");
+#elif defined(TRIGGER_BUG)
+\tBUG();
+#elif defined(TRIGGER_NULL)
+\t*null_ptr = 0;
+#endif
+\treturn -ECANCELED;
+}
+module_init(trigger_init);
+
+MODULE_DESCRIPTION("Serial monitor test trigger");
+MODULE_LICENSE("GPL");
+"""
+
+trigger_modules = {
+    "smt_warn": "TRIGGER_WARN",
+    "smt_bug": "TRIGGER_BUG",
+    "smt_null": "TRIGGER_NULL",
+}
+
+
+@pytest.mark.require_plugin("serial_monitor")
+def test_serial_monitor_kernel_warning():
+    """
+    title: Serial monitor detects kernel warning.
+    description: |
+      Trigger WARN() in kernel module on DUT and check if serial monitor catches it
+      on the serial console without interrupting the test.
+    pass_criteria:
+      - Kernel warning is reported by serial monitor
+      - Test is not interrupted
+    """
+    serial_monitor = TestRun.plugin_manager.get_plugin("serial_monitor")
+
+    with TestRun.step("Build trigger modules"):
+        modules_dir = build_trigger_modules()
+
+    with TestRun.step("Trigger kernel warning"):
+        with serial_monitor.expect("warning", r"WARNING: .*smt_warn"):
+            output = TestRun.executor.run(f"insmod {modules_dir}/smt_warn.ko")
+            if output.exit_code == 0:
+                TestRun.fail("Trigger module loaded, warning was not triggered")
+
+    with TestRun.step("Check if DUT is still responsive"):
+        TestRun.executor.run_expect_success("true")
+
+
+@pytest.mark.require_plugin("serial_monitor")
+@pytest.mark.require_plugin("power_control")
+@pytest.mark.parametrize(
+    "module, pattern",
+    [
+        ("smt_bug", r"kernel BUG at .*smt_bug\.c"),
+        ("smt_null", r"BUG: kernel NULL pointer dereference"),
+    ],
+    ids=["bug", "null_deref"],
+)
+def test_serial_monitor_kernel_bug(module, pattern):
+    """
+    title: Serial monitor interrupts test on kernel bug.
+    description: |
+      Trigger kernel bug on DUT while test waits for a long running command and check
+      if serial monitor interrupts the test.
+    pass_criteria:
+      - Kernel bug is reported by serial monitor
+      - Test is interrupted before the command finishes
+    """
+    serial_monitor = TestRun.plugin_manager.get_plugin("serial_monitor")
+    power_control = TestRun.plugin_manager.get_plugin("power_control")
+    command_duration = 120
+    trigger_delay = 5
+
+    with TestRun.step("Power cycle DUT if trigger module from previous run is stuck"):
+        # Module that died in init stays in loading state and blocks loading it again
+        if check_if_directory_exists(f"/sys/module/{module}"):
+            power_control.power_cycle(wait_for_connection=True)
+
+    with TestRun.step("Build trigger modules"):
+        modules_dir = build_trigger_modules()
+
+    with TestRun.step("Trigger kernel bug while waiting for long running command"):
+        TestRun.executor.run_in_background(
+            f"sleep {trigger_delay} && insmod {modules_dir}/{module}.ko"
+        )
+        start = time.monotonic()
+        with serial_monitor.expect("bug", pattern):
+            try:
+                TestRun.executor.run(f"sleep {command_duration}")
+            except pytest.fail.Exception as e:
+                TestRun.LOGGER.info(f"Test interrupted: {e}")
+            else:
+                TestRun.fail("Test was not interrupted by kernel bug")
+        elapsed = time.monotonic() - start
+        TestRun.LOGGER.info(f"Interrupted after {elapsed:.1f}s")
+        if elapsed > command_duration / 2:
+            TestRun.fail(f"Test interrupted too late ({elapsed:.1f}s)")
+
+    with TestRun.step("Power cycle DUT to recover from kernel bug"):
+        power_control.power_cycle(wait_for_connection=True)
+
+
+def build_trigger_modules():
+    modules_dir = posixpath.join(TestRun.TEST_RUN_DATA_PATH, "serial_monitor_trigger")
+    create_directory(modules_dir, parents=True)
+    write_file(posixpath.join(modules_dir, "trigger.c"), trigger_source)
+    kbuild = f"obj-m := {' '.join(f'{name}.o' for name in trigger_modules)}\n"
+    for name, define in trigger_modules.items():
+        TestRun.executor.run_expect_success(
+            f"cp {modules_dir}/trigger.c {modules_dir}/{name}.c"
+        )
+        kbuild += f"CFLAGS_{name}.o := -D{define}\n"
+    write_file(posixpath.join(modules_dir, "Kbuild"), kbuild)
+    TestRun.executor.run_expect_success(
+        f"make -C /lib/modules/$(uname -r)/build M={modules_dir} modules"
+    )
+    return modules_dir