test(discovery): add provisioning drivers

Discovery/tests/provision/__init__.py

@@ -0,0 +1 @@
+"""Guest lifecycle. Nothing above this package knows which hypervisor is in play."""

Discovery/tests/provision/driver.py

@@ -0,0 +1,113 @@
+"""The guest interface, and the driver that needs no guest.
+
+One interface, three hypervisors. Keeping the Linux phase behind this contract
+is what stops it hard-coding assumptions that macOS and Windows later have to
+unpick - and the dry driver is what lets the runner be tested at all before
+any image exists.
+"""
+
+from __future__ import annotations
+
+import shlex
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Protocol, Sequence, Tuple
+
+DEFAULT_TIMEOUT = 600
+
+
+@dataclass(frozen=True)
+class Result:
+    argv: Tuple[str, ...]
+    returncode: int = 0
+    stdout: str = ""
+    stderr: str = ""
+
+    @property
+    def ok(self) -> bool:
+        return self.returncode == 0
+
+    @property
+    def command(self) -> str:
+        return " ".join(shlex.quote(a) for a in self.argv)
+
+
+class Driver(Protocol):
+    """Restore, run, push, pull. Deliberately four verbs and no more."""
+
+    def restore(self) -> None:
+        """Back to the golden snapshot."""
+
+    def run(self, argv: Sequence[str], *, timeout: int = DEFAULT_TIMEOUT) -> Result:
+        """Execute in the guest."""
+
+    def push(self, local: str, remote: str) -> None:
+        """Host to guest."""
+
+    def pull(self, remote: str, local: str) -> None:
+        """Guest to host."""
+
+
+@dataclass
+class DryDriver:
+    """Records what would have happened, and runs nothing.
+
+    Every command is recorded in order so the runner's dependency ordering can
+    be asserted directly - which is the property most worth testing and the
+    one a real guest makes slowest to check.
+    """
+
+    name: str = "dry"
+    restores: int = 0
+    commands: List[Tuple[str, ...]] = field(default_factory=list)
+    pushed: List[Tuple[str, str]] = field(default_factory=list)
+    pulled: List[Tuple[str, str]] = field(default_factory=list)
+    failures: Dict[str, int] = field(default_factory=dict)
+    outputs: Dict[str, str] = field(default_factory=dict)
+    files: Dict[str, str] = field(default_factory=dict)
+
+    def restore(self) -> None:
+        self.restores += 1
+        self.commands.clear()
+
+    def run(self, argv: Sequence[str], *, timeout: int = DEFAULT_TIMEOUT) -> Result:
+        argv = tuple(str(a) for a in argv)
+        self.commands.append(argv)
+        joined = " ".join(argv)
+        for needle, code in self.failures.items():
+            if needle in joined:
+                return Result(argv, returncode=code, stderr=f"dry failure for {needle!r}")
+        return Result(argv, returncode=0, stdout=self.outputs.get(joined, ""))
+
+    def push(self, local: str, remote: str) -> None:
+        self.pushed.append((local, remote))
+
+    def pull(self, remote: str, local: str) -> None:
+        self.pulled.append((remote, local))
+
+    #: A dry guest still has a home, because paths are written relative to one.
+    home: str = "/home/dry"
+
+    def sh(self, script: str, *, timeout: int = DEFAULT_TIMEOUT) -> Result:
+        return self.run(["sh", "-lc", script], timeout=timeout)
+
+    def expand(self, path: str) -> str:
+        return self.home.rstrip("/") + path[1:] if path.startswith("~/") else path
+
+    def write(self, path: str, body: str) -> Result:
+        self.files[self.expand(path)] = body
+        return self.run(["write", self.expand(path)])
+
+    def exists(self, path: str) -> bool:
+        return self.expand(path) in self.files
+
+    def ran(self, needle: str) -> bool:
+        return any(needle in " ".join(argv) for argv in self.commands)
+
+    def index_of(self, needle: str) -> Optional[int]:
+        for position, argv in enumerate(self.commands):
+            if needle in " ".join(argv):
+                return position
+        return None
+
+
+__all__ = ["DEFAULT_TIMEOUT", "Driver", "DryDriver", "Result"]

Discovery/tests/provision/lima.py

@@ -0,0 +1,84 @@
+"""The Linux guest, driven through lima.
+
+lima is what is actually installed on this host, so it is what the Linux phase
+uses. The four verbs are the same four every driver implements; nothing above
+this file knows that lima is in play.
+
+``restore`` is the honest exception. This guest runs under vz, where
+``limactl snapshot`` is unimplemented, so restore refuses loudly rather than
+returning silently and letting a caller believe it got a clean machine.
+"""
+
+from __future__ import annotations
+
+import shlex
+import subprocess
+from dataclasses import dataclass, field
+from typing import List, Optional, Sequence, Tuple
+
+from .driver import DEFAULT_TIMEOUT, Result
+
+
+class NoSnapshot(RuntimeError):
+    """This guest cannot be returned to a known state."""
+
+
+@dataclass
+class LimaDriver:
+    """Commands run inside a lima instance over ``limactl shell``."""
+
+    instance: str = "adr-disco-linux"
+    _home: Optional[str] = field(default=None, init=False, repr=False)
+    commands: List[Tuple[str, ...]] = field(default_factory=list)
+
+    @property
+    def home(self) -> str:
+        """The guest's home, cached; manifest paths are written relative to it."""
+        if self._home is None:
+            self._home = self.run(["sh", "-lc", "printf %s \"$HOME\""]).stdout.strip() or "~"
+        return self._home
+
+    def restore(self) -> None:
+        raise NoSnapshot(
+            f"{self.instance} runs under vz, where limactl snapshot is unimplemented; "
+            "a run against it mutates the guest permanently"
+        )
+
+    def run(self, argv: Sequence[str], *, timeout: int = DEFAULT_TIMEOUT) -> Result:
+        argv = tuple(str(a) for a in argv)
+        self.commands.append(argv)
+        completed = subprocess.run(
+            ["limactl", "shell", "--workdir", "/", self.instance, *argv],
+            capture_output=True, text=True, timeout=timeout, check=False,
+        )
+        return Result(argv, completed.returncode, completed.stdout, completed.stderr)
+
+    def sh(self, script: str, *, timeout: int = DEFAULT_TIMEOUT) -> Result:
+        """A convenience for the many recipes that are really one shell line."""
+        return self.run(["sh", "-lc", script], timeout=timeout)
+
+    def push(self, local: str, remote: str) -> None:
+        subprocess.run(["limactl", "copy", local, f"{self.instance}:{remote}"], check=True)
+
+    def pull(self, remote: str, local: str) -> None:
+        subprocess.run(["limactl", "copy", f"{self.instance}:{remote}", local], check=True)
+
+    def expand(self, path: str) -> str:
+        """``~`` means the guest's home, not this Mac's."""
+        if path.startswith("~/"):
+            return self.home.rstrip("/") + path[1:]
+        return path
+
+    def write(self, path: str, body: str) -> Result:
+        """Create a file in the guest, parents and all."""
+        target = self.expand(path)
+        quoted = shlex.quote(target)
+        return self.sh(
+            f"mkdir -p \"$(dirname {quoted})\" && cat > {quoted} <<'ADR_HARNESS_EOF'\n{body}\nADR_HARNESS_EOF"
+        )
+
+    def exists(self, path: str) -> bool:
+        return self.sh(f"test -e {shlex.quote(self.expand(path))}").ok
+
+
+__all__ = ["LimaDriver", "NoSnapshot"]

Discovery/tests/provision/tart.py

@@ -0,0 +1,108 @@
+"""The macOS guest, driven through tart over SSH.
+
+macOS guests must run on Apple hardware, so this driver only ever works on an
+Apple-silicon host. SSH rather than anything Mac-specific, so the guest is
+driven with the same verbs, the same quoting and the same recipe code as the
+Linux one.
+"""
+
+from __future__ import annotations
+
+import shlex
+import subprocess
+from dataclasses import dataclass, field
+from typing import List, Optional, Sequence, Tuple
+
+from .driver import DEFAULT_TIMEOUT, Result
+
+SSH_OPTIONS = (
+    "-o", "StrictHostKeyChecking=no",
+    "-o", "UserKnownHostsFile=/dev/null",
+    "-o", "BatchMode=yes",
+    "-o", "ConnectTimeout=10",
+    "-o", "LogLevel=ERROR",
+)
+
+
+class NoSnapshot(RuntimeError):
+    """This guest cannot be returned to a known state."""
+
+
+@dataclass
+class TartDriver:
+    """Commands run inside a tart VM over SSH."""
+
+    vm: str = "adr-macos"
+    user: str = "admin"
+    _ip: Optional[str] = field(default=None, init=False, repr=False)
+    _home: Optional[str] = field(default=None, init=False, repr=False)
+    commands: List[Tuple[str, ...]] = field(default_factory=list)
+
+    @property
+    def ip(self) -> str:
+        if self._ip is None:
+            found = subprocess.run(["tart", "ip", self.vm], capture_output=True,
+                                   text=True, check=True)
+            self._ip = found.stdout.strip()
+        return self._ip
+
+    @property
+    def home(self) -> str:
+        if self._home is None:
+            self._home = self.sh('printf %s "$HOME"').stdout.strip() or f"/Users/{self.user}"
+        return self._home
+
+    def restore(self) -> None:
+        """`tart clone` from a golden VM is the restore path, and it is not wired up.
+
+        Refused rather than silently skipped: a caller that believes it has a
+        clean machine and does not will report the previous run's leftovers as
+        this run's findings.
+        """
+        raise NoSnapshot(
+            f"no golden clone configured for {self.vm}; restore would leave the "
+            "guest carrying whatever the last run installed"
+        )
+
+    def run(self, argv: Sequence[str], *, timeout: int = DEFAULT_TIMEOUT) -> Result:
+        argv = tuple(str(a) for a in argv)
+        self.commands.append(argv)
+        remote = " ".join(shlex.quote(a) for a in argv)
+        completed = subprocess.run(
+            ["ssh", *SSH_OPTIONS, f"{self.user}@{self.ip}", remote],
+            capture_output=True, text=True, timeout=timeout, check=False,
+        )
+        return Result(argv, completed.returncode, completed.stdout, completed.stderr)
+
+    #: A login shell here picks up neither Homebrew nor pipx's bin directory, so
+    #: tools installed with either are invisible to every command the harness
+    #: runs - and would be reported missing when they are sitting right there.
+    #: A real user's shell has both, so the harness gives itself the same one.
+    PRELUDE = ('if [ -x /opt/homebrew/bin/brew ]; then eval "$(/opt/homebrew/bin/brew shellenv)"; fi; '
+               'PATH="$HOME/.local/bin:$PATH"; export PATH; ')
+
+    def sh(self, script: str, *, timeout: int = DEFAULT_TIMEOUT) -> Result:
+        return self.run(["/bin/sh", "-lc", self.PRELUDE + script], timeout=timeout)
+
+    def push(self, local: str, remote: str) -> None:
+        subprocess.run(["scp", *SSH_OPTIONS, local, f"{self.user}@{self.ip}:{remote}"], check=True)
+
+    def pull(self, remote: str, local: str) -> None:
+        subprocess.run(["scp", *SSH_OPTIONS, f"{self.user}@{self.ip}:{remote}", local], check=True)
+
+    def expand(self, path: str) -> str:
+        if path.startswith("~/"):
+            return self.home.rstrip("/") + path[1:]
+        return path
+
+    def write(self, path: str, body: str) -> Result:
+        target = shlex.quote(self.expand(path))
+        return self.sh(
+            f"mkdir -p \"$(dirname {target})\" && cat > {target} <<'ADR_HARNESS_EOF'\n{body}\nADR_HARNESS_EOF"
+        )
+
+    def exists(self, path: str) -> bool:
+        return self.sh(f"test -e {shlex.quote(self.expand(path))}").ok
+
+
+__all__ = ["NoSnapshot", "SSH_OPTIONS", "TartDriver"]