diff --git a/nixos/lib/test-driver/pyproject.toml b/nixos/lib/test-driver/pyproject.toml index fe2ce75fd632..ac83eed268d9 100644 --- a/nixos/lib/test-driver/pyproject.toml +++ b/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"] diff --git a/nixos/lib/test-driver/test_driver/driver.py b/nixos/lib/test-driver/test_driver/driver.py index 6f37af954bc5..ca778a576f72 100644 --- a/nixos/lib/test-driver/test_driver/driver.py +++ b/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: diff --git a/nixos/lib/test-driver/test_driver/machine.py b/nixos/lib/test-driver/test_driver/machine.py index f4ec494beee2..c423ad8a3fc0 100644 --- a/nixos/lib/test-driver/test_driver/machine.py +++ b/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., diff --git a/nixos/lib/test-driver/test_driver/polling_condition.py b/nixos/lib/test-driver/test_driver/polling_condition.py index 1cccaf2c71e7..1a8091cf4471 100644 --- a/nixos/lib/test-driver/test_driver/polling_condition.py +++ b/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