nixos/test-driver: add option to force kvm use (#509553)

nixos/doc/manual/redirects.json

@@ -2246,6 +2246,9 @@
   "test-opt-passthru": [
     "index.html#test-opt-passthru"
   ],
+  "test-opt-qemu.forceAccel": [
+    "index.html#test-opt-qemu.forceAccel"
+  ],
   "test-opt-qemu.package": [
     "index.html#test-opt-qemu.package"
   ],

nixos/lib/qemu-common.nix

@@ -24,30 +24,37 @@ rec {
     else
       throw "Unknown QEMU serial device for system '${stdenv.hostPlatform.system}'";
 
-  qemuBinary =
-    qemuPkg:
+  qemuBinary = qemuPkg: qemuBinaryWith { inherit qemuPkg; };
+
+  qemuBinaryWith =
+    {
+      qemuPkg,
+      forceAccel ? false,
+    }:
     let
       hostStdenv = qemuPkg.stdenv;
       hostSystem = hostStdenv.system;
       guestSystem = stdenv.hostPlatform.system;
 
+      accel = accelName: if forceAccel then accelName else "${accelName}:tcg";
+
       linuxHostGuestMatrix = {
-        x86_64-linux = "${qemuPkg}/bin/qemu-system-x86_64 -machine accel=kvm:tcg -cpu max";
-        armv7l-linux = "${qemuPkg}/bin/qemu-system-arm -machine virt,accel=kvm:tcg -cpu max";
-        aarch64-linux = "${qemuPkg}/bin/qemu-system-aarch64 -machine virt,gic-version=max,accel=kvm:tcg -cpu max";
+        x86_64-linux = "${qemuPkg}/bin/qemu-system-x86_64 -machine accel=${accel "kvm"} -cpu max";
+        armv7l-linux = "${qemuPkg}/bin/qemu-system-arm -machine virt,accel=${accel "kvm"} -cpu max";
+        aarch64-linux = "${qemuPkg}/bin/qemu-system-aarch64 -machine virt,gic-version=max,accel=${accel "kvm"} -cpu max";
         powerpc64le-linux = "${qemuPkg}/bin/qemu-system-ppc64 -machine powernv";
         powerpc64-linux = "${qemuPkg}/bin/qemu-system-ppc64 -machine powernv";
         riscv32-linux = "${qemuPkg}/bin/qemu-system-riscv32 -machine virt";
         riscv64-linux = "${qemuPkg}/bin/qemu-system-riscv64 -machine virt";
-        x86_64-darwin = "${qemuPkg}/bin/qemu-system-x86_64 -machine accel=kvm:tcg -cpu max";
+        x86_64-darwin = "${qemuPkg}/bin/qemu-system-x86_64 -machine accel=${accel "kvm"} -cpu max";
       };
       otherHostGuestMatrix = {
         aarch64-darwin = {
-          aarch64-linux = "${qemuPkg}/bin/qemu-system-aarch64 -machine virt,gic-version=2,accel=hvf:tcg -cpu max";
+          aarch64-linux = "${qemuPkg}/bin/qemu-system-aarch64 -machine virt,gic-version=2,accel=${accel "hvf"} -cpu max";
           inherit (otherHostGuestMatrix.x86_64-darwin) x86_64-linux;
         };
         x86_64-darwin = {
-          x86_64-linux = "${qemuPkg}/bin/qemu-system-x86_64 -machine type=q35,accel=hvf:tcg -cpu max";
+          x86_64-linux = "${qemuPkg}/bin/qemu-system-x86_64 -machine type=q35,accel=${accel "hvf"} -cpu max";
         };
       };
 

nixos/lib/test-driver/default.nix

@@ -7,11 +7,11 @@
   imagemagick_light,
   ipython,
   junit-xml,
-  mypy,
   ptpython,
   python,
-  ruff,
   remote-pdb,
+  ruff,
+  ty,
 
   netpbm,
   nixosTests,
@@ -72,13 +72,13 @@ buildPythonApplication {
   doCheck = true;
 
   nativeCheckInputs = [
-    mypy
     ruff
+    ty
   ];
 
   checkPhase = ''
-    echo -e "\x1b[32m## run mypy\x1b[0m"
-    mypy test_driver extract-docstrings.py
+    echo -e "\x1b[32m## run ty\x1b[0m"
+    ty check --error-on-warning test_driver extract-docstrings.py
     echo -e "\x1b[32m## run ruff check\x1b[0m"
     ruff check .
     echo -e "\x1b[32m## run ruff format\x1b[0m"

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

@@ -17,27 +17,8 @@ find = {}
 test_driver = ["py.typed"]
 
 [tool.ruff]
-target-version = "py312"
+target-version = "py313"
 line-length = 88
 
 lint.select = ["E", "F", "I", "U", "N"]
 lint.ignore = ["E501", "N818"]
-
-# xxx: we can import https://pypi.org/project/types-colorama/ here
-[[tool.mypy.overrides]]
-module = "colorama.*"
-ignore_missing_imports = true
-
-[[tool.mypy.overrides]]
-module = "ptpython.*"
-ignore_missing_imports = true
-
-[[tool.mypy.overrides]]
-module = "junit_xml.*"
-ignore_missing_imports = true
-
-[tool.mypy]
-warn_redundant_casts = true
-disallow_untyped_calls = true
-disallow_untyped_defs = true
-no_implicit_optional = true

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

@@ -23,7 +23,7 @@ class EnvDefault(argparse.Action):
     environment variable as the flags default value.
     """
 
-    def __init__(self, envvar, required=False, default=None, nargs=None, **kwargs):  # type: ignore
+    def __init__(self, envvar, required=False, default=None, nargs=None, **kwargs):
         if not default and envvar:
             if envvar in os.environ:
                 if nargs is not None and (nargs.isdigit() or nargs in ["*", "+"]):
@@ -37,7 +37,7 @@ class EnvDefault(argparse.Action):
             required = False
         super().__init__(default=default, required=required, nargs=nargs, **kwargs)
 
-    def __call__(self, parser, namespace, values, option_string=None):  # type: ignore
+    def __call__(self, parser, namespace, values, option_string=None):
         setattr(namespace, self.dest, values)
 
 
@@ -219,7 +219,7 @@ def main() -> None:
             ptpython.ipython.embed(
                 user_ns=driver.test_symbols(),
                 history_filename=history_path,
-            )  # type:ignore
+            )
         else:
             tic = time.time()
             driver.run_tests()

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

@@ -6,7 +6,7 @@ import subprocess
 import sys
 from abc import ABC, abstractmethod
 
-from remote_pdb import RemotePdb  # type:ignore
+from remote_pdb import RemotePdb
 
 from test_driver.logger import AbstractLogger
 
@@ -42,12 +42,12 @@ class Debug(DebugAbstract):
         self.logger.log_test_error(
             f"Breakpoint reached, run 'sudo {self.attach} {pattern}'"
         )
-        os.environ["bashInteractive"] = shutil.which("bash")  # type:ignore
+        os.environ["bashInteractive"] = shutil.which("bash")  # ty: ignore[invalid-assignment]
         if os.fork() == 0:
             subprocess.run(["sleep", pattern])
         else:
             # RemotePdb writes log messages to both stderr AND the logger,
             # which is the same here. Hence, disabling the remote_pdb logger
             # to avoid duplicate messages in the build log.
-            logging.root.manager.loggerDict["remote_pdb"].disabled = True  # type:ignore
+            logging.root.manager.loggerDict["remote_pdb"].disabled = True  # ty: ignore[invalid-assignment]
             RemotePdb(host=host, port=port).set_trace(sys._getframe().f_back)

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

@@ -336,10 +336,22 @@ class Driver:
     def start_all(self) -> None:
         """Start all machines"""
         with self.logger.nested("start all VMs"):
+            errors: list[tuple[str, BaseException]] = []
+
+            def start_machine(machine: BaseMachine) -> None:
+                try:
+                    machine.start()
+                except Exception as e:
+                    errors.append((machine.name, e))
+
             threads = []
             for machine in self.machines:
                 # Create a thread for each machine's start method
-                t = threading.Thread(target=machine.start, name=f"start-{machine.name}")
+                t = threading.Thread(
+                    target=start_machine,
+                    args=(machine,),
+                    name=f"start-{machine.name}",
+                )
                 threads.append(t)
                 t.start()
 
@@ -347,6 +359,12 @@ class Driver:
             for t in threads:
                 t.join()
 
+            if errors:
+                messages = [f"{name}: {e}" for name, e in errors]
+                raise MachineError(
+                    "Failed to start the following machines:\n" + "\n".join(messages)
+                )
+
     def join_all(self) -> None:
         """Wait for all machines to shut down"""
         with self.logger.nested("wait for all VMs to finish"):
@@ -418,7 +436,7 @@ class Driver:
             def __enter__(self) -> None:
                 driver.polling_conditions.append(self.condition)
 
-            def __exit__(self, a, b, c) -> None:  # type: ignore
+            def __exit__(self, a, b, c) -> None:
                 res = driver.polling_conditions.pop()
                 assert res is self.condition
 

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

@@ -1,5 +1,4 @@
 import atexit
-import codecs
 import os
 import sys
 import time
@@ -40,19 +39,19 @@ class AbstractLogger(ABC):
         pass
 
     @abstractmethod
-    def info(self, *args, **kwargs) -> None:  # type: ignore
+    def info(self, *args, **kwargs) -> None:
         pass
 
     @abstractmethod
-    def warning(self, *args, **kwargs) -> None:  # type: ignore
+    def warning(self, *args, **kwargs) -> None:
         pass
 
     @abstractmethod
-    def error(self, *args, **kwargs) -> None:  # type: ignore
+    def error(self, *args, **kwargs) -> None:
         pass
 
     @abstractmethod
-    def log_test_error(self, *args, **kwargs) -> None:  # type:ignore
+    def log_test_error(self, *args, **kwargs) -> None:
         pass
 
     @abstractmethod
@@ -103,19 +102,19 @@ class JunitXMLLogger(AbstractLogger):
         self.log(message)
         yield
 
-    def info(self, *args, **kwargs) -> None:  # type: ignore
+    def info(self, *args, **kwargs) -> None:
         if self._log_level <= LogLevel.INFO:
             self.tests[self.currentSubtest].stdout += args[0] + os.linesep
 
-    def warning(self, *args, **kwargs) -> None:  # type: ignore
+    def warning(self, *args, **kwargs) -> None:
         if self._log_level <= LogLevel.WARNING:
             self.tests[self.currentSubtest].stdout += args[0] + os.linesep
 
-    def error(self, *args, **kwargs) -> None:  # type: ignore
+    def error(self, *args, **kwargs) -> None:
         self.tests[self.currentSubtest].stderr += args[0] + os.linesep
         self.tests[self.currentSubtest].failure = True
 
-    def log_test_error(self, *args, **kwargs) -> None:  # type: ignore
+    def log_test_error(self, *args, **kwargs) -> None:
         self.error(*args, **kwargs)
 
     def log_serial(self, message: str, machine: str) -> None:
@@ -172,19 +171,19 @@ class CompositeLogger(AbstractLogger):
                 stack.enter_context(logger.nested(message, attributes))
             yield
 
-    def info(self, *args, **kwargs) -> None:  # type: ignore
+    def info(self, *args, **kwargs) -> None:
         for logger in self.logger_list:
             logger.info(*args, **kwargs)
 
-    def warning(self, *args, **kwargs) -> None:  # type: ignore
+    def warning(self, *args, **kwargs) -> None:
         for logger in self.logger_list:
             logger.warning(*args, **kwargs)
 
-    def log_test_error(self, *args, **kwargs) -> None:  # type: ignore
+    def log_test_error(self, *args, **kwargs) -> None:
         for logger in self.logger_list:
             logger.log_test_error(*args, **kwargs)
 
-    def error(self, *args, **kwargs) -> None:  # type: ignore
+    def error(self, *args, **kwargs) -> None:
         for logger in self.logger_list:
             logger.error(*args, **kwargs)
         sys.exit(1)
@@ -228,7 +227,8 @@ class TerminalLogger(AbstractLogger):
     def nested(self, message: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         self._eprint(
             self.maybe_prefix(
-                Style.BRIGHT + Fore.GREEN + message + Style.RESET_ALL, attributes
+                Style.BRIGHT + Fore.GREEN + message + Style.RESET_ALL,  # ty: ignore[unsupported-operator]
+                attributes,
             )
         )
 
@@ -237,15 +237,15 @@ class TerminalLogger(AbstractLogger):
         toc = time.time()
         self.log(f"(finished: {message}, in {toc - tic:.2f} seconds)", attributes)
 
-    def info(self, *args, **kwargs) -> None:  # type: ignore
+    def info(self, *args, **kwargs) -> None:
         if self._log_level <= LogLevel.INFO:
             self.log(*args, **kwargs)
 
-    def warning(self, *args, **kwargs) -> None:  # type: ignore
+    def warning(self, *args, **kwargs) -> None:
         if self._log_level <= LogLevel.WARNING:
             self.log(*args, **kwargs)
 
-    def error(self, *args, **kwargs) -> None:  # type: ignore
+    def error(self, *args, **kwargs) -> None:
         self.log(*args, **kwargs)
 
     def print_serial_logs(self, enable: bool) -> None:
@@ -258,17 +258,17 @@ class TerminalLogger(AbstractLogger):
         if not self._print_serial_logs:
             return
 
-        self._eprint(Style.DIM + f"{machine} # {message}" + Style.RESET_ALL)
+        self._eprint(Style.DIM + f"{machine} # {message}" + Style.RESET_ALL)  # ty: ignore[unsupported-operator]
 
-    def log_test_error(self, *args, **kwargs) -> None:  # type: ignore
-        prefix = Fore.RED + "!!! " + Style.RESET_ALL
+    def log_test_error(self, *args, **kwargs) -> None:
+        prefix = Fore.RED + "!!! " + Style.RESET_ALL  # ty: ignore[unsupported-operator]
         # NOTE: using `warning` instead of `error` to ensure it does not exit after printing the first log
         self.warning(f"{prefix}{args[0]}", *args[1:], **kwargs)
 
 
 class XMLLogger(AbstractLogger):
     def __init__(self, outfile: str) -> None:
-        self.logfile_handle = codecs.open(outfile, "wb")
+        self.logfile_handle = open(outfile, "wb")
         self.xml = XMLGenerator(self.logfile_handle, encoding="utf-8")
         self.queue: Queue[dict[str, str]] = Queue()
 
@@ -296,18 +296,18 @@ class XMLLogger(AbstractLogger):
         self.xml.characters(message)
         self.xml.endElement("line")
 
-    def info(self, *args, **kwargs) -> None:  # type: ignore
+    def info(self, *args, **kwargs) -> None:
         if self._log_level <= LogLevel.INFO:
             self.log(*args, **kwargs)
 
-    def warning(self, *args, **kwargs) -> None:  # type: ignore
+    def warning(self, *args, **kwargs) -> None:
         if self._log_level <= LogLevel.WARNING:
             self.log(*args, **kwargs)
 
-    def error(self, *args, **kwargs) -> None:  # type: ignore
+    def error(self, *args, **kwargs) -> None:
         self.log(*args, **kwargs)
 
-    def log_test_error(self, *args, **kwargs) -> None:  # type: ignore
+    def log_test_error(self, *args, **kwargs) -> None:
         self.log(*args, **kwargs)
 
     def log(self, message: str, attributes: dict[str, str] = {}) -> None:

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

@@ -24,9 +24,11 @@ from typing import Any
 
 from test_driver.errors import MachineError, RequestedAssertionFailed
 from test_driver.logger import AbstractLogger
-
-from .ocr import perform_ocr_on_screenshot, perform_ocr_variants_on_screenshot
-from .qmp import QMPSession
+from test_driver.machine.ocr import (
+    perform_ocr_on_screenshot,
+    perform_ocr_variants_on_screenshot,
+)
+from test_driver.machine.qmp import QMPSession
 
 CHAR_TO_KEY = {
     "A": "shift-a",
@@ -210,6 +212,7 @@ class QemuStartCommand:
             ),
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
             shell=True,
             cwd=state_dir,
             env=self.build_environment(state_dir, shared_dir),
@@ -1225,8 +1228,29 @@ class QemuMachine(BaseMachine):
             self.shell_path,
             allow_reboot,
         )
-        self.monitor, _ = monitor_socket.accept()
-        self.shell, _ = shell_socket.accept()
+
+        def accept_or_fail(sock: socket.socket, name: str) -> socket.socket:
+            """Accept a connection on a socket, polling the status to check
+            if the QEMU process is still alive. Without this, socket.accept()
+            would block forever if QEMU exits before connecting.
+            """
+            assert self.process
+            while True:
+                readable, _, _ = select.select([sock], [], [], 1.0)
+                if readable:
+                    conn, _ = sock.accept()
+                    return conn
+                rc = self.process.poll()
+                if rc is not None:
+                    output = ""
+                    if self.process.stdout:
+                        output = self.process.stdout.read().decode(errors="ignore")
+                    raise MachineError(
+                        f"QEMU process exited with code {rc} before connecting to {name} socket.\n{output}"
+                    )
+
+        self.monitor = accept_or_fail(monitor_socket, "monitor")
+        self.shell = accept_or_fail(shell_socket, "shell")
         self.qmp_client = QMPSession.from_path(self.qmp_path)
 
         # Store last serial console lines for use

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

@@ -25,7 +25,7 @@ class PollingCondition:
         seconds_interval: float = 2.0,
         description: str | None = None,
     ):
-        self.condition = condition  # type: ignore
+        self.condition = condition  # ty: ignore[invalid-assignment]
         self.seconds_interval = seconds_interval
         self.logger = logger
 
@@ -33,7 +33,7 @@ class PollingCondition:
             if condition.__doc__:
                 self.description = condition.__doc__
             else:
-                self.description = condition.__name__
+                self.description = condition.__name__  # ty: ignore[unresolved-attribute]
         else:
             self.description = str(description)
 
@@ -54,7 +54,7 @@ class PollingCondition:
 
             self.logger.info(last_message)
             try:
-                res = self.condition()  # type: ignore
+                res = self.condition()
             except Exception:
                 res = False
             res = res is None or res
@@ -89,7 +89,7 @@ class PollingCondition:
     def __enter__(self) -> None:
         self.entry_count += 1
 
-    def __exit__(self, exc_type, exc_value, traceback) -> None:  # type: ignore
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
         assert self.entered
         self.entry_count -= 1
         self.last_called = time.monotonic()

nixos/lib/testing/driver.nix

@@ -159,6 +159,16 @@ in
       defaultText = "hostPkgs.qemu_test";
     };
 
+    qemu.forceAccel = mkOption {
+      description = ''
+        Whether to force the use of hardware-accelerated virtualisation.
+        When enabled, QEMU will not fall back to the slower software emulation
+        (TCG) and will instead error out if the accelerator is not available.
+      '';
+      type = types.bool;
+      default = false;
+    };
+
     globalTimeout = mkOption {
       description = ''
         A global timeout for the complete test, expressed in seconds.

nixos/lib/testing/nodes.nix

@@ -71,7 +71,9 @@ let
       config.nodeDefaults
       {
         key = "base-qemu";
-        virtualisation.qemu.package = testModuleArgs.config.qemu.package;
+        virtualisation.qemu = {
+          inherit (testModuleArgs.config.qemu) package forceAccel;
+        };
         virtualisation.host.pkgs = hostPkgs;
       }
       testModuleArgs.config.extraBaseNodeModules

nixos/modules/virtualisation/qemu-vm.nix

@@ -306,8 +306,42 @@ let
       (builtins.concatStringsSep "")
     ]}
 
+    ${lib.optionalString cfg.qemu.forceAccel (
+      if hostPkgs.stdenv.hostPlatform.isLinux then
+        ''
+          # Check for hardware-accelerated virtualisation support (KVM)
+          if [ ! -e /dev/kvm ]; then
+            echo "forceAccel is enabled but /dev/kvm does not exist." >&2
+            echo "Hardware-accelerated virtualisation (KVM) is not available on this system." >&2
+            exit 1
+          elif [ ! -r /dev/kvm ] || [ ! -w /dev/kvm ]; then
+            echo "forceAccel is enabled but /dev/kvm is not accessible (permission denied)." >&2
+            echo "Check that the nix build user is in the 'kvm' group or that /dev/kvm has the correct permissions." >&2
+            exit 1
+          fi
+        ''
+      else if hostPkgs.stdenv.hostPlatform.isDarwin then
+        ''
+          # Check for hardware-accelerated virtualisation support (HVF)
+          if ! sysctl -n kern.hv_support 2>/dev/null | grep -q 1; then
+            echo "forceAccel is enabled but Hypervisor.framework is not available on this system." >&2
+            exit 1
+          fi
+        ''
+      else
+        ''
+          echo "forceAccel is enabled but no known accelerator is available for this platform." >&2
+          exit 1
+        ''
+    )}
+
     # Start QEMU.
-    exec ${qemu-common.qemuBinary qemu} \
+    exec ${
+      qemu-common.qemuBinaryWith {
+        qemuPkg = qemu;
+        forceAccel = cfg.qemu.forceAccel;
+      }
+    } \
         -name ${config.system.name} \
         -m ${toString config.virtualisation.memorySize} \
         -smp ${toString config.virtualisation.cores} \
@@ -728,6 +762,17 @@ in
         description = "QEMU package to use.";
       };
 
+      forceAccel = mkOption {
+        type = types.bool;
+        default = false;
+        description = ''
+          Whether to force the use of hardware-accelerated virtualisation.
+          When enabled, QEMU will not fall back to the slower software
+          emulation (TCG) and will instead error out if the accelerator is not
+          available.
+        '';
+      };
+
       options = mkOption {
         type = types.listOf types.str;
         default = [ ];