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

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

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

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:

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

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")
   '';
 }

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:

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",