nixos/test-driver: apply ruff check suggestions (#358150)

nixos/lib/test-driver/default.nix

···
-{ lib
-, python3Packages
-, enableOCR ? false
-, qemu_pkg ? qemu_test
-, coreutils
-, imagemagick_light
-, netpbm
-, qemu_test
-, socat
-, ruff
-, tesseract4
-, vde2
-, extraPythonPackages ? (_ : [])
-, nixosTests
+{
+  lib,
+  python3Packages,
+  enableOCR ? false,
+  qemu_pkg ? qemu_test,
+  coreutils,
+  imagemagick_light,
+  netpbm,
+  qemu_test,
+  socat,
+  ruff,
+  tesseract4,
+  vde2,
+  extraPythonPackages ? (_: [ ]),
+  nixosTests,
 }:
 let
   fs = lib.fileset;
···
 python3Packages.buildPythonApplication {
   pname = "nixos-test-driver";
   version = "1.1";
+  pyproject = true;
+
   src = fs.toSource {
     root = ./.;
     fileset = fs.unions [
···
       ./extract-docstrings.py
     ];
   };
-  pyproject = true;
 
-  propagatedBuildInputs = [
-    coreutils
-    netpbm
-    python3Packages.colorama
-    python3Packages.junit-xml
-    python3Packages.ptpython
-    qemu_pkg
-    socat
-    vde2
-  ]
-    ++ (lib.optionals enableOCR [ imagemagick_light tesseract4 ])
+  build-system = with python3Packages; [
+    setuptools
+  ];
+
+  dependencies =
+    with python3Packages;
+    [
+      colorama
+      junit-xml
+      ptpython
+    ]
     ++ extraPythonPackages python3Packages;
 
-  nativeBuildInputs = [
-    python3Packages.setuptools
-  ];
+  propagatedBuildInputs =
+    [
+      coreutils
+      netpbm
+      qemu_pkg
+      socat
+      vde2
+    ]
+    ++ lib.optionals enableOCR [
+      imagemagick_light
+      tesseract4
+    ];
 
   passthru.tests = {
     inherit (nixosTests.nixos-test-driver) driver-timeout;
   };
 
   doCheck = true;
-  nativeCheckInputs = with python3Packages; [ mypy ruff black ];
+
+  nativeCheckInputs = with python3Packages; [
+    mypy
+    ruff
+  ];
+
   checkPhase = ''
     echo -e "\x1b[32m## run mypy\x1b[0m"
     mypy test_driver extract-docstrings.py
-    echo -e "\x1b[32m## run ruff\x1b[0m"
+    echo -e "\x1b[32m## run ruff check\x1b[0m"
     ruff check .
-    echo -e "\x1b[32m## run black\x1b[0m"
-    black --check --diff .
+    echo -e "\x1b[32m## run ruff format\x1b[0m"
+    ruff format --check --diff .
   '';
 }

nixos/lib/test-driver/pyproject.toml

···
 test_driver = ["py.typed"]
 
 [tool.ruff]
+target-version = "py312"
 line-length = 88
 
 lint.select = ["E", "F", "I", "U", "N"]
···
 [[tool.mypy.overrides]]
 module = "junit_xml.*"
 ignore_missing_imports = true
-
-[tool.black]
-line-length = 88
-target-version = ['py39']
-include = '\.pyi?$'
 
 [tool.mypy]
 warn_redundant_casts = true

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

···
 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 colorama import Fore, Style
 
···
     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
     logger: AbstractLogger
 
     def __init__(
         self,
-        start_scripts: List[str],
-        vlans: List[int],
+        start_scripts: list[str],
+        vlans: list[int],
         tests: str,
         out_dir: Path,
         logger: AbstractLogger,
···
             vlans = list(set(vlans))
             self.vlans = [VLan(nr, tmp_dir, self.logger) 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)
 
···
                 self.logger.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)
···
         self,
         start_command: str | dict,
         *,
-        name: Optional[str] = None,
+        name: str | None = None,
         keep_vm_state: bool = False,
     ) -> Machine:
         # Legacy args handling
···
 
     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/logger.py

···
 import time
 import unicodedata
 from abc import ABC, abstractmethod
+from collections.abc import Iterator
 from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from queue import Empty, Queue
-from typing import Any, Dict, Iterator, List
+from typing import Any
 from xml.sax.saxutils import XMLGenerator
 from xml.sax.xmlreader import AttributesImpl
 
···
 
 class AbstractLogger(ABC):
     @abstractmethod
-    def log(self, message: str, attributes: Dict[str, str] = {}) -> None:
+    def log(self, message: str, attributes: dict[str, str] = {}) -> None:
         pass
 
     @abstractmethod
     @contextmanager
-    def subtest(self, name: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def subtest(self, name: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         pass
 
     @abstractmethod
     @contextmanager
-    def nested(self, message: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def nested(self, message: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         pass
 
     @abstractmethod
···
         self._print_serial_logs = True
         atexit.register(self.close)
 
-    def log(self, message: str, attributes: Dict[str, str] = {}) -> None:
+    def log(self, message: str, attributes: dict[str, str] = {}) -> None:
         self.tests[self.currentSubtest].stdout += message + os.linesep
 
     @contextmanager
-    def subtest(self, name: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def subtest(self, name: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         old_test = self.currentSubtest
         self.tests.setdefault(name, self.TestCaseState())
         self.currentSubtest = name
···
         self.currentSubtest = old_test
 
     @contextmanager
-    def nested(self, message: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def nested(self, message: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         self.log(message)
         yield
 
···
 
 
 class CompositeLogger(AbstractLogger):
-    def __init__(self, logger_list: List[AbstractLogger]) -> None:
+    def __init__(self, logger_list: list[AbstractLogger]) -> None:
         self.logger_list = logger_list
 
     def add_logger(self, logger: AbstractLogger) -> None:
         self.logger_list.append(logger)
 
-    def log(self, message: str, attributes: Dict[str, str] = {}) -> None:
+    def log(self, message: str, attributes: dict[str, str] = {}) -> None:
         for logger in self.logger_list:
             logger.log(message, attributes)
 
     @contextmanager
-    def subtest(self, name: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def subtest(self, name: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         with ExitStack() as stack:
             for logger in self.logger_list:
                 stack.enter_context(logger.subtest(name, attributes))
             yield
 
     @contextmanager
-    def nested(self, message: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def nested(self, message: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         with ExitStack() as stack:
             for logger in self.logger_list:
                 stack.enter_context(logger.nested(message, attributes))
···
     def __init__(self) -> None:
         self._print_serial_logs = True
 
-    def maybe_prefix(self, message: str, attributes: Dict[str, str]) -> str:
+    def maybe_prefix(self, message: str, attributes: dict[str, str]) -> str:
         if "machine" in attributes:
             return f"{attributes['machine']}: {message}"
         return message
···
     def _eprint(*args: object, **kwargs: Any) -> None:
         print(*args, file=sys.stderr, **kwargs)
 
-    def log(self, message: str, attributes: Dict[str, str] = {}) -> None:
+    def log(self, message: str, attributes: dict[str, str] = {}) -> None:
         self._eprint(self.maybe_prefix(message, attributes))
 
     @contextmanager
-    def subtest(self, name: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def subtest(self, name: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         with self.nested("subtest: " + name, attributes):
             yield
 
     @contextmanager
-    def nested(self, message: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    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
···
     def sanitise(self, message: str) -> str:
         return "".join(ch for ch in message if unicodedata.category(ch)[0] != "C")
 
-    def maybe_prefix(self, message: str, attributes: Dict[str, str]) -> str:
+    def maybe_prefix(self, message: str, attributes: dict[str, str]) -> str:
         if "machine" in attributes:
             return f"{attributes['machine']}: {message}"
         return message
 
-    def log_line(self, message: str, attributes: Dict[str, str]) -> None:
+    def log_line(self, message: str, attributes: dict[str, str]) -> None:
         self.xml.startElement("line", attrs=AttributesImpl(attributes))
         self.xml.characters(message)
         self.xml.endElement("line")
···
     def error(self, *args, **kwargs) -> None:  # type: ignore
         self.log(*args, **kwargs)
 
-    def log(self, message: str, attributes: Dict[str, str] = {}) -> None:
+    def log(self, message: str, attributes: dict[str, str] = {}) -> None:
         self.drain_log_queue()
         self.log_line(message, attributes)
 
···
 
         self.enqueue({"msg": message, "machine": machine, "type": "serial"})
 
-    def enqueue(self, item: Dict[str, str]) -> None:
+    def enqueue(self, item: dict[str, str]) -> None:
         self.queue.put(item)
 
     def drain_log_queue(self) -> None:
···
             pass
 
     @contextmanager
-    def subtest(self, name: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def subtest(self, name: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         with self.nested("subtest: " + name, attributes):
             yield
 
     @contextmanager
-    def nested(self, message: str, attributes: Dict[str, str] = {}) -> Iterator[None]:
+    def nested(self, message: str, attributes: dict[str, str] = {}) -> Iterator[None]:
         self.xml.startElement("nest", attrs=AttributesImpl({}))
         self.xml.startElement("head", attrs=AttributesImpl(attributes))
         self.xml.characters(message)

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

···
 import tempfile
 import threading
 import time
+from collections.abc import Callable, Iterable
 from contextlib import _GeneratorContextManager, nullcontext
 from pathlib import Path
 from queue import Queue
-from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
+from typing import Any
 
 from test_driver.logger import AbstractLogger
 
···
 
 def _perform_ocr_on_screenshot(
     screenshot_path: str, model_ids: Iterable[int]
-) -> List[str]:
+) -> list[str]:
     if shutil.which("tesseract") is None:
         raise Exception("OCR requested but enableOCR is false")
 
···
     start_command: StartCommand
     keep_vm_state: bool
 
-    process: Optional[subprocess.Popen]
-    pid: Optional[int]
-    monitor: Optional[socket.socket]
-    qmp_client: Optional[QMPSession]
-    shell: Optional[socket.socket]
-    serial_thread: Optional[threading.Thread]
+    process: subprocess.Popen | None
+    pid: int | None
+    monitor: socket.socket | None
+    qmp_client: QMPSession | None
+    shell: socket.socket | None
+    serial_thread: threading.Thread | None
 
     booted: bool
     connected: bool
     # Store last serial console lines for use
     # of wait_for_console_text
     last_lines: Queue = Queue()
-    callbacks: List[Callable]
+    callbacks: list[Callable]
 
     def __repr__(self) -> str:
         return f"<Machine '{self.name}'>"
···
         logger: AbstractLogger,
         name: str = "machine",
         keep_vm_state: bool = False,
-        callbacks: Optional[List[Callable]] = None,
+        callbacks: list[Callable] | None = None,
     ) -> None:
         self.out_dir = out_dir
         self.tmp_dir = tmp_dir
···
     def log_serial(self, msg: str) -> None:
         self.logger.log_serial(msg, self.name)
 
-    def nested(self, msg: str, attrs: Dict[str, str] = {}) -> _GeneratorContextManager:
+    def nested(self, msg: str, attrs: dict[str, str] = {}) -> _GeneratorContextManager:
         my_attrs = {"machine": self.name}
         my_attrs.update(attrs)
         return self.logger.nested(msg, my_attrs)
···
         return self.wait_for_monitor_prompt()
 
     def wait_for_unit(
-        self, unit: str, user: Optional[str] = None, timeout: int = 900
+        self, unit: str, user: str | None = None, timeout: int = 900
     ) -> None:
         """
         Wait for a systemd unit to get into "active" state.
···
         ):
             retry(check_active, timeout)
 
-    def get_unit_info(self, unit: str, user: Optional[str] = None) -> Dict[str, str]:
+    def get_unit_info(self, unit: str, user: str | None = None) -> dict[str, str]:
         status, lines = self.systemctl(f'--no-pager show "{unit}"', user)
         if status != 0:
             raise Exception(
···
 
         line_pattern = re.compile(r"^([^=]+)=(.*)$")
 
-        def tuple_from_line(line: str) -> Tuple[str, str]:
+        def tuple_from_line(line: str) -> tuple[str, str]:
             match = line_pattern.match(line)
             assert match is not None
             return match[1], match[2]
···
         self,
         unit: str,
         property: str,
-        user: Optional[str] = None,
+        user: str | None = None,
     ) -> str:
         status, lines = self.systemctl(
             f'--no-pager show "{unit}" --property="{property}"',
···
         assert match[1] == property, invalid_output_message
         return match[2]
 
-    def systemctl(self, q: str, user: Optional[str] = None) -> Tuple[int, str]:
+    def systemctl(self, q: str, user: str | None = None) -> tuple[int, str]:
         """
         Runs `systemctl` commands with optional support for
         `systemctl --user`
···
         command: str,
         check_return: bool = True,
         check_output: bool = True,
-        timeout: Optional[int] = 900,
-    ) -> Tuple[int, str]:
+        timeout: int | None = 900,
+    ) -> tuple[int, str]:
         """
         Execute a shell command, returning a list `(status, stdout)`.
 
···
 
         return (rc, output.decode(errors="replace"))
 
-    def shell_interact(self, address: Optional[str] = None) -> None:
+    def shell_interact(self, address: str | None = None) -> None:
         """
         Allows you to directly interact with the guest shell. This should
         only be used during test development, not in production tests.
···
                 break
             self.send_console(char.decode())
 
-    def succeed(self, *commands: str, timeout: Optional[int] = None) -> str:
+    def succeed(self, *commands: str, timeout: int | None = None) -> str:
         """
         Execute a shell command, raising an exception if the exit status is
         not zero, otherwise returning the standard output. Similar to `execute`,
···
                 output += out
         return output
 
-    def fail(self, *commands: str, timeout: Optional[int] = None) -> str:
+    def fail(self, *commands: str, timeout: int | None = None) -> str:
         """
         Like `succeed`, but raising an exception if the command returns a zero
         status.
···
         with self.nested(f"waiting for {regexp} to appear on tty {tty}"):
             retry(tty_matches, timeout)
 
-    def send_chars(self, chars: str, delay: Optional[float] = 0.01) -> None:
+    def send_chars(self, chars: str, delay: float | None = 0.01) -> None:
         """
         Simulate typing a sequence of characters on the virtual keyboard,
         e.g., `send_chars("foobar\n")` will type the string `foobar`
···
         with self.nested(f"waiting for TCP port {port} on {addr} to be closed"):
             retry(port_is_closed, timeout)
 
-    def start_job(self, jobname: str, user: Optional[str] = None) -> Tuple[int, str]:
+    def start_job(self, jobname: str, user: str | None = None) -> tuple[int, str]:
         return self.systemctl(f"start {jobname}", user)
 
-    def stop_job(self, jobname: str, user: Optional[str] = None) -> Tuple[int, str]:
+    def stop_job(self, jobname: str, user: str | None = None) -> tuple[int, str]:
         return self.systemctl(f"stop {jobname}", user)
 
     def wait_for_job(self, jobname: str) -> None:
···
         """Debugging: Dump the contents of the TTY<n>"""
         self.execute(f"fold -w 80 /dev/vcs{tty} | systemd-cat")
 
-    def _get_screen_text_variants(self, model_ids: Iterable[int]) -> List[str]:
+    def _get_screen_text_variants(self, model_ids: Iterable[int]) -> list[str]:
         with tempfile.TemporaryDirectory() as tmpdir:
             screenshot_path = os.path.join(tmpdir, "ppm")
             self.send_monitor_command(f"screendump {screenshot_path}")
             return _perform_ocr_on_screenshot(screenshot_path, model_ids)
 
-    def get_screen_text_variants(self) -> List[str]:
+    def get_screen_text_variants(self) -> list[str]:
         """
         Return a list of different interpretations of what is currently
         visible on the machine's screen using optical character
···
                     pass
 
     def send_key(
-        self, key: str, delay: Optional[float] = 0.01, log: Optional[bool] = True
+        self, key: str, delay: float | None = 0.01, log: bool | None = True
     ) -> None:
         """
         Simulate pressing keys on the virtual keyboard, e.g.,
···
         with self.nested("waiting for the X11 server"):
             retry(check_x, timeout)
 
-    def get_window_names(self) -> List[str]:
+    def get_window_names(self) -> list[str]:
         return self.succeed(
             r"xwininfo -root -tree | sed 's/.*0x[0-9a-f]* \"\([^\"]*\)\".*/\1/; t; d'"
         ).splitlines()

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

···
 import time
+from collections.abc import Callable
 from math import isfinite
-from typing import Callable, Optional
 
 from test_driver.logger import AbstractLogger
 
···
 class PollingCondition:
     condition: Callable[[], bool]
     seconds_interval: float
-    description: Optional[str]
+    description: str | None
     logger: AbstractLogger
 
     last_called: float
···
 
     def __init__(
         self,
-        condition: Callable[[], Optional[bool]],
+        condition: Callable[[], bool | None],
         logger: AbstractLogger,
         seconds_interval: float = 2.0,
-        description: Optional[str] = None,
+        description: str | None = None,
     ):
         self.condition = condition  # type: ignore
         self.seconds_interval = seconds_interval

pkgs/by-name/ru/ruff/package.nix

···
   nix-update-script,
   versionCheckHook,
   libiconv,
+  nixosTests,
 }:
 
 python3Packages.buildPythonPackage rec {
···
   passthru = {
     tests = {
       inherit ruff-lsp;
+      nixos-test-driver-busybox = nixosTests.nixos-test-driver.busybox;
     };
     updateScript = nix-update-script { };
   };