From 6d521ceac9fe37078822cca71c75770b9cf78ac7 Mon Sep 17 00:00:00 2001 From: Robert Hensing Date: Fri, 12 Sep 2025 16:21:43 +0200 Subject: [PATCH] nixos/test-driver: Add machine.get_console_log() --- .../src/test_driver/machine/__init__.py | 16 +++ nixos/tests/all-tests.nix | 1 + nixos/tests/nixos-test-driver/console-log.nix | 117 ++++++++++++++++++ 3 files changed, 134 insertions(+) create mode 100644 nixos/tests/nixos-test-driver/console-log.nix diff --git a/nixos/lib/test-driver/src/test_driver/machine/__init__.py b/nixos/lib/test-driver/src/test_driver/machine/__init__.py index 8115a5967a2d..6f0985479803 100644 --- a/nixos/lib/test-driver/src/test_driver/machine/__init__.py +++ b/nixos/lib/test-driver/src/test_driver/machine/__init__.py @@ -241,6 +241,8 @@ class Machine: # Store last serial console lines for use # of wait_for_console_text last_lines: Queue = Queue() + # Store all console output for full log retrieval + full_console_log: list[str] callbacks: list[Callable] def __repr__(self) -> str: @@ -263,6 +265,7 @@ class Machine: self.start_command = start_command self.callbacks = callbacks if callbacks is not None else [] self.logger = logger + self.full_console_log = [] # set up directories self.shared_dir = self.tmp_dir / "shared-xchg" @@ -996,6 +999,8 @@ class Machine: Wait until the supplied regular expressions match a line of the serial console output. This method is useful when OCR is not possible or inaccurate. + + When this method returns, the console output that includes the match has already become part of get_console_log(). """ # Buffer the console output, this is needed # to match multiline regexes. @@ -1020,6 +1025,13 @@ class Machine: while not console_matches(False): pass + def get_console_log(self) -> str: + """ + Get the full console output from the machine since boot. + Returns all serial console output as a single string. + """ + return "\n".join(self.full_console_log) + def send_key( self, key: str, delay: float | None = 0.01, log: bool | None = True ) -> None: @@ -1086,6 +1098,8 @@ class Machine: # Store last serial console lines for use # of wait_for_console_text self.last_lines: Queue = Queue() + # Re-initialize (if this is not the first start) + self.full_console_log: list[str] = [] def process_serial_output() -> None: assert self.process @@ -1093,6 +1107,8 @@ class Machine: for _line in self.process.stdout: # Ignore undecodable bytes that may occur in boot menus line = _line.decode(errors="ignore").replace("\r", "").rstrip() + self.full_console_log.append(line) + # Put on queue after adding to full_console_log to guarantee ordering self.last_lines.put(line) self.log_serial(line) diff --git a/nixos/tests/all-tests.nix b/nixos/tests/all-tests.nix index 46fa3160cb69..92f917ca51ef 100644 --- a/nixos/tests/all-tests.nix +++ b/nixos/tests/all-tests.nix @@ -162,6 +162,7 @@ in lib-extend = handleTestOn [ "x86_64-linux" "aarch64-linux" ] ./nixos-test-driver/lib-extend.nix { }; node-name = runTest ./nixos-test-driver/node-name.nix; busybox = runTest ./nixos-test-driver/busybox.nix; + console-log = runTest ./nixos-test-driver/console-log.nix; driver-timeout = pkgs.runCommand "ensure-timeout-induced-failure" { diff --git a/nixos/tests/nixos-test-driver/console-log.nix b/nixos/tests/nixos-test-driver/console-log.nix new file mode 100644 index 000000000000..5fc8a5e8ae78 --- /dev/null +++ b/nixos/tests/nixos-test-driver/console-log.nix @@ -0,0 +1,117 @@ +# Run with: nix-build -A nixosTests.nixos-test-driver.console-log +{ + name = "nixos-test-driver.console-log"; + + nodes = { + machine = { + # Configure the machine to print some distinctive messages to console + boot.kernelParams = [ "console=ttyS0,115200n8" ]; + + # TODO: add a system service that logs to /dev/console, and test that + # blocked on https://github.com/NixOS/nixpkgs/issues/442382 + }; + }; + + testScript = '' + # Start the machine and wait for it to boot + machine.start() + machine.wait_for_unit("multi-user.target") + + with subtest("get_console_log returns console output"): + # Get the console log + console_log = machine.get_console_log() + # Verify it's a non-empty string + assert isinstance(console_log, str), f"Expected string, got {type(console_log)}" + assert len(console_log) > 0, "Console log should not be empty" + + print(f"Console log length: {len(console_log)} characters") + print("Console log preview (first 500 chars):") + print(console_log[:500]) + + with subtest("console log contains boot messages"): + # Check for typical boot messages + assert "systemd" in console_log, "Console log should contain systemd messages" + + # Test direct stderr capture + machine.succeed("echo 'DIRECT_TEST_MESSAGE_12345' >&2") + import time + time.sleep(1) + + updated_log = machine.get_console_log() + assert "DIRECT_TEST_MESSAGE_12345" in updated_log, "Console log should capture stderr messages" + print("Successfully captured stderr message in console log") + + with subtest("console log captures command output"): + # Execute a command that prints to console via kernel messages + machine.succeed("echo 'KERNEL_TEST_MESSAGE_12345' > /dev/kmsg") + + # Wait a moment for the message to be captured + import time + time.sleep(2) + + # Get updated console log + updated_log = machine.get_console_log() + if "KERNEL_TEST_MESSAGE_12345" in updated_log: + print("Successfully captured kernel message in console log") + else: + print("Kernel message not found in console log - this may be expected depending on log level") + + with subtest("console log persists across multiple calls"): + # Get console log twice and verify they're consistent + log1 = machine.get_console_log() + log2 = machine.get_console_log() + + # Both logs should contain the same content (log2 might have additional content) + assert "systemd" in log1, "First log should contain systemd messages" + assert "systemd" in log2, "Second log should contain systemd messages" + assert len(log2) >= len(log1), "Second log should be at least as long as first log" + assert "KERNEL_TEST_MESSAGE_12345" in log2, "Second log should contain our kernel test message" + + with subtest("console log contains kernel messages"): + # Look for typical kernel boot messages + kernel_indicators = ["Linux version", "Command line:", "Kernel command line"] + kernel_found = any(indicator in console_log for indicator in kernel_indicators) + assert kernel_found, f"Console log should contain kernel messages. Indicators checked: {kernel_indicators}" + + with subtest("console log is cleared on machine restart"): + # Get the current console log and verify it contains our test messages + pre_shutdown_log = machine.get_console_log() + assert "KERNEL_TEST_MESSAGE_12345" in pre_shutdown_log, "Pre-shutdown log should contain our kernel test message" + assert "DIRECT_TEST_MESSAGE_12345" in pre_shutdown_log, "Pre-shutdown log should contain our stderr test message" + pre_shutdown_length = len(pre_shutdown_log) + print(f"Pre-shutdown console log length: {pre_shutdown_length} characters") + + # Shutdown the machine + machine.shutdown() + + # Start the machine again + machine.start() + machine.wait_for_unit("multi-user.target") + + # Get the console log after restart + post_restart_log = machine.get_console_log() + post_restart_length = len(post_restart_log) + print(f"Post-restart console log length: {post_restart_length} characters") + + # Verify the old messages are gone (log should be fresh) + assert "KERNEL_TEST_MESSAGE_12345" not in post_restart_log, "Post-restart log should not contain old kernel test message" + assert "DIRECT_TEST_MESSAGE_12345" not in post_restart_log, "Post-restart log should not contain old stderr test message" + + # Verify we still have boot messages (log should contain new boot sequence) + assert "systemd" in post_restart_log, "Post-restart log should contain systemd messages from new boot" + kernel_found_restart = any(indicator in post_restart_log for indicator in kernel_indicators) + assert kernel_found_restart, "Post-restart log should contain kernel messages from new boot" + + # Add a new test message to verify the log is working after restart + machine.succeed("echo 'POST_RESTART_TEST_MESSAGE_67890' > /dev/kmsg") + import time + time.sleep(1) + + final_log = machine.get_console_log() + assert "POST_RESTART_TEST_MESSAGE_67890" in final_log, "Should capture new messages after restart" + + print("Console log successfully cleared on restart and is functional again") + + print("All console log tests passed successfully!") + ''; +}