commit d0c304d4c1a38afea52ad9e2a79c4fe6daddf551 · pyrox.dev/nixpkgs

+3 -2

nixos/doc/manual/development/writing-nixos-tests.section.md

···

       121
        
       ```py

     

       122
        
       machine.start()

     

       123
        
       machine.wait_for_unit("default.target")

     

       124
       -
       if not "Linux" in machine.succeed("uname"):

     

       125
       -
         raise Exception("Wrong OS")

     

       126
        
       ```

     

       127
        
       

     

       128
        
       The first line is technically unnecessary; machines are implicitly started

     
···

       133
        
       ```py

     

       134
        
       start_all()

     

       135
        
       ```

     

       0
        
       
     

       0
        
       
     

       136
        
       

     

       137
        
       If the hostname of a node contains characters that can't be used in a

     

       138
        
       Python variable name, those characters will be replaced with

···

       121
        
       ```py

     

       122
        
       machine.start()

     

       123
        
       machine.wait_for_unit("default.target")

     

       124
       +
       t.assertIn("Linux", machine.succeed("uname"), "Wrong OS")

     

       0
        
       
     

       125
        
       ```

     

       126
        
       

     

       127
        
       The first line is technically unnecessary; machines are implicitly started

     
···

       132
        
       ```py

     

       133
        
       start_all()

     

       134
        
       ```

     

       135
       +
       

     

       136
       +
       Under the variable `t`, all assertions from [`unittest.TestCase`](https://docs.python.org/3/library/unittest.html) are available.

     

       137
        
       

     

       138
        
       If the hostname of a node contains characters that can't be used in a

     

       139
        
       Python variable name, those characters will be replaced with

nixos/lib/test-driver/default.nix

···

       31
        
             colorama

     

       32
        
             junit-xml

     

       33
        
             ptpython

     

       0
        
       
     

       34
        
           ]

     

       35
        
           ++ extraPythonPackages python3Packages;

     

       36

···

       31
        
             colorama

     

       32
        
             junit-xml

     

       33
        
             ptpython

     

       34
       +
             ipython

     

       35
        
           ]

     

       36
        
           ++ extraPythonPackages python3Packages;

     

       37

+1 -1

nixos/lib/test-driver/src/pyproject.toml

···

       21
        
       line-length = 88

     

       22
        
       

     

       23
        
       lint.select = ["E", "F", "I", "U", "N"]

     

       24
       -
       lint.ignore = ["E501"]

     

       25
        
       

     

       26
        
       # xxx: we can import https://pypi.org/project/types-colorama/ here

     

       27
        
       [[tool.mypy.overrides]]

···

       21
        
       line-length = 88

     

       22
        
       

     

       23
        
       lint.select = ["E", "F", "I", "U", "N"]

     

       24
       +
       lint.ignore = ["E501", "N818"]

     

       25
        
       

     

       26
        
       # xxx: we can import https://pypi.org/project/types-colorama/ here

     

       27
        
       [[tool.mypy.overrides]]

+4 -5

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

···

       3
        
       import time

     

       4
        
       from pathlib import Path

     

       5
        
       

     

       6
       -
       import ptpython.repl

     

       7
        
       

     

       8
        
       from test_driver.driver import Driver

     

       9
        
       from test_driver.logger import (

     
···

       136
        
               if args.interactive:

     

       137
        
                   history_dir = os.getcwd()

     

       138
        
                   history_path = os.path.join(history_dir, ".nixos-test-history")

     

       139
       -
                   ptpython.repl.embed(

     

       140
       -
                       driver.test_symbols(),

     

       141
       -
                       {},

     

       142
        
                       history_filename=history_path,

     

       143
       -
                   )

     

       144
        
               else:

     

       145
        
                   tic = time.time()

     

       146
        
                   driver.run_tests()

···

       3
        
       import time

     

       4
        
       from pathlib import Path

     

       5
        
       

     

       6
       +
       import ptpython.ipython

     

       7
        
       

     

       8
        
       from test_driver.driver import Driver

     

       9
        
       from test_driver.logger import (

     
···

       136
        
               if args.interactive:

     

       137
        
                   history_dir = os.getcwd()

     

       138
        
                   history_path = os.path.join(history_dir, ".nixos-test-history")

     

       139
       +
                   ptpython.ipython.embed(

     

       140
       +
                       user_ns=driver.test_symbols(),

     

       0
        
       
     

       141
        
                       history_filename=history_path,

     

       142
       +
                   )  # type:ignore

     

       143
        
               else:

     

       144
        
                   tic = time.time()

     

       145
        
                   driver.run_tests()

+48 -2

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

···

       1
        
       import os

     

       2
        
       import re

     

       3
        
       import signal

     

       0
        
       
     

       4
        
       import tempfile

     

       5
        
       import threading

     

       0
        
       
     

       6
        
       from collections.abc import Callable, Iterator

     

       7
        
       from contextlib import AbstractContextManager, contextmanager

     

       8
        
       from pathlib import Path

     

       9
        
       from typing import Any

     

       0
        
       
     

       10
        
       

     

       0
        
       
     

       11
        
       from test_driver.logger import AbstractLogger

     

       12
        
       from test_driver.machine import Machine, NixStartScript, retry

     

       13
        
       from test_driver.polling_condition import PollingCondition

     

       14
        
       from test_driver.vlan import VLan

     

       15
        
       

     

       16
        
       SENTINEL = object()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       17
        
       

     

       18
        
       

     

       19
        
       def get_tmp_dir() -> Path:

     
···

       115
        
                   try:

     

       116
        
                       yield

     

       117
        
                   except Exception as e:

     

       118
       -
                       self.logger.error(f'Test "{name}" failed with error: "{e}"')

     

       119
        
                       raise e

     

       120
        
       

     

       121
        
           def test_symbols(self) -> dict[str, Any]:

     
···

       140
        
                   serial_stdout_on=self.serial_stdout_on,

     

       141
        
                   polling_condition=self.polling_condition,

     

       142
        
                   Machine=Machine,  # for typing

     

       0
        
       
     

       143
        
               )

     

       144
        
               machine_symbols = {pythonize_name(m.name): m for m in self.machines}

     

       145
        
               # If there's exactly one machine, make it available under the name

     
···

       163
        
               """Run the test script"""

     

       164
        
               with self.logger.nested("run the VM test script"):

     

       165
        
                   symbols = self.test_symbols()  # call eagerly

     

       166
       -
                   exec(self.tests, symbols, None)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       167
        
       

     

       168
        
           def run_tests(self) -> None:

     

       169
        
               """Run the test script (for non-interactive test runs)"""

···

       1
        
       import os

     

       2
        
       import re

     

       3
        
       import signal

     

       4
       +
       import sys

     

       5
        
       import tempfile

     

       6
        
       import threading

     

       7
       +
       import traceback

     

       8
        
       from collections.abc import Callable, Iterator

     

       9
        
       from contextlib import AbstractContextManager, contextmanager

     

       10
        
       from pathlib import Path

     

       11
        
       from typing import Any

     

       12
       +
       from unittest import TestCase

     

       13
        
       

     

       14
       +
       from test_driver.errors import MachineError, RequestedAssertionFailed

     

       15
        
       from test_driver.logger import AbstractLogger

     

       16
        
       from test_driver.machine import Machine, NixStartScript, retry

     

       17
        
       from test_driver.polling_condition import PollingCondition

     

       18
        
       from test_driver.vlan import VLan

     

       19
        
       

     

       20
        
       SENTINEL = object()

     

       21
       +
       

     

       22
       +
       

     

       23
       +
       class AssertionTester(TestCase):

     

       24
       +
           """

     

       25
       +
           Subclass of `unittest.TestCase` which is used in the

     

       26
       +
           `testScript` to perform assertions.

     

       27
       +
       

     

       28
       +
           It throws a custom exception whose parent class

     

       29
       +
           gets special treatment in the logs.

     

       30
       +
           """

     

       31
       +
       

     

       32
       +
           failureException = RequestedAssertionFailed

     

       33
        
       

     

       34
        
       

     

       35
        
       def get_tmp_dir() -> Path:

     
···

       131
        
                   try:

     

       132
        
                       yield

     

       133
        
                   except Exception as e:

     

       134
       +
                       self.logger.log_test_error(f'Test "{name}" failed with error: "{e}"')

     

       135
        
                       raise e

     

       136
        
       

     

       137
        
           def test_symbols(self) -> dict[str, Any]:

     
···

       156
        
                   serial_stdout_on=self.serial_stdout_on,

     

       157
        
                   polling_condition=self.polling_condition,

     

       158
        
                   Machine=Machine,  # for typing

     

       159
       +
                   t=AssertionTester(),

     

       160
        
               )

     

       161
        
               machine_symbols = {pythonize_name(m.name): m for m in self.machines}

     

       162
        
               # If there's exactly one machine, make it available under the name

     
···

       180
        
               """Run the test script"""

     

       181
        
               with self.logger.nested("run the VM test script"):

     

       182
        
                   symbols = self.test_symbols()  # call eagerly

     

       183
       +
                   try:

     

       184
       +
                       exec(self.tests, symbols, None)

     

       185
       +
                   except MachineError:

     

       186
       +
                       for line in traceback.format_exc().splitlines():

     

       187
       +
                           self.logger.log_test_error(line)

     

       188
       +
                       sys.exit(1)

     

       189
       +
                   except RequestedAssertionFailed:

     

       190
       +
                       exc_type, exc, tb = sys.exc_info()

     

       191
       +
                       # We manually print the stack frames, keeping only the ones from the test script

     

       192
       +
                       # (note: because the script is not a real file, the frame filename is `<string>`)

     

       193
       +
                       filtered = [

     

       194
       +
                           frame

     

       195
       +
                           for frame in traceback.extract_tb(tb)

     

       196
       +
                           if frame.filename == "<string>"

     

       197
       +
                       ]

     

       198
       +
       

     

       199
       +
                       self.logger.log_test_error("Traceback (most recent call last):")

     

       200
       +
       

     

       201
       +
                       code = self.tests.splitlines()

     

       202
       +
                       for frame, line in zip(filtered, traceback.format_list(filtered)):

     

       203
       +
                           self.logger.log_test_error(line.rstrip())

     

       204
       +
                           if lineno := frame.lineno:

     

       205
       +
                               self.logger.log_test_error(f"    {code[lineno - 1].strip()}")

     

       206
       +
       

     

       207
       +
                       self.logger.log_test_error("")  # blank line for readability

     

       208
       +
                       exc_prefix = exc_type.__name__ if exc_type is not None else "Error"

     

       209
       +
                       for line in f"{exc_prefix}: {exc}".splitlines():

     

       210
       +
                           self.logger.log_test_error(line)

     

       211
       +
       

     

       212
       +
                       sys.exit(1)

     

       213
        
       

     

       214
        
           def run_tests(self) -> None:

     

       215
        
               """Run the test script (for non-interactive test runs)"""

+20

nixos/lib/test-driver/src/test_driver/errors.py

···

       1
       +
       class MachineError(Exception):

     

       2
       +
           """

     

       3
       +
           Exception that indicates an error that is NOT the user's fault,

     

       4
       +
           i.e. something went wrong without the test being necessarily invalid,

     

       5
       +
           such as failing OCR.

     

       6
       +
       

     

       7
       +
           To make it easier to spot, this exception (and its subclasses)

     

       8
       +
           get a `!!!` prefix in the log output.

     

       9
       +
           """

     

       10
       +
       

     

       11
       +
       

     

       12
       +
       class RequestedAssertionFailed(AssertionError):

     

       13
       +
           """

     

       14
       +
           Special assertion that gets thrown on an assertion error,

     

       15
       +
           e.g. a failing `t.assertEqual(...)` or `machine.succeed(...)`.

     

       16
       +
       

     

       17
       +
           This gets special treatment in error reporting: i.e. it gets

     

       18
       +
           `!!!` as prefix just as `MachineError`, but only stack frames coming

     

       19
       +
           from `testScript` will show up in logs.

     

       20
       +
           """

+20 -1

nixos/lib/test-driver/src/test_driver/logger.py

···

       45
        
               pass

     

       46
        
       

     

       47
        
           @abstractmethod

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       48
        
           def log_serial(self, message: str, machine: str) -> None:

     

       49
        
               pass

     

       50
        
       

     
···

       97
        
               self.tests[self.currentSubtest].stderr += args[0] + os.linesep

     

       98
        
               self.tests[self.currentSubtest].failure = True

     

       99
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       100
        
           def log_serial(self, message: str, machine: str) -> None:

     

       101
        
               if not self._print_serial_logs:

     

       102
        
                   return

     
···

       155
        
           def warning(self, *args, **kwargs) -> None:  # type: ignore

     

       156
        
               for logger in self.logger_list:

     

       157
        
                   logger.warning(*args, **kwargs)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       158
        
       

     

       159
        
           def error(self, *args, **kwargs) -> None:  # type: ignore

     

       160
        
               for logger in self.logger_list:

     
···

       202
        
               tic = time.time()

     

       203
        
               yield

     

       204
        
               toc = time.time()

     

       205
       -
               self.log(f"(finished: {message}, in {toc - tic:.2f} seconds)")

     

       206
        
       

     

       207
        
           def info(self, *args, **kwargs) -> None:  # type: ignore

     

       208
        
               self.log(*args, **kwargs)

     
···

       221
        
                   return

     

       222
        
       

     

       223
        
               self._eprint(Style.DIM + f"{machine} # {message}" + Style.RESET_ALL)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       224
        
       

     

       225
        
       

     

       226
        
       class XMLLogger(AbstractLogger):

     
···

       259
        
               self.log(*args, **kwargs)

     

       260
        
       

     

       261
        
           def error(self, *args, **kwargs) -> None:  # type: ignore

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       262
        
               self.log(*args, **kwargs)

     

       263
        
       

     

       264
        
           def log(self, message: str, attributes: dict[str, str] = {}) -> None:

···

       45
        
               pass

     

       46
        
       

     

       47
        
           @abstractmethod

     

       48
       +
           def log_test_error(self, *args, **kwargs) -> None:  # type:ignore

     

       49
       +
               pass

     

       50
       +
       

     

       51
       +
           @abstractmethod

     

       52
        
           def log_serial(self, message: str, machine: str) -> None:

     

       53
        
               pass

     

       54
        
       

     
···

       101
        
               self.tests[self.currentSubtest].stderr += args[0] + os.linesep

     

       102
        
               self.tests[self.currentSubtest].failure = True

     

       103
        
       

     

       104
       +
           def log_test_error(self, *args, **kwargs) -> None:  # type: ignore

     

       105
       +
               self.error(*args, **kwargs)

     

       106
       +
       

     

       107
        
           def log_serial(self, message: str, machine: str) -> None:

     

       108
        
               if not self._print_serial_logs:

     

       109
        
                   return

     
···

       162
        
           def warning(self, *args, **kwargs) -> None:  # type: ignore

     

       163
        
               for logger in self.logger_list:

     

       164
        
                   logger.warning(*args, **kwargs)

     

       165
       +
       

     

       166
       +
           def log_test_error(self, *args, **kwargs) -> None:  # type: ignore

     

       167
       +
               for logger in self.logger_list:

     

       168
       +
                   logger.log_test_error(*args, **kwargs)

     

       169
        
       

     

       170
        
           def error(self, *args, **kwargs) -> None:  # type: ignore

     

       171
        
               for logger in self.logger_list:

     
···

       213
        
               tic = time.time()

     

       214
        
               yield

     

       215
        
               toc = time.time()

     

       216
       +
               self.log(f"(finished: {message}, in {toc - tic:.2f} seconds)", attributes)

     

       217
        
       

     

       218
        
           def info(self, *args, **kwargs) -> None:  # type: ignore

     

       219
        
               self.log(*args, **kwargs)

     
···

       232
        
                   return

     

       233
        
       

     

       234
        
               self._eprint(Style.DIM + f"{machine} # {message}" + Style.RESET_ALL)

     

       235
       +
       

     

       236
       +
           def log_test_error(self, *args, **kwargs) -> None:  # type: ignore

     

       237
       +
               prefix = Fore.RED + "!!! " + Style.RESET_ALL

     

       238
       +
               # NOTE: using `warning` instead of `error` to ensure it does not exit after printing the first log

     

       239
       +
               self.warning(f"{prefix}{args[0]}", *args[1:], **kwargs)

     

       240
        
       

     

       241
        
       

     

       242
        
       class XMLLogger(AbstractLogger):

     
···

       275
        
               self.log(*args, **kwargs)

     

       276
        
       

     

       277
        
           def error(self, *args, **kwargs) -> None:  # type: ignore

     

       278
       +
               self.log(*args, **kwargs)

     

       279
       +
       

     

       280
       +
           def log_test_error(self, *args, **kwargs) -> None:  # type: ignore

     

       281
        
               self.log(*args, **kwargs)

     

       282
        
       

     

       283
        
           def log(self, message: str, attributes: dict[str, str] = {}) -> None:

+19 -12

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

···

       19
        
       from queue import Queue

     

       20
        
       from typing import Any

     

       21
        
       

     

       0
        
       
     

       22
        
       from test_driver.logger import AbstractLogger

     

       23
        
       

     

       24
        
       from .qmp import QMPSession

     
···

       129
        
           )

     

       130
        
       

     

       131
        
           if ret.returncode != 0:

     

       132
       -
               raise Exception(

     

       133
        
                   f"Image processing failed with exit code {ret.returncode}, stdout: {ret.stdout.decode()}, stderr: {ret.stderr.decode()}"

     

       134
        
               )

     

       135
        
       

     
···

       140
        
           screenshot_path: str, model_ids: Iterable[int]

     

       141
        
       ) -> list[str]:

     

       142
        
           if shutil.which("tesseract") is None:

     

       143
       -
               raise Exception("OCR requested but enableOCR is false")

     

       144
        
       

     

       145
        
           processed_image = _preprocess_screenshot(screenshot_path, negate=False)

     

       146
        
           processed_negative = _preprocess_screenshot(screenshot_path, negate=True)

     
···

       163
        
                       capture_output=True,

     

       164
        
                   )

     

       165
        
                   if ret.returncode != 0:

     

       166
       -
                       raise Exception(f"OCR failed with exit code {ret.returncode}")

     

       167
        
                   model_results.append(ret.stdout.decode("utf-8"))

     

       168
        
       

     

       169
        
           return model_results

     
···

       180
        
               time.sleep(1)

     

       181
        
       

     

       182
        
           if not fn(True):

     

       183
       -
               raise Exception(f"action timed out after {timeout} seconds")

     

       0
        
       
     

       0
        
       
     

       184
        
       

     

       185
        
       

     

       186
        
       class StartCommand:

     
···

       409
        
               def check_active(_last_try: bool) -> bool:

     

       410
        
                   state = self.get_unit_property(unit, "ActiveState", user)

     

       411
        
                   if state == "failed":

     

       412
       -
                       raise Exception(f'unit "{unit}" reached state "{state}"')

     

       413
        
       

     

       414
        
                   if state == "inactive":

     

       415
        
                       status, jobs = self.systemctl("list-jobs --full 2>&1", user)

     

       416
        
                       if "No jobs" in jobs:

     

       417
        
                           info = self.get_unit_info(unit, user)

     

       418
        
                           if info["ActiveState"] == state:

     

       419
       -
                               raise Exception(

     

       420
        
                                   f'unit "{unit}" is inactive and there are no pending jobs'

     

       421
        
                               )

     

       422
        
       

     
···

       431
        
           def get_unit_info(self, unit: str, user: str | None = None) -> dict[str, str]:

     

       432
        
               status, lines = self.systemctl(f'--no-pager show "{unit}"', user)

     

       433
        
               if status != 0:

     

       434
       -
                   raise Exception(

     

       435
        
                       f'retrieving systemctl info for unit "{unit}"'

     

       436
        
                       + ("" if user is None else f' under user "{user}"')

     

       437
        
                       + f" failed with exit code {status}"

     
···

       461
        
                   user,

     

       462
        
               )

     

       463
        
               if status != 0:

     

       464
       -
                   raise Exception(

     

       465
        
                       f'retrieving systemctl property "{property}" for unit "{unit}"'

     

       466
        
                       + ("" if user is None else f' under user "{user}"')

     

       467
        
                       + f" failed with exit code {status}"

     
···

       509
        
                   info = self.get_unit_info(unit)

     

       510
        
                   state = info["ActiveState"]

     

       511
        
                   if state != require_state:

     

       512
       -
                       raise Exception(

     

       513
        
                           f"Expected unit '{unit}' to to be in state "

     

       514
        
                           f"'{require_state}' but it is in state '{state}'"

     

       515
        
                       )

     
···

       663
        
                       (status, out) = self.execute(command, timeout=timeout)

     

       664
        
                       if status != 0:

     

       665
        
                           self.log(f"output: {out}")

     

       666
       -
                           raise Exception(f"command `{command}` failed (exit code {status})")

     

       0
        
       
     

       0
        
       
     

       667
        
                       output += out

     

       668
        
               return output

     

       669
        
       

     
···

       677
        
                   with self.nested(f"must fail: {command}"):

     

       678
        
                       (status, out) = self.execute(command, timeout=timeout)

     

       679
        
                       if status == 0:

     

       680
       -
                           raise Exception(f"command `{command}` unexpectedly succeeded")

     

       0
        
       
     

       0
        
       
     

       681
        
                       output += out

     

       682
        
               return output

     

       683
        
       

     
···

       922
        
                   ret = subprocess.run(f"pnmtopng '{tmp}' > '{filename}'", shell=True)

     

       923
        
                   os.unlink(tmp)

     

       924
        
                   if ret.returncode != 0:

     

       925
       -
                       raise Exception("Cannot convert screenshot")

     

       926
        
       

     

       927
        
           def copy_from_host_via_shell(self, source: str, target: str) -> None:

     

       928
        
               """Copy a file from the host into the guest by piping it over the

···

       19
        
       from queue import Queue

     

       20
        
       from typing import Any

     

       21
        
       

     

       22
       +
       from test_driver.errors import MachineError, RequestedAssertionFailed

     

       23
        
       from test_driver.logger import AbstractLogger

     

       24
        
       

     

       25
        
       from .qmp import QMPSession

     
···

       130
        
           )

     

       131
        
       

     

       132
        
           if ret.returncode != 0:

     

       133
       +
               raise MachineError(

     

       134
        
                   f"Image processing failed with exit code {ret.returncode}, stdout: {ret.stdout.decode()}, stderr: {ret.stderr.decode()}"

     

       135
        
               )

     

       136
        
       

     
···

       141
        
           screenshot_path: str, model_ids: Iterable[int]

     

       142
        
       ) -> list[str]:

     

       143
        
           if shutil.which("tesseract") is None:

     

       144
       +
               raise MachineError("OCR requested but enableOCR is false")

     

       145
        
       

     

       146
        
           processed_image = _preprocess_screenshot(screenshot_path, negate=False)

     

       147
        
           processed_negative = _preprocess_screenshot(screenshot_path, negate=True)

     
···

       164
        
                       capture_output=True,

     

       165
        
                   )

     

       166
        
                   if ret.returncode != 0:

     

       167
       +
                       raise MachineError(f"OCR failed with exit code {ret.returncode}")

     

       168
        
                   model_results.append(ret.stdout.decode("utf-8"))

     

       169
        
       

     

       170
        
           return model_results

     
···

       181
        
               time.sleep(1)

     

       182
        
       

     

       183
        
           if not fn(True):

     

       184
       +
               raise RequestedAssertionFailed(

     

       185
       +
                   f"action timed out after {timeout} tries with one-second pause in-between"

     

       186
       +
               )

     

       187
        
       

     

       188
        
       

     

       189
        
       class StartCommand:

     
···

       412
        
               def check_active(_last_try: bool) -> bool:

     

       413
        
                   state = self.get_unit_property(unit, "ActiveState", user)

     

       414
        
                   if state == "failed":

     

       415
       +
                       raise RequestedAssertionFailed(f'unit "{unit}" reached state "{state}"')

     

       416
        
       

     

       417
        
                   if state == "inactive":

     

       418
        
                       status, jobs = self.systemctl("list-jobs --full 2>&1", user)

     

       419
        
                       if "No jobs" in jobs:

     

       420
        
                           info = self.get_unit_info(unit, user)

     

       421
        
                           if info["ActiveState"] == state:

     

       422
       +
                               raise RequestedAssertionFailed(

     

       423
        
                                   f'unit "{unit}" is inactive and there are no pending jobs'

     

       424
        
                               )

     

       425
        
       

     
···

       434
        
           def get_unit_info(self, unit: str, user: str | None = None) -> dict[str, str]:

     

       435
        
               status, lines = self.systemctl(f'--no-pager show "{unit}"', user)

     

       436
        
               if status != 0:

     

       437
       +
                   raise RequestedAssertionFailed(

     

       438
        
                       f'retrieving systemctl info for unit "{unit}"'

     

       439
        
                       + ("" if user is None else f' under user "{user}"')

     

       440
        
                       + f" failed with exit code {status}"

     
···

       464
        
                   user,

     

       465
        
               )

     

       466
        
               if status != 0:

     

       467
       +
                   raise RequestedAssertionFailed(

     

       468
        
                       f'retrieving systemctl property "{property}" for unit "{unit}"'

     

       469
        
                       + ("" if user is None else f' under user "{user}"')

     

       470
        
                       + f" failed with exit code {status}"

     
···

       512
        
                   info = self.get_unit_info(unit)

     

       513
        
                   state = info["ActiveState"]

     

       514
        
                   if state != require_state:

     

       515
       +
                       raise RequestedAssertionFailed(

     

       516
        
                           f"Expected unit '{unit}' to to be in state "

     

       517
        
                           f"'{require_state}' but it is in state '{state}'"

     

       518
        
                       )

     
···

       666
        
                       (status, out) = self.execute(command, timeout=timeout)

     

       667
        
                       if status != 0:

     

       668
        
                           self.log(f"output: {out}")

     

       669
       +
                           raise RequestedAssertionFailed(

     

       670
       +
                               f"command `{command}` failed (exit code {status})"

     

       671
       +
                           )

     

       672
        
                       output += out

     

       673
        
               return output

     

       674
        
       

     
···

       682
        
                   with self.nested(f"must fail: {command}"):

     

       683
        
                       (status, out) = self.execute(command, timeout=timeout)

     

       684
        
                       if status == 0:

     

       685
       +
                           raise RequestedAssertionFailed(

     

       686
       +
                               f"command `{command}` unexpectedly succeeded"

     

       687
       +
                           )

     

       688
        
                       output += out

     

       689
        
               return output

     

       690
        
       

     
···

       929
        
                   ret = subprocess.run(f"pnmtopng '{tmp}' > '{filename}'", shell=True)

     

       930
        
                   os.unlink(tmp)

     

       931
        
                   if ret.returncode != 0:

     

       932
       +
                       raise MachineError("Cannot convert screenshot")

     

       933
        
       

     

       934
        
           def copy_from_host_via_shell(self, source: str, target: str) -> None:

     

       935
        
               """Copy a file from the host into the guest by piping it over the

nixos/lib/test-script-prepend.py

···

       8
        
       from typing import Callable, Iterator, ContextManager, Optional, List, Dict, Any, Union

     

       9
        
       from typing_extensions import Protocol

     

       10
        
       from pathlib import Path

     

       0
        
       
     

       11
        
       

     

       12
        
       

     

       13
        
       class RetryProtocol(Protocol):

     
···

       51
        
       serial_stdout_off: Callable[[], None]

     

       52
        
       serial_stdout_on: Callable[[], None]

     

       53
        
       polling_condition: PollingConditionProtocol

     

       0

···

       8
        
       from typing import Callable, Iterator, ContextManager, Optional, List, Dict, Any, Union

     

       9
        
       from typing_extensions import Protocol

     

       10
        
       from pathlib import Path

     

       11
       +
       from unittest import TestCase

     

       12
        
       

     

       13
        
       

     

       14
        
       class RetryProtocol(Protocol):

     
···

       52
        
       serial_stdout_off: Callable[[], None]

     

       53
        
       serial_stdout_on: Callable[[], None]

     

       54
        
       polling_condition: PollingConditionProtocol

     

       55
       +
       t: TestCase