nixos/test-driver: use timedelta for timeouts and durations

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")
 ```

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:

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

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.",
+        )

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<n>"""
@@ -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,
         )

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()

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:

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