commit 6d521ceac9fe37078822cca71c75770b9cf78ac7 · pyrox.dev/nixpkgs

+16

nixos/lib/test-driver/src/test_driver/machine/__init__.py

···

       241
       241
        
           # Store last serial console lines for use

     

       242
       242
        
           # of wait_for_console_text

     

       243
       243
        
           last_lines: Queue = Queue()

     

       244
       244
       +
           # Store all console output for full log retrieval

     

       245
       245
       +
           full_console_log: list[str]

     

       244
       246
        
           callbacks: list[Callable]

     

       245
       247
        
       

     

       246
       248
        
           def __repr__(self) -> str:

     
···

       263
       265
        
               self.start_command = start_command

     

       264
       266
        
               self.callbacks = callbacks if callbacks is not None else []

     

       265
       267
        
               self.logger = logger

     

       268
       268
       +
               self.full_console_log = []

     

       266
       269
        
       

     

       267
       270
        
               # set up directories

     

       268
       271
        
               self.shared_dir = self.tmp_dir / "shared-xchg"

     
···

       996
       999
        
               Wait until the supplied regular expressions match a line of the

     

       997
       1000
        
               serial console output.

     

       998
       1001
        
               This method is useful when OCR is not possible or inaccurate.

     

       1002
       1002
       +
       

     

       1003
       1003
       +
               When this method returns, the console output that includes the match has already become part of get_console_log().

     

       999
       1004
        
               """

     

       1000
       1005
        
               # Buffer the console output, this is needed

     

       1001
       1006
        
               # to match multiline regexes.

     
···

       1019
       1024
        
                   else:

     

       1020
       1025
        
                       while not console_matches(False):

     

       1021
       1026
        
                           pass

     

       1027
       1027
       +
       

     

       1028
       1028
       +
           def get_console_log(self) -> str:

     

       1029
       1029
       +
               """

     

       1030
       1030
       +
               Get the full console output from the machine since boot.

     

       1031
       1031
       +
               Returns all serial console output as a single string.

     

       1032
       1032
       +
               """

     

       1033
       1033
       +
               return "\n".join(self.full_console_log)

     

       1022
       1034
        
       

     

       1023
       1035
        
           def send_key(

     

       1024
       1036
        
               self, key: str, delay: float | None = 0.01, log: bool | None = True

     
···

       1086
       1098
        
               # Store last serial console lines for use

     

       1087
       1099
        
               # of wait_for_console_text

     

       1088
       1100
        
               self.last_lines: Queue = Queue()

     

       1101
       1101
       +
               # Re-initialize (if this is not the first start)

     

       1102
       1102
       +
               self.full_console_log: list[str] = []

     

       1089
       1103
        
       

     

       1090
       1104
        
               def process_serial_output() -> None:

     

       1091
       1105
        
                   assert self.process

     
···

       1093
       1107
        
                   for _line in self.process.stdout:

     

       1094
       1108
        
                       # Ignore undecodable bytes that may occur in boot menus

     

       1095
       1109
        
                       line = _line.decode(errors="ignore").replace("\r", "").rstrip()

     

       1110
       1110
       +
                       self.full_console_log.append(line)

     

       1111
       1111
       +
                       # Put on queue after adding to full_console_log to guarantee ordering

     

       1096
       1112
        
                       self.last_lines.put(line)

     

       1097
       1113
        
                       self.log_serial(line)

     

       1098
       1114

nixos/tests/all-tests.nix

···

       162
       162
        
           lib-extend = handleTestOn [ "x86_64-linux" "aarch64-linux" ] ./nixos-test-driver/lib-extend.nix { };

     

       163
       163
        
           node-name = runTest ./nixos-test-driver/node-name.nix;

     

       164
       164
        
           busybox = runTest ./nixos-test-driver/busybox.nix;

     

       165
       165
       +
           console-log = runTest ./nixos-test-driver/console-log.nix;

     

       165
       166
        
           driver-timeout =

     

       166
       167
        
             pkgs.runCommand "ensure-timeout-induced-failure"

     

       167
       168
        
               {

+117

nixos/tests/nixos-test-driver/console-log.nix

···

       1
       1
       +
       # Run with: nix-build -A nixosTests.nixos-test-driver.console-log

     

       2
       2
       +
       {

     

       3
       3
       +
         name = "nixos-test-driver.console-log";

     

       4
       4
       +
       

     

       5
       5
       +
         nodes = {

     

       6
       6
       +
           machine = {

     

       7
       7
       +
             # Configure the machine to print some distinctive messages to console

     

       8
       8
       +
             boot.kernelParams = [ "console=ttyS0,115200n8" ];

     

       9
       9
       +
       

     

       10
       10
       +
             # TODO: add a system service that logs to /dev/console, and test that

     

       11
       11
       +
             #       blocked on https://github.com/NixOS/nixpkgs/issues/442382

     

       12
       12
       +
           };

     

       13
       13
       +
         };

     

       14
       14
       +
       

     

       15
       15
       +
         testScript = ''

     

       16
       16
       +
           # Start the machine and wait for it to boot

     

       17
       17
       +
           machine.start()

     

       18
       18
       +
           machine.wait_for_unit("multi-user.target")

     

       19
       19
       +
       

     

       20
       20
       +
           with subtest("get_console_log returns console output"):

     

       21
       21
       +
             # Get the console log

     

       22
       22
       +
             console_log = machine.get_console_log()

     

       23
       23
       +
             # Verify it's a non-empty string

     

       24
       24
       +
             assert isinstance(console_log, str), f"Expected string, got {type(console_log)}"

     

       25
       25
       +
             assert len(console_log) > 0, "Console log should not be empty"

     

       26
       26
       +
       

     

       27
       27
       +
             print(f"Console log length: {len(console_log)} characters")

     

       28
       28
       +
             print("Console log preview (first 500 chars):")

     

       29
       29
       +
             print(console_log[:500])

     

       30
       30
       +
       

     

       31
       31
       +
           with subtest("console log contains boot messages"):

     

       32
       32
       +
             # Check for typical boot messages

     

       33
       33
       +
             assert "systemd" in console_log, "Console log should contain systemd messages"

     

       34
       34
       +
       

     

       35
       35
       +
             # Test direct stderr capture

     

       36
       36
       +
             machine.succeed("echo 'DIRECT_TEST_MESSAGE_12345' >&2")

     

       37
       37
       +
             import time

     

       38
       38
       +
             time.sleep(1)

     

       39
       39
       +
       

     

       40
       40
       +
             updated_log = machine.get_console_log()

     

       41
       41
       +
             assert "DIRECT_TEST_MESSAGE_12345" in updated_log, "Console log should capture stderr messages"

     

       42
       42
       +
             print("Successfully captured stderr message in console log")

     

       43
       43
       +
       

     

       44
       44
       +
           with subtest("console log captures command output"):

     

       45
       45
       +
             # Execute a command that prints to console via kernel messages

     

       46
       46
       +
             machine.succeed("echo 'KERNEL_TEST_MESSAGE_12345' > /dev/kmsg")

     

       47
       47
       +
       

     

       48
       48
       +
             # Wait a moment for the message to be captured

     

       49
       49
       +
             import time

     

       50
       50
       +
             time.sleep(2)

     

       51
       51
       +
       

     

       52
       52
       +
             # Get updated console log

     

       53
       53
       +
             updated_log = machine.get_console_log()

     

       54
       54
       +
             if "KERNEL_TEST_MESSAGE_12345" in updated_log:

     

       55
       55
       +
               print("Successfully captured kernel message in console log")

     

       56
       56
       +
             else:

     

       57
       57
       +
               print("Kernel message not found in console log - this may be expected depending on log level")

     

       58
       58
       +
       

     

       59
       59
       +
           with subtest("console log persists across multiple calls"):

     

       60
       60
       +
             # Get console log twice and verify they're consistent

     

       61
       61
       +
             log1 = machine.get_console_log()

     

       62
       62
       +
             log2 = machine.get_console_log()

     

       63
       63
       +
       

     

       64
       64
       +
             # Both logs should contain the same content (log2 might have additional content)

     

       65
       65
       +
             assert "systemd" in log1, "First log should contain systemd messages"

     

       66
       66
       +
             assert "systemd" in log2, "Second log should contain systemd messages"

     

       67
       67
       +
             assert len(log2) >= len(log1), "Second log should be at least as long as first log"

     

       68
       68
       +
             assert "KERNEL_TEST_MESSAGE_12345" in log2, "Second log should contain our kernel test message"

     

       69
       69
       +
       

     

       70
       70
       +
           with subtest("console log contains kernel messages"):

     

       71
       71
       +
             # Look for typical kernel boot messages

     

       72
       72
       +
             kernel_indicators = ["Linux version", "Command line:", "Kernel command line"]

     

       73
       73
       +
             kernel_found = any(indicator in console_log for indicator in kernel_indicators)

     

       74
       74
       +
             assert kernel_found, f"Console log should contain kernel messages. Indicators checked: {kernel_indicators}"

     

       75
       75
       +
       

     

       76
       76
       +
           with subtest("console log is cleared on machine restart"):

     

       77
       77
       +
             # Get the current console log and verify it contains our test messages

     

       78
       78
       +
             pre_shutdown_log = machine.get_console_log()

     

       79
       79
       +
             assert "KERNEL_TEST_MESSAGE_12345" in pre_shutdown_log, "Pre-shutdown log should contain our kernel test message"

     

       80
       80
       +
             assert "DIRECT_TEST_MESSAGE_12345" in pre_shutdown_log, "Pre-shutdown log should contain our stderr test message"

     

       81
       81
       +
             pre_shutdown_length = len(pre_shutdown_log)

     

       82
       82
       +
             print(f"Pre-shutdown console log length: {pre_shutdown_length} characters")

     

       83
       83
       +
       

     

       84
       84
       +
             # Shutdown the machine

     

       85
       85
       +
             machine.shutdown()

     

       86
       86
       +
       

     

       87
       87
       +
             # Start the machine again

     

       88
       88
       +
             machine.start()

     

       89
       89
       +
             machine.wait_for_unit("multi-user.target")

     

       90
       90
       +
       

     

       91
       91
       +
             # Get the console log after restart

     

       92
       92
       +
             post_restart_log = machine.get_console_log()

     

       93
       93
       +
             post_restart_length = len(post_restart_log)

     

       94
       94
       +
             print(f"Post-restart console log length: {post_restart_length} characters")

     

       95
       95
       +
       

     

       96
       96
       +
             # Verify the old messages are gone (log should be fresh)

     

       97
       97
       +
             assert "KERNEL_TEST_MESSAGE_12345" not in post_restart_log, "Post-restart log should not contain old kernel test message"

     

       98
       98
       +
             assert "DIRECT_TEST_MESSAGE_12345" not in post_restart_log, "Post-restart log should not contain old stderr test message"

     

       99
       99
       +
       

     

       100
       100
       +
             # Verify we still have boot messages (log should contain new boot sequence)

     

       101
       101
       +
             assert "systemd" in post_restart_log, "Post-restart log should contain systemd messages from new boot"

     

       102
       102
       +
             kernel_found_restart = any(indicator in post_restart_log for indicator in kernel_indicators)

     

       103
       103
       +
             assert kernel_found_restart, "Post-restart log should contain kernel messages from new boot"

     

       104
       104
       +
       

     

       105
       105
       +
             # Add a new test message to verify the log is working after restart

     

       106
       106
       +
             machine.succeed("echo 'POST_RESTART_TEST_MESSAGE_67890' > /dev/kmsg")

     

       107
       107
       +
             import time

     

       108
       108
       +
             time.sleep(1)

     

       109
       109
       +
       

     

       110
       110
       +
             final_log = machine.get_console_log()

     

       111
       111
       +
             assert "POST_RESTART_TEST_MESSAGE_67890" in final_log, "Should capture new messages after restart"

     

       112
       112
       +
       

     

       113
       113
       +
             print("Console log successfully cleared on restart and is functional again")

     

       114
       114
       +
       

     

       115
       115
       +
           print("All console log tests passed successfully!")

     

       116
       116
       +
         '';

     

       117
       117
       +
       }