test(a5): add pytest regression tests for AGENTS.md errors #9-#12

12 dedicated tests covering: (#9) clone_linked surfaces 'should not be powered on' as typed BackendError; (#10) is_running uses 'vmrun list' only and never falls back to getGuestIPAddress (incl. on vmrun list failure); (#11) two consecutive 'vm new' invocations with identical inputs MUST produce distinct clone names/destinations (timestamp suffix), guarding the orchestrator-side safety net against the machine-id duplicate-IP failure mode; (#12) SshTransport defaults to paramiko AutoAddPolicy with no known_hosts, never shells out to ssh-keygen/ssh.exe/scp.exe, wraps paramiko import/exec/SFTP errors as TransportConnectError. Coverage: 80.10% (gate 80%).

tests/python/test_agents_errors.py

@@ -0,0 +1,489 @@
+"""Regression tests for ``AGENTS.md`` "Errori frequenti da evitare" #9-#12.
+
+Each test is a guard against re-introducing a class of bug that already
+caused incidents in the PowerShell stack and that the Python rewrite must
+keep fixed. If you find yourself relaxing one of these tests, update
+``AGENTS.md`` first and explain why the constraint no longer applies.
+
+Cross-reference:
+
+* ``AGENTS.md`` §"Errori frequenti da evitare", entries 9-12.
+* ``plans/implementation-plan-A-B.md`` step A5.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import types
+from pathlib import Path
+from typing import Any
+
+import pytest
+from click.testing import CliRunner
+
+from ci_orchestrator.backends.errors import BackendError, BackendOperationFailed
+from ci_orchestrator.backends.protocol import VmHandle
+from ci_orchestrator.backends.workstation import WorkstationVmrunBackend
+from ci_orchestrator.commands.vm import vm as vm_group
+from ci_orchestrator.transport.ssh import SshTransport
+
+# ────────────────────────────────────────────────────────────────────── helpers
+
+
+def _completed(returncode: int = 0, stdout: str = "", stderr: str = "") -> Any:
+    return subprocess.CompletedProcess(
+        args=[], returncode=returncode, stdout=stdout, stderr=stderr
+    )
+
+
+@pytest.fixture
+def fake_vmrun(tmp_path: Path) -> str:
+    p = tmp_path / "vmrun.exe"
+    p.write_bytes(b"")
+    return str(p)
+
+
+# ──────────────────────────────────────────────── AGENTS.md #9 — powered-on snapshot
+
+
+class TestError9PoweredOnSnapshot:
+    """``vmrun clone ... linked -snapshot=<name>`` rejects a snapshot
+    captured with the template VM still powered on (memory ``*.vmem`` files
+    present). The backend MUST surface that error as a typed
+    :class:`BackendError` with the original vmrun message preserved so the
+    operator can act (re-take the snapshot from a fully powered-off VM).
+    """
+
+    _POWERED_ON_STDERR = (
+        "Error: The virtual machine should not be powered on. "
+        "It is already running.\n"
+    )
+
+    def test_clone_linked_propagates_powered_on_error_typed(
+        self, monkeypatch: pytest.MonkeyPatch, fake_vmrun: str
+    ) -> None:
+        monkeypatch.setattr(
+            subprocess,
+            "run",
+            lambda *a, **kw: _completed(255, stderr=self._POWERED_ON_STDERR),
+        )
+        backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
+
+        with pytest.raises(BackendOperationFailed) as exc_info:
+            backend.clone_linked(
+                template=r"C:\tpl\WinBuild2025\WinBuild2025.vmx",
+                snapshot="BaseClean",
+                name="job-9",
+                destination=r"C:\bvm\job-9\job-9.vmx",
+            )
+
+        # Must be a BackendError subclass (typed, not a bare RuntimeError).
+        assert isinstance(exc_info.value, BackendError)
+        # Operation name must match so callers can pattern-match.
+        assert exc_info.value.operation == "clone"
+        assert exc_info.value.returncode == 255
+        # Original vmrun message must be preserved verbatim for triage.
+        assert "should not be powered on" in exc_info.value.output
+
+
+# ───────────────────────────────────────────── AGENTS.md #10 — is_running via vmrun list
+
+
+class TestError10IsRunningUsesVmrunList:
+    """``vmrun getGuestIPAddress`` blocks 30-60s when VMware Tools are not
+    yet up; using it as a "VM running?" probe makes ``is_running`` appear
+    hung. The backend MUST use ``vmrun list`` parsing instead.
+    """
+
+    def test_is_running_calls_vmrun_list_only(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str
+    ) -> None:
+        vmx = tmp_path / "job10" / "job10.vmx"
+        vmx.parent.mkdir()
+        vmx.write_text("")
+
+        captured: list[list[str]] = []
+
+        def fake_run(cmd: list[str], **_kwargs: object) -> Any:
+            captured.append(cmd)
+            assert cmd[3] == "list", (
+                f"is_running invoked unexpected vmrun op: {cmd[3]!r} "
+                "(MUST be 'list', see AGENTS.md error #10)"
+            )
+            return _completed(0, stdout=f"Total running VMs: 1\n{vmx}\n")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
+
+        assert backend.is_running(VmHandle(str(vmx))) is True
+
+        # Exactly one subprocess call, and its operation MUST be 'list'.
+        assert len(captured) == 1
+        ops = [cmd[3] for cmd in captured]
+        assert ops == ["list"]
+        # Hard guard: no getGuestIPAddress invocation, ever.
+        for cmd in captured:
+            assert "getGuestIPAddress" not in cmd
+
+    def test_is_running_false_when_not_listed(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str
+    ) -> None:
+        vmx = tmp_path / "absent.vmx"
+        vmx.write_text("")
+
+        captured: list[str] = []
+
+        def fake_run(cmd: list[str], **_kwargs: object) -> Any:
+            captured.append(cmd[3])
+            return _completed(0, stdout="Total running VMs: 0\n")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
+        assert backend.is_running(VmHandle(str(vmx))) is False
+        assert captured == ["list"]
+        assert "getGuestIPAddress" not in captured
+
+    def test_is_running_false_on_vmrun_failure_without_getip_fallback(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str
+    ) -> None:
+        """Even on ``vmrun list`` failure, the backend MUST NOT fall back
+        to ``getGuestIPAddress`` (which would block on Tools)."""
+        vmx = tmp_path / "absent.vmx"
+        vmx.write_text("")
+
+        captured: list[str] = []
+
+        def fake_run(cmd: list[str], **_kwargs: object) -> Any:
+            captured.append(cmd[3])
+            return _completed(1, stderr="vmrun broke")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
+        assert backend.is_running(VmHandle(str(vmx))) is False
+        assert captured == ["list"]
+        assert "getGuestIPAddress" not in captured
+
+
+# ──────────────────────────────────────── AGENTS.md #11 — machine-id / unique clones
+
+
+class TestError11UniqueCloneNames:
+    """Ubuntu cloud images ship with a fixed ``/etc/machine-id``. Two
+    clones that share machine-id collide on the VMware DHCP lease and
+    deadlock on duplicate IPs. The template-side fix lives in
+    :file:`template/Prepare-LinuxBuild2404.ps1` (``truncate -s 0
+    /etc/machine-id`` before snapshot).
+
+    On the orchestrator side, the safety net is that ``vm new`` MUST never
+    reuse a clone identifier — otherwise even a correct template would
+    end up cloning into an existing VMX directory and the duplicate-IP
+    failure mode would re-appear from a different angle. We assert that
+    two ``vm new`` invocations with identical inputs produce distinct
+    destination paths (timestamp-suffixed clone names).
+    """
+
+    def _invoke_vm_new(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+        fake_vmrun: str,
+    ) -> tuple[CliRunner, list[dict[str, str]]]:
+        # Ensure load_config has env keys it expects (no .toml present).
+        for k in ("CI_ROOT", "CI_TEMPLATES", "CI_BUILD_VMS", "CI_ARTIFACTS", "CI_KEYS"):
+            monkeypatch.delenv(k, raising=False)
+
+        template = tmp_path / "tpl" / "tpl.vmx"
+        template.parent.mkdir()
+        template.write_text("")
+
+        clone_base = tmp_path / "clones"
+
+        calls: list[dict[str, str]] = []
+
+        def fake_clone(
+            self: WorkstationVmrunBackend,
+            template: str,
+            snapshot: str,
+            name: str,
+            destination: str | None = None,
+        ) -> VmHandle:
+            assert destination is not None
+            # Materialise the clone VMX so vm_new's post-condition passes.
+            dest = Path(destination)
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            dest.write_text("")
+            calls.append(
+                {"name": name, "destination": destination, "snapshot": snapshot}
+            )
+            return VmHandle(identifier=destination, name=name)
+
+        monkeypatch.setattr(
+            WorkstationVmrunBackend, "clone_linked", fake_clone, raising=True
+        )
+
+        runner = CliRunner()
+        args = [
+            "new",
+            "--template-path",
+            str(template),
+            "--snapshot-name",
+            "BaseClean",
+            "--clone-base-dir",
+            str(clone_base),
+            "--job-id",
+            "job-11",
+            "--vmrun-path",
+            fake_vmrun,
+        ]
+        # Two consecutive runs.
+        for _ in range(2):
+            result = runner.invoke(vm_group, args, catch_exceptions=False)
+            assert result.exit_code == 0, result.output
+        return runner, calls
+
+    def test_two_vm_new_invocations_produce_distinct_clones(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+        fake_vmrun: str,
+    ) -> None:
+        # Drive the timestamp generator so the two invocations land in
+        # different seconds (vm_new uses %Y%m%d_%H%M%S).
+        from ci_orchestrator.commands import vm as vm_module
+
+        ticks = iter(["20260514_120000", "20260514_120001"])
+
+        class _FakeDT:
+            @staticmethod
+            def now(tz: object | None = None) -> Any:
+                class _F:
+                    def strftime(self_inner, _fmt: str) -> str:
+                        return next(ticks)
+
+                return _F()
+
+        monkeypatch.setattr(vm_module, "datetime", _FakeDT)
+
+        _runner, calls = self._invoke_vm_new(monkeypatch, tmp_path, fake_vmrun)
+
+        assert len(calls) == 2
+        names = {c["name"] for c in calls}
+        destinations = {c["destination"] for c in calls}
+        assert len(names) == 2, (
+            f"vm new must never produce duplicate clone names; got {names} "
+            "(see AGENTS.md error #11 — machine-id / DHCP collision)"
+        )
+        assert len(destinations) == 2, (
+            "Two vm new invocations produced the same destination VMX path; "
+            "this would clone over an existing build VM and risks reproducing "
+            "the duplicate-IP failure mode (AGENTS.md error #11)."
+        )
+
+
+# ──────────────────────────────────── AGENTS.md #12 — no ssh-keygen / AutoAddPolicy
+
+
+class TestError12SshTransportNoSshKeygen:
+    """``& ssh-keygen -R <ip> 2>$null`` does NOT suppress stderr in
+    Windows PowerShell 5.1 with ``$ErrorActionPreference='Stop'``: stderr
+    is converted to a terminating ErrorRecord before ``2>$null`` can
+    discard it. The Python rewrite sidesteps the entire problem by using
+    paramiko with :class:`paramiko.AutoAddPolicy` and an empty
+    ``known_hosts`` (mirroring ``StrictHostKeyChecking=no
+    UserKnownHostsFile=NUL``). We assert there is no native shell-out to
+    ``ssh-keygen`` and that ``AutoAddPolicy`` is installed by default.
+    """
+
+    def _install_fake_paramiko(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> tuple[Any, Any]:
+        """Install a stub paramiko module that records all attribute access."""
+
+        accessed: list[str] = []
+
+        class _Channel:
+            def recv_exit_status(self) -> int:
+                return 0
+
+        class _Stream:
+            channel = _Channel()
+
+            def read(self) -> bytes:
+                return b""
+
+        class _SFTP:
+            def __enter__(self) -> _SFTP:
+                return self
+
+            def __exit__(self, *_e: object) -> None:
+                pass
+
+            def put(self, _l: str, _r: str) -> None:  # pragma: no cover - unused
+                pass
+
+            def get(self, _r: str, _l: str) -> None:  # pragma: no cover - unused
+                pass
+
+        class _Client:
+            def __init__(self) -> None:
+                self.policy: Any = None
+                self.connect_kwargs: dict[str, Any] = {}
+                self.commands: list[str] = []
+
+            def set_missing_host_key_policy(self, policy: Any) -> None:
+                self.policy = policy
+
+            def load_host_keys(self, _path: str) -> None:  # pragma: no cover
+                pass
+
+            def connect(self, **kwargs: Any) -> None:
+                self.connect_kwargs = kwargs
+
+            def exec_command(
+                self, command: str, timeout: float | None = None
+            ) -> Any:
+                self.commands.append(command)
+                return None, _Stream(), _Stream()
+
+            def open_sftp(self) -> _SFTP:  # pragma: no cover - unused
+                return _SFTP()
+
+            def close(self) -> None:
+                pass
+
+        client = _Client()
+
+        class _ModWrap(types.ModuleType):
+            def __getattr__(self, item: str) -> Any:
+                accessed.append(item)
+                raise AttributeError(item)
+
+        fake = _ModWrap("paramiko")
+        # Pre-define exactly the attributes SshTransport is allowed to use.
+        fake.SSHClient = lambda: client  # type: ignore[attr-defined]
+        fake.AutoAddPolicy = lambda: "auto-add"  # type: ignore[attr-defined]
+
+        monkeypatch.setitem(sys.modules, "paramiko", fake)
+        return client, accessed
+
+    def test_default_policy_is_auto_add_no_known_hosts(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        client, _accessed = self._install_fake_paramiko(monkeypatch)
+        t = SshTransport("10.0.0.99", username="ci_build", key_path="/tmp/k")
+        t.run("true")
+        # AutoAddPolicy is the marker that we're matching the
+        # `StrictHostKeyChecking=no` permissive default.
+        assert client.policy == "auto-add", (
+            "SshTransport must default to paramiko.AutoAddPolicy "
+            "(see AGENTS.md error #12)."
+        )
+        # Ensure the connect call did NOT pass any known_hosts file path,
+        # which would re-introduce the host-key churn problem.
+        assert "key_filename" in client.connect_kwargs
+
+    def test_no_ssh_keygen_invocation(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """SshTransport must not shell out to ``ssh-keygen`` (or anything
+        else): all transport goes through paramiko."""
+        called: list[list[str]] = []
+
+        def fake_run(cmd: list[str], *_a: object, **_kw: object) -> Any:
+            called.append(cmd)
+            return _completed(0)
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+
+        # Also patch Popen as a belt-and-braces guard.
+        def fake_popen(*a: object, **_kw: object) -> Any:  # pragma: no cover
+            called.append(list(a))
+            raise AssertionError(
+                "SshTransport must not Popen any external process "
+                "(AGENTS.md error #12)."
+            )
+
+        monkeypatch.setattr(subprocess, "Popen", fake_popen)
+
+        client, _accessed = self._install_fake_paramiko(monkeypatch)
+        t = SshTransport("10.0.0.99", key_path="/tmp/k")
+        t.run("uname -a")
+        t.close()
+
+        assert called == [], (
+            "SshTransport spawned a subprocess; it must rely solely on "
+            "paramiko (no ssh-keygen, no ssh.exe, no scp.exe). Calls: "
+            f"{called!r}"
+        )
+        assert client.commands == ["uname -a"]
+
+    def test_known_hosts_optional_param_is_off_by_default(self) -> None:
+        """The ``known_hosts`` constructor parameter must default to None
+        so ephemeral CI VMs never hit a stale host-key error that would
+        previously have driven the broken ``ssh-keygen -R`` workaround."""
+        t = SshTransport("10.0.0.99", key_path="/tmp/k")
+        # Internal attribute is part of the documented contract for this
+        # regression: flipping the default would silently break ephemeral
+        # VM workflows. If you need to change it, update AGENTS.md #12.
+        assert t._known_hosts is None
+
+    def test_paramiko_missing_raises_typed_error(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """When paramiko is not installed, the transport must raise a
+        typed :class:`TransportConnectError` — not a bare ``ImportError``
+        leaking through. Symmetric to the PowerShell guard that wrapped
+        ``ssh.exe`` failures."""
+        from ci_orchestrator.transport.errors import TransportConnectError
+
+        monkeypatch.setitem(sys.modules, "paramiko", None)
+        t = SshTransport("10.0.0.99", key_path="/tmp/k")
+        with pytest.raises(TransportConnectError):
+            t.run("true")
+
+    def test_exec_command_exception_is_wrapped(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Channel-level paramiko errors must surface as
+        :class:`TransportConnectError`, never as raw ``Exception``."""
+        from ci_orchestrator.transport.errors import TransportConnectError
+
+        client, _ = self._install_fake_paramiko(monkeypatch)
+
+        def boom(*_a: object, **_kw: object) -> Any:
+            raise OSError("channel closed")
+
+        client.exec_command = boom  # type: ignore[method-assign]
+        t = SshTransport("10.0.0.99", key_path="/tmp/k")
+        with pytest.raises(TransportConnectError):
+            t.run("uname")
+
+    def test_sftp_failures_are_wrapped(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from ci_orchestrator.transport.errors import TransportConnectError
+
+        client, _ = self._install_fake_paramiko(monkeypatch)
+
+        def boom_sftp() -> Any:
+            raise OSError("sftp denied")
+
+        client.open_sftp = boom_sftp  # type: ignore[method-assign]
+        t = SshTransport("10.0.0.99", key_path="/tmp/k")
+        with pytest.raises(TransportConnectError):
+            t.copy("local.bin", "/remote/local.bin")
+        with pytest.raises(TransportConnectError):
+            t.fetch("/remote/out.bin", "out.bin")
+
+    def test_is_ready_returns_false_on_typed_error(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        client, _ = self._install_fake_paramiko(monkeypatch)
+
+        def boom(*_a: object, **_kw: object) -> Any:
+            raise OSError("dead")
+
+        client.exec_command = boom  # type: ignore[method-assign]
+        t = SshTransport("10.0.0.99", key_path="/tmp/k")
+        assert t.is_ready() is False