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