From 84fe17e86953dfb45b7b20b6eddc842fda562080 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kier=C3=A1n=20Meinhardt?= Date: Tue, 24 Mar 2026 15:37:39 +0100 Subject: [PATCH] nixos/tests: fix machine type hints With the addition of test driver machines based on systemd-nspawn, the `Machine` class was superseded by an abstract base class `BaseMachine` and a subclass `QemuMachine` for conventional QEMU-based test nodes. This commit fixes the usage of the `Machine` class in type annotations across our tests. It also makes the tests use the more specific `machines_qemu` variable if QEMU-specific features are required from the machines. (`machines` provides only the methods shared by both `NspawnMachine` and `Qemumachine`.) --- nixos/tests/garage/common.nix | 12 ++++++------ nixos/tests/incus/incus_machine.py | 2 +- nixos/tests/networking-proxy.nix | 2 +- nixos/tests/pacemaker.nix | 6 +++--- .../tests/password-option-override-ordering.nix | 16 ++++++++-------- nixos/tests/rtkit.nix | 8 ++++---- nixos/tests/systemd-journal-upload.nix | 2 +- 7 files changed, 24 insertions(+), 24 deletions(-) diff --git a/nixos/tests/garage/common.nix b/nixos/tests/garage/common.nix index c8392eee6dba..5928cc3c55a6 100644 --- a/nixos/tests/garage/common.nix +++ b/nixos/tests/garage/common.nix @@ -21,14 +21,14 @@ node_id: str host: str - def get_node_fqn(machine: Machine) -> GarageNode: + def get_node_fqn(machine: BaseMachine) -> GarageNode: node_id, host = machine.succeed("garage node id").split('@') return GarageNode(node_id=node_id, host=host) - def get_node_id(machine: Machine) -> str: + def get_node_id(machine: BaseMachine) -> str: return get_node_fqn(machine).node_id - def get_layout_version(machine: Machine) -> int: + def get_layout_version(machine: BaseMachine) -> int: version_data = machine.succeed("garage layout show") m = cur_version_regex.search(version_data) if m and m.group('ver') is not None: @@ -36,17 +36,17 @@ else: raise ValueError('Cannot find current layout version') - def apply_garage_layout(machine: Machine, layouts: List[str]): + def apply_garage_layout(machine: BaseMachine, layouts: List[str]): for layout in layouts: machine.succeed(f"garage layout assign {layout}") version = get_layout_version(machine) machine.succeed(f"garage layout apply --version {version}") - def create_api_key(machine: Machine, key_name: str) -> S3Key: + def create_api_key(machine: BaseMachine, key_name: str) -> S3Key: output = machine.succeed(f"garage key create {key_name}") return parse_api_key_data(output) - def get_api_key(machine: Machine, key_pattern: str) -> S3Key: + def get_api_key(machine: BaseMachine, key_pattern: str) -> S3Key: output = machine.succeed(f"garage key info {key_pattern}") return parse_api_key_data(output) diff --git a/nixos/tests/incus/incus_machine.py b/nixos/tests/incus/incus_machine.py index c25a7804b871..eeeb0bdecf4c 100644 --- a/nixos/tests/incus/incus_machine.py +++ b/nixos/tests/incus/incus_machine.py @@ -1,7 +1,7 @@ import json -class IncusHost(Machine): +class IncusHost(QemuMachine): def __init__(self, base): with subtest("Wait for startup"): base.wait_for_unit("incus.service") diff --git a/nixos/tests/networking-proxy.nix b/nixos/tests/networking-proxy.nix index 2f8a64b5affa..ec784f2bd657 100644 --- a/nixos/tests/networking-proxy.nix +++ b/nixos/tests/networking-proxy.nix @@ -72,7 +72,7 @@ in from typing import Dict, Optional - def get_machine_env(machine: Machine, user: Optional[str] = None) -> Dict[str, str]: + def get_machine_env(machine: BaseMachine, user: Optional[str] = None) -> Dict[str, str]: """ Gets the environment from a given machine, and returns it as a dictionary in the form: diff --git a/nixos/tests/pacemaker.nix b/nixos/tests/pacemaker.nix index 826c4be7df14..8023d6f14bc8 100644 --- a/nixos/tests/pacemaker.nix +++ b/nixos/tests/pacemaker.nix @@ -86,7 +86,7 @@ rec { output = node1.succeed("crm_resource -r cat --locate") match = re.search("is running on: (.+)", output) if match: - for machine in machines: + for machine in machines_qemu: if machine.name == match.group(1): current_node = machine break @@ -96,7 +96,7 @@ rec { current_node.crash() # pick another node that's still up - for machine in machines: + for machine in machines_qemu: if machine.booted: check_node = machine # find where the service has been started next @@ -105,7 +105,7 @@ rec { match = re.search("is running on: (.+)", output) # output will remain the old current_node until the crash is detected by pacemaker if match and match.group(1) != current_node.name: - for machine in machines: + for machine in machines_qemu: if machine.name == match.group(1): next_node = machine break diff --git a/nixos/tests/password-option-override-ordering.nix b/nixos/tests/password-option-override-ordering.nix index caa0246735ca..71c449064a0b 100644 --- a/nixos/tests/password-option-override-ordering.nix +++ b/nixos/tests/password-option-override-ordering.nix @@ -112,22 +112,22 @@ in assert stored_hash == pass_hash, f"{username} user password does not match" with subtest("alice user has correct password"): - for machine in machines: + for machine in machines_qemu: assert_password_sha512crypt_match(machine, "alice", "${password1}") assert "${hashed_sha512crypt}" not in machine.succeed("getent shadow alice"), f"{machine}: alice user password is not correct" with subtest("bob user has correct password"): - for machine in machines: + for machine in machines_qemu: print(machine.succeed("getent shadow bob")) assert "${hashed_bcrypt}" in machine.succeed("getent shadow bob"), f"{machine}: bob user password is not correct" with subtest("cat user has correct password"): - for machine in machines: + for machine in machines_qemu: print(machine.succeed("getent shadow cat")) assert "${hashed_bcrypt}" in machine.succeed("getent shadow cat"), f"{machine}: cat user password is not correct" with subtest("dan user has correct password"): - for machine in machines: + for machine in machines_qemu: print(machine.succeed("getent shadow dan")) assert "${hashed_bcrypt}" in machine.succeed("getent shadow dan"), f"{machine}: dan user password is not correct" @@ -138,11 +138,11 @@ in assert_password_sha512crypt_match(immutable, "greg", "${password1}") assert "${hashed_sha512crypt}" not in immutable.succeed("getent shadow greg"), "greg user password is not correct" - for machine in machines: + for machine in machines_qemu: machine.wait_for_unit("multi-user.target") machine.wait_until_succeeds("pgrep -f 'agetty.*tty1'") - def check_login(machine: Machine, tty_number: str, username: str, password: str): + def check_login(machine: QemuMachine, tty_number: str, username: str, password: str): machine.send_key(f"alt-f{tty_number}") machine.wait_until_succeeds(f"[ $(fgconsole) = {tty_number} ]") machine.wait_for_unit(f"getty@tty{tty_number}.service") @@ -158,11 +158,11 @@ in assert username in machine.succeed(f"cat /tmp/{tty_number}"), f"{machine}: {username} password is not correct" with subtest("Test initialPassword override"): - for machine in machines: + for machine in machines_qemu: check_login(machine, "2", "egon", "${password1}") with subtest("Test initialHashedPassword override"): - for machine in machines: + for machine in machines_qemu: check_login(machine, "3", "fran", "meow") ''; } diff --git a/nixos/tests/rtkit.nix b/nixos/tests/rtkit.nix index d884a588a16e..74bf69e02a83 100644 --- a/nixos/tests/rtkit.nix +++ b/nixos/tests/rtkit.nix @@ -103,7 +103,7 @@ Result = namedtuple("Result", ["command", "machine", "status", "out", "value"]) Value = namedtuple("Value", ["type", "data"]) - def busctl(node: Machine, *args: Any, user: Optional[str] = None) -> Result: + def busctl(node: BaseMachine, *args: Any, user: Optional[str] = None) -> Result: command = f"busctl --json=short {shlex.join(map(str, args))}" if user is not None: command = f"su - {user} -c {shlex.quote(command)}" @@ -121,7 +121,7 @@ if result.status == 0: raise Exception(f"command `{result.command}` unexpectedly succeeded") - def rtkit_make_process_realtime(node: Machine, pid: int, priority: int, user: Optional[str] = None) -> Result: + def rtkit_make_process_realtime(node: BaseMachine, pid: int, priority: int, user: Optional[str] = None) -> Result: return busctl(node, "call", "org.freedesktop.RealtimeKit1", "/org/freedesktop/RealtimeKit1", "org.freedesktop.RealtimeKit1", "MakeThreadRealtimeWithPID", "ttu", pid, 0, priority, user=user) def get_max_realtime_priority() -> int: @@ -133,7 +133,7 @@ def parse_chrt(out: str, field: str) -> str: return next(map(lambda l: l.split(": ")[1], filter(lambda l: field in l, out.splitlines()))) - def get_pid(node: Machine, unit: str, user: Optional[str] = None) -> int: + def get_pid(node: BaseMachine, unit: str, user: Optional[str] = None) -> int: node.wait_for_unit(unit, user=user) (status, out) = node.systemctl(f"show -P MainPID {unit}", user=user) if status == 0: @@ -142,7 +142,7 @@ node.log(out) raise Exception(f"unable to determine MainPID of {unit} (systemctl exit code {status})") - def assert_sched(node: Machine, pid: int, policy: Optional[str] = None, priority: Optional[int] = None): + def assert_sched(node: BaseMachine, pid: int, policy: Optional[str] = None, priority: Optional[int] = None): out = node.succeed(f"chrt -p {pid}") node.log(out) if policy is not None: diff --git a/nixos/tests/systemd-journal-upload.nix b/nixos/tests/systemd-journal-upload.nix index 00dbba82b88b..af7817bd97d8 100644 --- a/nixos/tests/systemd-journal-upload.nix +++ b/nixos/tests/systemd-journal-upload.nix @@ -72,7 +72,7 @@ server.wait_for_unit("multi-user.target") client.wait_for_unit("multi-user.target") - def copy_pems(machine: Machine, domain: str): + def copy_pems(machine: BaseMachine, domain: str): machine.succeed("mkdir /run/secrets") machine.copy_from_host( source=f"{tmpdir}/{domain}/cert.pem",