nixosTests.nixos-test-driver: init

nixos/lib/test-driver/.coveragerc

@@ -0,0 +1,5 @@
+[paths]
+source =
+  ./test_driver/
+  /build/test_driver/
+  **/site-packages/test_driver/

nixos/lib/test-driver/.gitignore

@@ -0,0 +1,3 @@
+.mutmut-cache
+.coverage
+coverage.xml

nixos/lib/test-driver/default.nix

@@ -46,4 +46,6 @@ python3Packages.buildPythonApplication {
     echo -e "\x1b[32m## run black\x1b[0m"
     black --check --diff .
   '';
+
+  passthru.tests = nixosTests.nixos-test-driver.driver;
 }

nixos/lib/test-driver/mutmut_config.py

@@ -0,0 +1,4 @@
+def pre_mutation(context):
+    line = context.current_source_line.strip()
+    if line.startswith("self.log") or line.startswith("raise") or ".nested" in line:
+        context.skip = True

nixos/lib/test-driver/run_mutmut.sh

@@ -0,0 +1,12 @@
+#! /usr/bin/env nix-shell
+#! nix-shell -I nixpkgs=../../.. -p black mutmut python3Packages.coverage -i bash
+
+set -euxo pipefail
+
+coverage combine --keep $(nix-build ../../.. -A nixosTests.nixos-test-driver)
+
+mutmut run \
+    --runner "nix build -f ../../.. nixosTests.nixos-test-driver" \
+    --paths-to-mutate "./test_driver/machine.py" \
+    --use-coverage \
+    --post-mutation "black test_driver -q"

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

@@ -3,9 +3,10 @@
 import signal
 import tempfile
 import threading
-from contextlib import contextmanager
+from collections.abc import Callable, Iterator
+from contextlib import AbstractContextManager, contextmanager
 from pathlib import Path
-from typing import Any, Callable, ContextManager, Dict, Iterator, List, Optional, Union
+from typing import Any
 
 from test_driver.logger import rootlog
 from test_driver.machine import Machine, NixStartScript, retry
@@ -35,21 +36,47 @@ def pythonize_name(name: str) -> str:
     return re.sub(r"^[^A-z_]|[^A-z0-9_]", "_", name)
 
 
+@contextmanager
+def must_raise(
+    regex: str,
+    exception: type[BaseException] = Exception,
+) -> Iterator[None]:
+    """Context manager that enforces that an exception is raised.
+    If no exception is raised, or the wrong kind of exception is raised,
+    or the exception string does not match the regex, an exception is raised.
+    """
+    short_desc = f"{exception.__name__} {regex!r}"
+    msg = f"Expected {short_desc}"
+
+    with rootlog.nested(f"Waiting for {short_desc}"):
+        try:
+            yield
+        except exception as e:
+            if re.search(regex, str(e)):
+                return
+            raise Exception(f"{msg}, but string {str(e)!r} did not match") from e
+        except Exception as e:
+            raise Exception(
+                f"{msg}, but {e!r} is not an instance of {exception!r}"
+            ) from e
+        raise Exception(f"{msg}, but no exception was raised")
+
+
 class Driver:
     """A handle to the driver that sets up the environment
     and runs the tests"""
 
     tests: str
-    vlans: List[VLan]
-    machines: List[Machine]
-    polling_conditions: List[PollingCondition]
+    vlans: list[VLan]
+    machines: list[Machine]
+    polling_conditions: list[PollingCondition]
     global_timeout: int
     race_timer: threading.Timer
 
     def __init__(
         self,
-        start_scripts: List[str],
-        vlans: List[int],
+        start_scripts: list[str],
+        vlans: list[int],
         tests: str,
         out_dir: Path,
         keep_vm_state: bool = False,
@@ -66,7 +93,7 @@ def __init__(
             vlans = list(set(vlans))
             self.vlans = [VLan(nr, tmp_dir) for nr in vlans]
 
-        def cmd(scripts: List[str]) -> Iterator[NixStartScript]:
+        def cmd(scripts: list[str]) -> Iterator[NixStartScript]:
             for s in scripts:
                 yield NixStartScript(s)
 
@@ -103,7 +130,7 @@ def subtest(self, name: str) -> Iterator[None]:
                 rootlog.error(f'Test "{name}" failed with error: "{e}"')
                 raise e
 
-    def test_symbols(self) -> Dict[str, Any]:
+    def test_symbols(self) -> dict[str, Any]:
         @contextmanager
         def subtest(name: str) -> Iterator[None]:
             return self.subtest(name)
@@ -125,6 +152,7 @@ def subtest(name: str) -> Iterator[None]:
             serial_stdout_on=self.serial_stdout_on,
             polling_condition=self.polling_condition,
             Machine=Machine,  # for typing
+            must_raise=must_raise,
         )
         machine_symbols = {pythonize_name(m.name): m for m in self.machines}
         # If there's exactly one machine, make it available under the name
@@ -187,7 +215,7 @@ def terminate_test(self) -> None:
             # to swallow them and prevent itself from terminating.
             os.kill(os.getpid(), signal.SIGTERM)
 
-    def create_machine(self, args: Dict[str, Any]) -> Machine:
+    def create_machine(self, args: dict[str, Any]) -> Machine:
         tmp_dir = get_tmp_dir()
 
         if args.get("startCommand"):
@@ -218,11 +246,11 @@ def check_polling_conditions(self) -> None:
 
     def polling_condition(
         self,
-        fun_: Optional[Callable] = None,
+        fun_: Callable | None = None,
         *,
         seconds_interval: float = 2.0,
-        description: Optional[str] = None,
-    ) -> Union[Callable[[Callable], ContextManager], ContextManager]:
+        description: str | None = None,
+    ) -> Callable[[Callable], AbstractContextManager] | AbstractContextManager:
         driver = self
 
         class Poll:

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

@@ -84,6 +84,8 @@
     ")": "shift-0x0B",
 }
 
+DEFAULT_TIMEOUT = 900
+
 
 def make_command(args: list) -> str:
     return " ".join(map(shlex.quote, (map(str, args))))
@@ -120,7 +122,7 @@ def _perform_ocr_on_screenshot(
     return model_results
 
 
-def retry(fn: Callable, timeout: int = 900) -> None:
+def retry(fn: Callable, timeout: int = DEFAULT_TIMEOUT) -> None:
     """Call the given function repeatedly, with 1 second intervals,
     until it returns True or a timeout is reached.
     """
@@ -547,7 +549,7 @@ def execute(
         command: str,
         check_return: bool = True,
         check_output: bool = True,
-        timeout: Optional[int] = 900,
+        timeout: Optional[int] = DEFAULT_TIMEOUT,
     ) -> Tuple[int, str]:
         """
         Execute a shell command, returning a list `(status, stdout)`.
@@ -693,7 +695,7 @@ def fail(self, *commands: str, timeout: Optional[int] = None) -> str:
                 output += out
         return output
 
-    def wait_until_succeeds(self, command: str, timeout: int = 900) -> str:
+    def wait_until_succeeds(self, command: str, timeout: int = DEFAULT_TIMEOUT) -> str:
         """
         Repeat a shell command with 1-second intervals until it succeeds.
         Has a default timeout of 900 seconds which can be modified, e.g.
@@ -712,7 +714,7 @@ def check_success(_: Any) -> bool:
             retry(check_success, timeout)
             return output
 
-    def wait_until_fails(self, command: str, timeout: int = 900) -> str:
+    def wait_until_fails(self, command: str, timeout: int = DEFAULT_TIMEOUT) -> str:
         """
         Like `wait_until_succeeds`, but repeating the command until it fails.
         """
@@ -747,7 +749,9 @@ def get_tty_text(self, tty: str) -> str:
         )
         return output
 
-    def wait_until_tty_matches(self, tty: str, regexp: str, timeout: int = 900) -> None:
+    def wait_until_tty_matches(
+        self, tty: str, regexp: str, timeout: int = DEFAULT_TIMEOUT
+    ) -> None:
         """Wait until the visible output on the chosen TTY matches regular
         expression. Throws an exception on timeout.
         """
@@ -775,7 +779,7 @@ def send_chars(self, chars: str, delay: Optional[float] = 0.01) -> None:
             for char in chars:
                 self.send_key(char, delay, log=False)
 
-    def wait_for_file(self, filename: str, timeout: int = 900) -> None:
+    def wait_for_file(self, filename: str, timeout: int = DEFAULT_TIMEOUT) -> None:
         """
         Waits until the file exists in the machine's file system.
         """
@@ -788,7 +792,7 @@ def check_file(_: Any) -> bool:
             retry(check_file, timeout)
 
     def wait_for_open_port(
-        self, port: int, addr: str = "localhost", timeout: int = 900
+        self, port: int, addr: str = "localhost", timeout: int = DEFAULT_TIMEOUT
     ) -> None:
         """
         Wait until a process is listening on the given TCP port and IP address
@@ -1014,7 +1018,7 @@ def get_screen_text(self) -> str:
         """
         return self._get_screen_text_variants([2])[0]
 
-    def wait_for_text(self, regex: str, timeout: int = 900) -> None:
+    def wait_for_text(self, regex: str, timeout: int = DEFAULT_TIMEOUT) -> None:
         """
         Wait until the supplied regular expressions matches the textual
         contents of the screen by using optical character recognition (see
@@ -1214,7 +1218,7 @@ def get_window_names(self) -> List[str]:
             r"xwininfo -root -tree | sed 's/.*0x[0-9a-f]* \"\([^\"]*\)\".*/\1/; t; d'"
         ).splitlines()
 
-    def wait_for_window(self, regexp: str, timeout: int = 900) -> None:
+    def wait_for_window(self, regexp: str, timeout: int = DEFAULT_TIMEOUT) -> None:
         """
         Wait until an X11 window has appeared whose name matches the given
         regular expression, e.g., `wait_for_window("Terminal")`.

nixos/lib/test-script-prepend.py

@@ -8,6 +8,7 @@
 from typing import Callable, Iterator, ContextManager, Optional, List, Dict, Any, Union
 from typing_extensions import Protocol
 from pathlib import Path
+import os
 
 
 class RetryProtocol(Protocol):
@@ -25,6 +26,13 @@ def __call__(
     ) -> Union[Callable[[Callable], ContextManager], ContextManager]:
         raise Exception("This is just type information for the Nix test driver")
 
+class MustRaiseProtocol(Protocol):
+    def __call__(
+        self,
+        message: str,
+        exception: type[BaseException] = Exception
+    ) -> ContextManager[None]:
+        raise Exception("This is just type information for the Nix test driver")
 
 start_all: Callable[[], None]
 subtest: Callable[[str], ContextManager[None]]
@@ -39,4 +47,5 @@ def __call__(
 join_all: Callable[[], None]
 serial_stdout_off: Callable[[], None]
 serial_stdout_on: Callable[[], None]
+must_raise: MustRaiseProtocol
 polling_condition: PollingConditionProtocol

nixos/lib/testing-python.nix

@@ -49,6 +49,7 @@ rec {
     , skipLint ? false
     , passthru ? {}
     , meta ? {}
+    , makeTestDriver ? (x: x)
     , # For meta.position
       pos ? # position used in error messages and for meta.position
         (if meta.description or null != null

nixos/lib/testing/driver.nix

@@ -4,13 +4,13 @@ let
 
   # Reifies and correctly wraps the python test driver for
   # the respective qemu version and with or without ocr support
-  testDriver = hostPkgs.callPackage ../test-driver {
+  initialTestDriver = hostPkgs.callPackage ../test-driver {
     inherit (config) enableOCR extraPythonPackages;
     qemu_pkg = config.qemu.package;
     imagemagick_light = hostPkgs.imagemagick_light.override { inherit (hostPkgs) libtiff; };
     tesseract4 = hostPkgs.tesseract4.override { enableLanguages = [ "eng" ]; };
   };
-
+  testDriver = config.makeTestDriver initialTestDriver;
 
   vlans = map (m: (
     m.virtualisation.vlans ++
@@ -109,6 +109,13 @@ in
       defaultText = literalMD "set by the test framework";
     };
 
+    makeTestDriver = mkOption {
+      description = mdDoc "Function that returns the package containing the test driver that will perform the testing instrumentation";
+      type = types.functionTo types.package;
+      defaultText = literalMD "set by the test framework";
+      internal = true;
+    };
+
     hostPkgs = mkOption {
       description = mdDoc "Nixpkgs attrset used outside the nodes.";
       type = types.raw;
@@ -196,6 +203,7 @@ in
     };
 
     driver = withChecks driver;
+    makeTestDriver = lib.id;
 
     # make available on the test runner
     passthru.driver = config.driver;

nixos/tests/all-tests.nix

@@ -98,6 +98,7 @@ in {
       [[ 143 = $(cat $failed/testBuildFailure.exit) ]]
       touch $out
     '';
+    driver = handleTest ./nixos-test-driver {};
   };
 
   # NixOS vm tests and non-vm unit tests

nixos/tests/nixos-test-driver/common.py

@@ -0,0 +1,29 @@
+from subprocess import check_call
+from contextlib import contextmanager
+from pathlib import Path
+
+exampleDir = Path("/tmp/exampleDir")
+exampleDir.mkdir(parents=True, exist_ok=True)
+
+exampleFile = exampleDir / "exampleFile"
+exampleFile.write_bytes(bytes(range(256)))
+assert exampleFile.exists()
+
+
+def host_files_same(path1, path2):
+    return check_call(["diff", path1, path2])
+
+
+@contextmanager
+def no_sleep():
+    import time
+
+    old_sleep = time.sleep
+    time.sleep = lambda x: None
+    yield
+    time.sleep = old_sleep
+
+
+out = Path(os.environ.get("out", ".")).resolve()
+
+start_all()

nixos/tests/nixos-test-driver/coverage-driver.nix

@@ -0,0 +1,39 @@
+pkgs: driver:
+let
+  coverage-rc = pkgs.writeText "coverage.rc" ''
+    [run]
+    include =
+        **/site-packages/test_driver/*
+  '';
+
+  coverage = "${pkgs.python3Packages.coverage}/bin/coverage";
+
+  pythonButItsCoverage = pkgs.writeScript "python-coverage" ''
+    #!${pkgs.runtimeShell}
+    set -euxo pipefail
+    COVERAGE_RCFILE=${coverage-rc} ${coverage} run -- "$@"
+    RETURN_CODE=$?
+
+    if whoami | grep "nixbld"; then
+      cp .coverage $out
+    fi
+
+    exit $RETURN_CODE
+  '';
+
+  coverageDriver = pkgs.runCommand "test-driver-coverage" { } ''
+    mkdir -p $out/bin
+    cat \
+      <(echo "#!${pythonButItsCoverage}") \
+      ${driver}/bin/.nixos-test-driver-wrapped \
+      > $out/bin/.nixos-test-driver-wrapped
+    sed "s@${driver}@$out@" ${driver}/bin/nixos-test-driver > $out/bin/nixos-test-driver
+    chmod +x $out/bin/nixos-test-driver
+    chmod +x $out/bin/.nixos-test-driver-wrapped
+    # exit 1
+  '';
+in
+pkgs.symlinkJoin {
+  name = "${driver.name}-coverage";
+  paths = [ coverageDriver driver ];
+}