nixos/test-driver: target python 3.12

nixos/lib/test-driver/pyproject.toml

@@ -17,6 +17,7 @@ find = {}
 test_driver = ["py.typed"]
 
 [tool.ruff]
+target-version = "py312"
 line-length = 88
 
 lint.select = ["E", "F", "I", "U", "N"]

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

@@ -3,10 +3,10 @@ import re
 import signal
 import tempfile
 import threading
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from contextlib import AbstractContextManager, contextmanager
 from pathlib import Path
-from typing import Any, Callable, Optional, Union
+from typing import Any
 
 from colorama import Fore, Style
 
@@ -208,7 +208,7 @@ class Driver:
         self,
         start_command: str | dict,
         *,
-        name: Optional[str] = None,
+        name: str | None = None,
         keep_vm_state: bool = False,
     ) -> Machine:
         # Legacy args handling
@@ -274,11 +274,11 @@ class Driver:
 
     def polling_condition(
         self,
-        fun_: Optional[Callable] = None,
+        fun_: Callable | None = None,
         *,
         seconds_interval: float = 2.0,
-        description: Optional[str] = None,
-    ) -> Union[Callable[[Callable], AbstractContextManager], AbstractContextManager]:
+        description: str | None = None,
+    ) -> Callable[[Callable], AbstractContextManager] | AbstractContextManager:
         driver = self
 
         class Poll:

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

@@ -12,11 +12,11 @@ import sys
 import tempfile
 import threading
 import time
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from contextlib import _GeneratorContextManager, nullcontext
 from pathlib import Path
 from queue import Queue
-from typing import Any, Callable, Optional
+from typing import Any
 
 from test_driver.logger import AbstractLogger
 
@@ -249,12 +249,12 @@ class Machine:
     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
@@ -274,7 +274,7 @@ class Machine:
         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
@@ -344,7 +344,7 @@ class Machine:
         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.
@@ -374,7 +374,7 @@ class Machine:
         ):
             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(
@@ -400,7 +400,7 @@ class Machine:
         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}"',
@@ -425,7 +425,7 @@ class Machine:
         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`
@@ -481,7 +481,7 @@ class Machine:
         command: str,
         check_return: bool = True,
         check_output: bool = True,
-        timeout: Optional[int] = 900,
+        timeout: int | None = 900,
     ) -> tuple[int, str]:
         """
         Execute a shell command, returning a list `(status, stdout)`.
@@ -549,7 +549,7 @@ class Machine:
 
         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.
@@ -596,7 +596,7 @@ class Machine:
                 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`,
@@ -613,7 +613,7 @@ class Machine:
                 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.
@@ -725,7 +725,7 @@ class Machine:
         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`
@@ -799,10 +799,10 @@ class Machine:
         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:
@@ -1029,7 +1029,7 @@ class Machine:
                     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.,

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

@@ -1,6 +1,6 @@
 import time
+from collections.abc import Callable
 from math import isfinite
-from typing import Callable, Optional
 
 from test_driver.logger import AbstractLogger
 
@@ -12,7 +12,7 @@ class PollingConditionError(Exception):
 class PollingCondition:
     condition: Callable[[], bool]
     seconds_interval: float
-    description: Optional[str]
+    description: str | None
     logger: AbstractLogger
 
     last_called: float
@@ -20,10 +20,10 @@ class PollingCondition:
 
     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