diff --git a/nixos/doc/manual/development/writing-nixos-tests.section.md b/nixos/doc/manual/development/writing-nixos-tests.section.md index c43f6bb34299..ef5dc6984c2d 100644 --- a/nixos/doc/manual/development/writing-nixos-tests.section.md +++ b/nixos/doc/manual/development/writing-nixos-tests.section.md @@ -276,12 +276,15 @@ with foo_running: `polling_condition` takes the following (optional) arguments: -`seconds_interval` +`interval` -: specifies how often the condition should be polled: +: specifies how often the condition should be polled, as a `datetime.timedelta`: ```py -@polling_condition(seconds_interval=10) +import datetime as dt + + +@polling_condition(interval=dt.timedelta(seconds=10)) def foo_running(): machine.succeed("pgrep -x foo") ``` diff --git a/nixos/doc/manual/release-notes/rl-2611.section.md b/nixos/doc/manual/release-notes/rl-2611.section.md index b03d72b733dd..8de36a99777a 100644 --- a/nixos/doc/manual/release-notes/rl-2611.section.md +++ b/nixos/doc/manual/release-notes/rl-2611.section.md @@ -166,6 +166,8 @@ - The `shell_interact()` function on interactive runs of NixOS VM tests has been deprecated. Use the SSH backdoor instead. +- NixOS VM tests now prefer to express durations and timeouts as `datetime.timedelta` values instead of bare numbers. Methods such as `machine.wait_until_succeeds`, `machine.sleep`, `retry`, and `polling_condition` now accept a `timedelta` (e.g., `machine.wait_for_unit("sshd.service", timeout=datetime.timedelta(minutes=1))`). Passing an `int`/`float` as seconds still works but now emits a deprecation warning. Argument names that explicitly defined units were preserved but have had `timedelta` equivalents introduced (`timeout_seconds` → `timeout`, `secs` → `duration`, `seconds_interval` → `interval`). + - `darwin.linux-builder-vz` has been added: a variant of `darwin.linux-builder` that runs the builder guest on Apple's Virtualization.framework via the new `vzvm` package, translating `x86_64-linux` builds with Rosetta instead of emulating them. Apple silicon hosts only. As part of this, the `nixos/modules/profiles/nix-builder-vm.nix` profile has been split into the backend-neutral `nixos/modules/profiles/nix-builder.nix` and a QEMU-specific part. Existing imports of `nix-builder-vm.nix` keep working unchanged. - [services.netbox](#opt-services.netbox.enable) has received a number of updates: diff --git a/nixos/lib/test-driver/src/test_driver/driver.py b/nixos/lib/test-driver/src/test_driver/driver.py index 36f819f0e94d..48ccae7c3c63 100644 --- a/nixos/lib/test-driver/src/test_driver/driver.py +++ b/nixos/lib/test-driver/src/test_driver/driver.py @@ -1,3 +1,4 @@ +import datetime as dt import json import os import re @@ -7,6 +8,7 @@ import sys import tempfile import threading import traceback +import warnings from collections.abc import Callable, Generator, Iterator from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass @@ -18,6 +20,7 @@ from colorama import Style from pydantic import BaseModel from test_driver.debug import DebugAbstract, DebugNop +from test_driver.duration import as_timedelta from test_driver.errors import MachineError, RequestedAssertionFailed from test_driver.logger import AbstractLogger from test_driver.machine import ( @@ -40,7 +43,7 @@ class DriverConfiguration(BaseModel): vms: dict[str, NodeConfiguration] containers: dict[str, NodeConfiguration] vlans: list[int] - global_timeout: int + global_timeout: dt.timedelta enable_ssh_backdoor: bool test_script: Path @@ -166,7 +169,7 @@ class Driver: def __enter__(self) -> "Driver": self.race_timer = threading.Timer( - self.config.global_timeout, self.terminate_test + self.config.global_timeout.total_seconds(), self.terminate_test ) tmp_dir = get_tmp_dir() @@ -428,7 +431,7 @@ class Driver: def run_tests(self) -> None: """Run the test script (for non-interactive test runs)""" self.logger.info( - f"Test will time out and terminate in {self.config.global_timeout} seconds" + f"Test will time out and terminate in {self.config.global_timeout.total_seconds()} seconds" ) self.race_timer.start() self.test_script() @@ -527,18 +530,32 @@ class Driver: self, fun_: Callable | None = None, *, - seconds_interval: float = 2.0, + interval: dt.timedelta | None = None, + seconds_interval: float | None = None, description: str | None = None, ) -> Callable[[Callable], AbstractContextManager] | AbstractContextManager: driver = self + if seconds_interval is not None: + if interval is not None: + raise TypeError( + "polling_condition() got both 'interval' and 'seconds_interval' arguments. Pass only 'interval'" + ) + warnings.warn( + "polling_condition(): The 'seconds_interval' argument is deprecated. Use 'interval' instead.", + ) + interval = as_timedelta(seconds_interval) + + if interval is None: + interval = dt.timedelta(seconds=2) + class Poll: def __init__(self, fun: Callable): self.condition = PollingCondition( fun, driver.logger, - seconds_interval, - description, + description=description, + interval=interval, ) def __enter__(self) -> None: @@ -548,7 +565,24 @@ class Driver: res = driver.polling_conditions.pop() assert res is self.condition - def wait(self, timeout: int = 900) -> None: + def wait( + self, + timeout: dt.timedelta | None = None, + timeout_seconds: int | None = None, + ) -> None: + if timeout_seconds is not None: + if timeout is not None: + raise TypeError( + "wait() got both 'timeout' and 'timeout_seconds' arguments. Pass only 'timeout'" + ) + warnings.warn( + "wait(): The 'timeout_seconds' argument is deprecated. Use 'timeout' instead.", + ) + timeout = as_timedelta(timeout_seconds) + + if timeout is None: + timeout = dt.timedelta(minutes=15) + def condition(last: bool) -> bool: if last: driver.logger.info( @@ -562,7 +596,7 @@ class Driver: return ret with driver.logger.nested(f"waiting for {self.condition.description}"): - retry(condition, timeout_seconds=timeout) + retry(condition, timeout=timeout) if fun_ is None: return Poll diff --git a/nixos/lib/test-driver/src/test_driver/duration.py b/nixos/lib/test-driver/src/test_driver/duration.py new file mode 100644 index 000000000000..ce085fbfa8a5 --- /dev/null +++ b/nixos/lib/test-driver/src/test_driver/duration.py @@ -0,0 +1,42 @@ +import datetime as dt +import warnings + +# A duration accepted by the public test-driver API. +# `datetime.timedelta` is the preferred form. Bare `int`/`float` values are still +# accepted for backwards compatibility (interpreted as a number of seconds), +# but doing so is deprecated: use `datetime.timedelta` instead. +Duration = dt.timedelta | int | float + + +def as_timedelta(duration: Duration) -> dt.timedelta: + """Coerce a `Duration` into a `datetime.timedelta`. + + Bare numbers are interpreted as seconds. This keeps existing test scripts + that pass plain integers working while `datetime.timedelta` becomes the preferred + way to express durations. + """ + if isinstance(duration, dt.timedelta): + return duration + if isinstance(duration, bool): + raise TypeError(f"expected a duration, got bool: {duration!r}") + if isinstance(duration, (int, float)): + return dt.timedelta(seconds=duration) + raise TypeError( + f"expected a timedelta, int, or float duration, got {type(duration).__name__}" + ) + + +def as_seconds(duration: Duration) -> float: + """Coerce a `Duration` into a floating-point number of seconds.""" + return as_timedelta(duration).total_seconds() + + +def _warn_if_numeric_duration(duration: Duration | None, func_name: str) -> None: + """Emit a warning if `duration` is a bare number instead of a `timedelta`.""" + if duration is None or isinstance(duration, bool): + return + if isinstance(duration, (int, float)): + warnings.warn( + f"{func_name}(): passing a bare int/float as a duration is " + "deprecated. Use datetime.timedelta instead.", + ) diff --git a/nixos/lib/test-driver/src/test_driver/machine/__init__.py b/nixos/lib/test-driver/src/test_driver/machine/__init__.py index f47d5ce7f7af..1ad13795e43d 100644 --- a/nixos/lib/test-driver/src/test_driver/machine/__init__.py +++ b/nixos/lib/test-driver/src/test_driver/machine/__init__.py @@ -1,4 +1,5 @@ import base64 +import datetime as dt import io import os import platform @@ -22,6 +23,12 @@ from pathlib import Path from queue import Queue from typing import Any +from test_driver.duration import ( + Duration, + _warn_if_numeric_duration, + as_seconds, + as_timedelta, +) from test_driver.efi import EfiVariable, EfiVars from test_driver.errors import MachineError, RequestedAssertionFailed from test_driver.logger import AbstractLogger @@ -99,24 +106,45 @@ def make_command(args: list) -> str: return " ".join(map(shlex.quote, (map(str, args)))) -def retry(fn: Callable, timeout_seconds: int = 900) -> None: +def retry( + fn: Callable, + timeout: dt.timedelta | None = None, + timeout_seconds: int | None = None, +) -> None: """Call the given function repeatedly, with a one second interval between retries, until it returns True or a timeout is reached. Note that the timeout shown will include the time of the last attempted run. + + Has a default timeout of 15 minutes which can be modified. + The ``timeout_seconds`` argument is deprecated. Use ``timeout`` instead. """ + if timeout_seconds is not None: + if timeout is not None: + raise TypeError( + "retry() got both 'timeout' and 'timeout_seconds' arguments. Pass only 'timeout'" + ) + warnings.warn( + "retry(): The 'timeout_seconds' argument is deprecated. Use 'timeout' instead.", + ) + timeout = as_timedelta(timeout_seconds) + + if timeout is None: + timeout = dt.timedelta(minutes=15) start_time = time.monotonic() - while time.monotonic() - start_time < timeout_seconds: + def elapsed() -> dt.timedelta: + return dt.timedelta(seconds=time.monotonic() - start_time) + + while elapsed() < timeout: if fn(False): return time.sleep(1) - elapsed = time.monotonic() - start_time - if not fn(True): raise RequestedAssertionFailed( - f"action timed out after {elapsed:.2f} seconds (timeout={timeout_seconds})" + f"action timed out after {elapsed().total_seconds():.2f} seconds " + f"(timeout={timeout.total_seconds()})" ) @@ -341,13 +369,17 @@ class BaseMachine(ABC): return self.execute(f"systemctl {q}") def wait_for_unit( - self, unit: str, user: str | None = None, timeout: int = 900 + self, + unit: str, + user: str | None = None, + timeout: Duration = dt.timedelta(minutes=15), ) -> None: """ Wait for a systemd unit to get into "active" state. Throws exceptions on "failed" and "inactive" states as well as after timing out. """ + _warn_if_numeric_duration(timeout, "wait_for_unit") def check_active(_last_try: bool) -> bool: state = self.get_unit_property(unit, "ActiveState", user) @@ -369,7 +401,7 @@ class BaseMachine(ABC): f"waiting for unit {unit}" + (f" with user {user}" if user is not None else "") ): - retry(check_active, timeout) + retry(check_active, as_timedelta(timeout)) def get_unit_info(self, unit: str, user: str | None = None) -> dict[str, str]: """ @@ -443,13 +475,14 @@ class BaseMachine(ABC): f"'{require_state}' but it is in state '{state}'" ) - def succeed(self, *commands: str, timeout: int | None = None) -> str: + def succeed(self, *commands: str, timeout: Duration | 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`, except that the timeout is `None` by default. See `execute` for details on command execution. """ + _warn_if_numeric_duration(timeout, "succeed") output = "" for command in commands: with self.nested(f"must succeed: {command}"): @@ -462,11 +495,12 @@ class BaseMachine(ABC): output += out return output - def fail(self, *commands: str, timeout: int | None = None) -> str: + def fail(self, *commands: str, timeout: Duration | None = None) -> str: """ Like `succeed`, but raising an exception if the command returns a zero status. """ + _warn_if_numeric_duration(timeout, "fail") output = "" for command in commands: with self.nested(f"must fail: {command}"): @@ -478,14 +512,23 @@ class BaseMachine(ABC): output += out return output - def wait_until_succeeds(self, command: str, timeout: int = 900) -> str: + def wait_until_succeeds( + self, command: str, timeout: Duration = dt.timedelta(minutes=15) + ) -> 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. - `wait_until_succeeds(cmd, timeout=10)`. See `execute` for details on - command execution. + Repeat a shell command on 1-second intervals until it succeeds. + Has a default timeout of 15 minutes which can be modified. + Example: + ```py + import datetime as dt + + wait_until_succeeds(cmd, timeout=dt.timedelta(seconds=10)) + ``` + A bare number is still accepted and interpreted as a number of seconds. + See `execute` for details on command execution. Throws an exception on timeout. """ + _warn_if_numeric_duration(timeout, "wait_until_succeeds") output = "" def check_success(_last_try: bool) -> bool: @@ -494,13 +537,16 @@ class BaseMachine(ABC): return status == 0 with self.nested(f"waiting for success: {command}"): - retry(check_success, timeout) + retry(check_success, as_timedelta(timeout)) return output - def wait_until_fails(self, command: str, timeout: int = 900) -> str: + def wait_until_fails( + self, command: str, timeout: Duration = dt.timedelta(minutes=15) + ) -> str: """ Like `wait_until_succeeds`, but repeating the command until it fails. """ + _warn_if_numeric_duration(timeout, "wait_until_fails") output = "" def check_failure(_last_try: bool) -> bool: @@ -509,47 +555,73 @@ class BaseMachine(ABC): return status != 0 with self.nested(f"waiting for failure: {command}"): - retry(check_failure, timeout) + retry(check_failure, as_timedelta(timeout)) return output - def sleep(self, secs: int) -> None: - # We want to sleep in *guest* time, not *host* time. - self.succeed(f"sleep {secs}") + def sleep( + self, duration: dt.timedelta | None = None, secs: int | None = None + ) -> None: + if secs is not None: + if duration is not None: + raise TypeError( + "sleep() got both 'duration' and 'secs'. Pass only 'duration'" + ) + warnings.warn( + "sleep(): The 'secs' argument is deprecated. Use 'duration' instead.", + ) + duration = as_timedelta(secs) - def wait_for_file(self, filename: str, timeout: int = 900) -> None: + if duration is None: + raise TypeError("sleep() missing required argument 'duration'") + + # We want to sleep in *guest* time, not *host* time. + self.succeed(f"sleep {as_seconds(duration)}") + + def wait_for_file( + self, filename: str, timeout: Duration = dt.timedelta(minutes=15) + ) -> None: """ Waits until the file exists in the machine's file system. """ + _warn_if_numeric_duration(timeout, "wait_for_file") def check_file(_last_try: bool) -> bool: status, _ = self.execute(f"test -e {filename}") return status == 0 with self.nested(f"waiting for file '{filename}'"): - retry(check_file, timeout) + retry(check_file, as_timedelta(timeout)) def wait_for_open_port( - self, port: int, addr: str = "localhost", timeout: int = 900 + self, + port: int, + addr: str = "localhost", + timeout: Duration = dt.timedelta(minutes=15), ) -> None: """ Wait until a process is listening on the given TCP port and IP address (default `localhost`). """ + _warn_if_numeric_duration(timeout, "wait_for_open_port") def port_is_open(_last_try: bool) -> bool: status, _ = self.execute(f"nc -z {addr} {port}") return status == 0 with self.nested(f"waiting for TCP port {port} on {addr}"): - retry(port_is_open, timeout) + retry(port_is_open, as_timedelta(timeout)) def wait_for_open_unix_socket( - self, addr: str, is_datagram: bool = False, timeout: int = 900 + self, + addr: str, + is_datagram: bool = False, + timeout: Duration = dt.timedelta(minutes=15), ) -> None: """ Wait until a process is listening on the given UNIX-domain socket (default to a UNIX-domain stream socket). """ + _warn_if_numeric_duration(timeout, "wait_for_open_unix_socket") nc_flags = [ "-z", @@ -563,22 +635,26 @@ class BaseMachine(ABC): with self.nested( f"waiting for UNIX-domain {'datagram' if is_datagram else 'stream'} on '{addr}'" ): - retry(socket_is_open, timeout) + retry(socket_is_open, as_timedelta(timeout)) def wait_for_closed_port( - self, port: int, addr: str = "localhost", timeout: int = 900 + self, + port: int, + addr: str = "localhost", + timeout: Duration = dt.timedelta(minutes=15), ) -> None: """ Wait until nobody is listening on the given TCP port and IP address (default `localhost`). """ + _warn_if_numeric_duration(timeout, "wait_for_closed_port") def port_is_closed(_last_try: bool) -> bool: status, _ = self.execute(f"nc -z {addr} {port}") return status != 0 with self.nested(f"waiting for TCP port {port} on {addr} to be closed"): - retry(port_is_closed, timeout) + retry(port_is_closed, as_timedelta(timeout)) def start_job(self, jobname: str, user: str | None = None) -> tuple[int, str]: """ @@ -597,7 +673,7 @@ class BaseMachine(ABC): command: str, check_return: bool = True, check_output: bool = True, - timeout: int | None = 900, + timeout: Duration | None = dt.timedelta(minutes=15), ) -> tuple[int, str]: """ Execute a shell command, returning a list `(status, stdout)`. @@ -627,16 +703,23 @@ class BaseMachine(ABC): the machine and would therefore break the pipe that would be used for retrieving the return code. - A timeout for the command can be specified (in seconds) using the optional - `timeout` parameter, e.g., `execute(cmd, timeout=10)` or - `execute(cmd, timeout=None)`. The default is 900 seconds. + Has a default timeout of 15 minutes which can be modified. + Example: + ```py + import datetime as dt + + execute(cmd, timeout=dt.timedelta(seconds=10)) + execute(cmd, timeout=None) + ``` + A bare number is still accepted and interpreted as a number of seconds. """ + _warn_if_numeric_duration(timeout, "execute") self.run_callbacks() return self._execute( command=command, check_return=check_return, check_output=check_output, - timeout=timeout, + timeout=as_timedelta(timeout) if timeout is not None else None, ) @abstractmethod @@ -645,7 +728,7 @@ class BaseMachine(ABC): command: str, check_return: bool = True, check_output: bool = True, - timeout: int | None = 900, + timeout: dt.timedelta | None = dt.timedelta(minutes=15), ) -> tuple[int, str]: ... def run_callbacks(self) -> None: @@ -861,10 +944,13 @@ class QemuMachine(BaseMachine): ) 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: Duration = dt.timedelta(minutes=15) + ) -> None: """Wait until the visible output on the chosen TTY matches regular expression. Throws an exception on timeout. """ + _warn_if_numeric_duration(timeout, "wait_until_tty_matches") matcher = re.compile(regexp) def tty_matches(last_try: bool) -> bool: @@ -877,7 +963,7 @@ class QemuMachine(BaseMachine): return len(matcher.findall(text)) > 0 with self.nested(f"waiting for {regexp} to appear on tty {tty}"): - retry(tty_matches, timeout) + retry(tty_matches, as_timedelta(timeout)) def dump_tty_contents(self, tty: str) -> None: """Debugging: Dump the contents of the TTY""" @@ -888,7 +974,7 @@ class QemuMachine(BaseMachine): command: str, check_return: bool = True, check_output: bool = True, - timeout: int | None = 900, + timeout: dt.timedelta | None = dt.timedelta(minutes=15), ) -> tuple[int, str]: self.connect() @@ -897,7 +983,7 @@ class QemuMachine(BaseMachine): timeout_str = "" if timeout is not None: - timeout_str = f"timeout {timeout}" + timeout_str = f"timeout {timeout.total_seconds()}" # While sh is bash on NixOS, this is not the case for every distro. # We explicitly call bash here to allow for the driver to boot other distros as well. @@ -989,7 +1075,9 @@ class QemuMachine(BaseMachine): self.connected = False def wait_for_qmp_event( - self, event_filter: Callable[[dict[str, Any]], bool], timeout: int = 60 * 10 + self, + event_filter: Callable[[dict[str, Any]], bool], + timeout: Duration = dt.timedelta(minutes=10), ) -> dict[str, Any]: """ Wait for a QMP event which you can filter with the `event_filter` function. @@ -999,56 +1087,64 @@ class QemuMachine(BaseMachine): It will skip all events received in the meantime, if you want to keep them, you have to do the bookkeeping yourself and store them somewhere. - By default, it will wait up to 10 minutes, `timeout` is in seconds. + By default, it will wait up to 10 minutes. """ + _warn_if_numeric_duration(timeout, "wait_for_qmp_event") if self.qmp_client is None: raise RuntimeError("QMP API is not ready yet, is the VM ready?") - start = time.time() + timeout = as_timedelta(timeout) + start = time.monotonic() while True: evt = self.qmp_client.wait_for_event(timeout=timeout) if event_filter(evt): return evt - elapsed = time.time() - start + elapsed = dt.timedelta(seconds=time.monotonic() - start) if elapsed >= timeout: raise TimeoutError - def send_chars(self, chars: str, delay: float | None = 0.01) -> None: + def send_chars( + self, chars: str, delay: Duration | None = dt.timedelta(milliseconds=10) + ) -> None: r""" Simulate typing a sequence of characters on the virtual keyboard, e.g., `send_chars("foobar\n")` will type the string `foobar` followed by the Enter key. """ + _warn_if_numeric_duration(delay, "send_chars") with self.nested(f"sending keys {repr(chars)}"): 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: Duration = dt.timedelta(minutes=15) + ) -> None: """ Waits until the file exists in the machine's file system. """ + _warn_if_numeric_duration(timeout, "wait_for_file") def check_file(_last_try: bool) -> bool: status, _ = self.execute(f"test -e {filename}") return status == 0 with self.nested(f"waiting for file '{filename}'"): - retry(check_file, timeout) + retry(check_file, as_timedelta(timeout)) def connect(self) -> None: """ Wait for a connection to the guest root shell """ - def shell_ready(timeout_secs: int) -> bool: + def shell_ready(timeout: dt.timedelta) -> bool: """We sent some data from the backdoor service running on the guest to indicate that the backdoor shell is ready. As soon as we read some data from the socket here, we assume that our root shell is operational. """ assert self.shell - (ready, _, _) = select.select([self.shell], [], [], timeout_secs) + (ready, _, _) = select.select([self.shell], [], [], timeout.total_seconds()) return bool(ready) if self.connected: @@ -1062,7 +1158,7 @@ class QemuMachine(BaseMachine): tic = time.time() for _ in range(10): - if shell_ready(timeout_secs=30): + if shell_ready(timeout=dt.timedelta(seconds=30)): break self.log("Guest root shell did not produce any data yet...") self.log( @@ -1147,7 +1243,9 @@ class QemuMachine(BaseMachine): with self._managed_screenshot() as screenshot_path: return perform_ocr_on_screenshot(screenshot_path) - def wait_for_text(self, regex: str, timeout: int = 900) -> None: + def wait_for_text( + self, regex: str, timeout: Duration = dt.timedelta(minutes=15) + ) -> None: """ Wait until the supplied regular expressions matches the textual contents of the screen by using optical character recognition (see @@ -1157,6 +1255,7 @@ class QemuMachine(BaseMachine): This requires [`enableOCR`](#test-opt-enableOCR) to be set to `true`. ::: """ + _warn_if_numeric_duration(timeout, "wait_for_text") def screen_matches(last_try: bool) -> bool: variants = self.get_screen_text_variants() @@ -1170,9 +1269,11 @@ class QemuMachine(BaseMachine): return False with self.nested(f"waiting for {regex} to appear on screen"): - retry(screen_matches, timeout) + retry(screen_matches, as_timedelta(timeout)) - def wait_for_console_text(self, regex: str, timeout: int | None = None) -> None: + def wait_for_console_text( + self, regex: str, timeout: Duration | None = None + ) -> None: """ Wait until the supplied regular expressions match a line of the serial console output. @@ -1180,6 +1281,7 @@ class QemuMachine(BaseMachine): When this method returns, the console output that includes the match has already become part of get_console_log(). """ + _warn_if_numeric_duration(timeout, "wait_for_console_text") # Buffer the console output, this is needed # to match multiline regexes. console = io.StringIO() @@ -1200,7 +1302,7 @@ class QemuMachine(BaseMachine): with self.nested(f"waiting for {regex} to appear on console"): if timeout is not None: - retry(console_matches, timeout) + retry(console_matches, as_timedelta(timeout)) else: console_matches(False, block=True) @@ -1212,7 +1314,10 @@ class QemuMachine(BaseMachine): return "\n".join(self.full_console_log) def send_key( - self, key: str, delay: float | None = 0.01, log: bool | None = True + self, + key: str, + delay: Duration | None = dt.timedelta(milliseconds=10), + log: bool | None = True, ) -> None: """ Simulate pressing keys on the virtual keyboard, e.g., @@ -1221,12 +1326,13 @@ class QemuMachine(BaseMachine): Please also refer to the QEMU documentation for more information on the input syntax: https://en.wikibooks.org/wiki/QEMU/Monitor#sendkey_keys """ + _warn_if_numeric_duration(delay, "send_key") key = CHAR_TO_KEY.get(key, key) context = self.nested(f"sending key {repr(key)}") if log else nullcontext() with context: self.send_monitor_command(f"sendkey {key}") if delay is not None: - time.sleep(delay) + time.sleep(as_seconds(delay)) def send_console(self, chars: str) -> None: r""" @@ -1355,10 +1461,11 @@ class QemuMachine(BaseMachine): self.send_key("ctrl-alt-delete") self.connected = False - def wait_for_x(self, timeout: int = 900) -> None: + def wait_for_x(self, timeout: Duration = dt.timedelta(minutes=15)) -> None: """ Wait until it is possible to connect to the X server. """ + _warn_if_numeric_duration(timeout, "wait_for_x") def check_x(_last_try: bool) -> bool: cmd = ( @@ -1372,18 +1479,21 @@ class QemuMachine(BaseMachine): return status == 0 with self.nested("waiting for the X11 server"): - retry(check_x, timeout) + retry(check_x, as_timedelta(timeout)) def get_window_names(self) -> list[str]: return self.succeed( 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: Duration = dt.timedelta(minutes=15) + ) -> None: """ Wait until an X11 window has appeared whose name matches the given regular expression, e.g., `wait_for_window("Terminal")`. """ + _warn_if_numeric_duration(timeout, "wait_for_window") pattern = re.compile(regexp) def window_is_visible(last_try: bool) -> bool: @@ -1397,7 +1507,7 @@ class QemuMachine(BaseMachine): return any(pattern.search(name) for name in names) with self.nested("waiting for a window to appear"): - retry(window_is_visible, timeout) + retry(window_is_visible, as_timedelta(timeout)) def forward_port(self, host_port: int = 8080, guest_port: int = 80) -> None: """ @@ -1647,7 +1757,7 @@ class NspawnMachine(BaseMachine): command: str, check_return: bool = True, check_output: bool = True, - timeout: int | None = 900, + timeout: dt.timedelta | None = dt.timedelta(minutes=15), ) -> tuple[int, str]: self.start() @@ -1679,7 +1789,7 @@ class NspawnMachine(BaseMachine): command, ], env={}, - timeout=timeout, + timeout=timeout.total_seconds() if timeout is not None else None, stdout=subprocess.PIPE, text=True, ) diff --git a/nixos/lib/test-driver/src/test_driver/machine/qmp.py b/nixos/lib/test-driver/src/test_driver/machine/qmp.py index 99c02ca1c120..f8d2b4e887d8 100644 --- a/nixos/lib/test-driver/src/test_driver/machine/qmp.py +++ b/nixos/lib/test-driver/src/test_driver/machine/qmp.py @@ -1,3 +1,4 @@ +import datetime as dt import json import logging import os @@ -74,15 +75,19 @@ class QMPSession: else: raise QMPAPIError(evt_or_result) - def wait_for_event(self, timeout: int = 10) -> dict[str, Any]: + def wait_for_event( + self, timeout: dt.timedelta = dt.timedelta(seconds=10) + ) -> dict[str, Any]: while self.pending_events.empty(): self.read_pending_messages() - return self.pending_events.get(timeout=timeout) + return self.pending_events.get(timeout=timeout.total_seconds()) - def events(self, timeout: int = 10) -> Iterator[dict[str, Any]]: + def events( + self, timeout: dt.timedelta = dt.timedelta(seconds=10) + ) -> Iterator[dict[str, Any]]: while not self.pending_events.empty(): - yield self.pending_events.get(timeout=timeout) + yield self.pending_events.get(timeout=timeout.total_seconds()) def send(self, cmd: str, args: dict[str, str] = {}) -> dict[str, str]: self.read_pending_messages() diff --git a/nixos/lib/test-driver/src/test_driver/polling_condition.py b/nixos/lib/test-driver/src/test_driver/polling_condition.py index bebdeca29702..0cf18bd9fe54 100644 --- a/nixos/lib/test-driver/src/test_driver/polling_condition.py +++ b/nixos/lib/test-driver/src/test_driver/polling_condition.py @@ -1,7 +1,10 @@ +import datetime as dt import time +import warnings from collections.abc import Callable from math import isfinite +from test_driver.duration import as_timedelta from test_driver.logger import AbstractLogger @@ -11,7 +14,7 @@ class PollingConditionError(Exception): class PollingCondition: condition: Callable[[], bool] - seconds_interval: float + interval: dt.timedelta description: str | None logger: AbstractLogger @@ -22,11 +25,23 @@ class PollingCondition: self, condition: Callable[[], bool | None], logger: AbstractLogger, - seconds_interval: float = 2.0, + seconds_interval: float | None = None, description: str | None = None, + *, + interval: dt.timedelta | None = None, ): + if seconds_interval is not None: + if interval is not None: + raise TypeError( + "PollingCondition() got both 'interval' and 'seconds_interval' arguments. Pass only 'interval'" + ) + warnings.warn( + "PollingCondition(): The 'seconds_interval' argument is deprecated. Use 'interval' instead.", + ) + interval = as_timedelta(seconds_interval) + self.condition = condition # ty: ignore[invalid-assignment] - self.seconds_interval = seconds_interval + self.interval = interval if interval is not None else dt.timedelta(seconds=2) self.logger = logger if description is None: @@ -78,7 +93,11 @@ class PollingCondition: @property def overdue(self) -> bool: - return self.last_called + self.seconds_interval < time.monotonic() + if not isfinite(self.last_called): + # `last_called` is `-inf` until the condition has run at least once. + return True + time_since_last = dt.timedelta(seconds=time.monotonic() - self.last_called) + return time_since_last > self.interval @property def entered(self) -> bool: diff --git a/nixos/lib/test-script-prepend.py b/nixos/lib/test-script-prepend.py index 39f193b21c5a..c9792558f4bd 100644 --- a/nixos/lib/test-script-prepend.py +++ b/nixos/lib/test-script-prepend.py @@ -1,6 +1,7 @@ # This file contains type hints that can be prepended to Nix test scripts so they can be type # checked. +import datetime as dt from contextlib import contextmanager from typing import Any, Callable, ContextManager, Generator, List, Optional, Union from unittest import TestCase @@ -32,10 +33,11 @@ class CreateMachineProtocol(Protocol): class PollingConditionProtocol(Protocol): def __call__( self, - fun_: Optional[Callable] = None, + fun_: Callable | None = None, *, - seconds_interval: float = 2.0, - description: Optional[str] = None, + interval: dt.timedelta | None = None, + seconds_interval: float | None = None, + description: str | None = None, ) -> Union[Callable[[Callable], ContextManager], ContextManager]: raise Exception("This is just type information for the Nix test driver") @@ -98,12 +100,20 @@ log: AbstractLogger = CompositeLogger([]) def polling_condition( - fun: Callable | None, seconds_interval: float = 0.0, description: str | None = None + fun_: Callable | None = None, + *, + interval: dt.timedelta | None = None, + seconds_interval: float | None = None, + description: str | None = None, ): pass -def retry(fn: Callable, timeout_seconds: int = 900) -> None: +def retry( + fn: Callable, + timeout: dt.timedelta | None = None, + timeout_seconds: int | None = None, +) -> None: pass