local-ci-cd-system/tests/python/test_workstation_backend.py

"""Unit tests for WorkstationVmrunBackend.

Cover AGENTS.md error #10: ``is_running`` MUST use ``vmrun list`` and
NOT ``getGuestIPAddress`` (which would block 30-60s without VMware Tools).
"""

from __future__ import annotations

import subprocess
from pathlib import Path
from typing import Any

import pytest

from ci_orchestrator.backends.errors import BackendNotAvailable, BackendOperationFailed
from ci_orchestrator.backends.protocol import VmHandle
from ci_orchestrator.backends.workstation import WorkstationVmrunBackend


@pytest.fixture
def fake_vmrun(tmp_path: Path) -> str:
    """A real (empty) file we can pass as vmrun_path."""
    p = tmp_path / "vmrun.exe"
    p.write_bytes(b"")
    return str(p)


def _make_completed(returncode: int = 0, stdout: str = "", stderr: str = "") -> Any:
    return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr)


def test_init_raises_when_vmrun_missing(tmp_path: Path) -> None:
    missing = tmp_path / "nope.exe"
    with pytest.raises(BackendNotAvailable):
        WorkstationVmrunBackend(vmrun_path=str(missing))


def test_init_uses_explicit_path(fake_vmrun: str) -> None:
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.vmrun_path == fake_vmrun


def test_clone_linked_invokes_vmrun(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    captured: dict[str, list[str]] = {}

    def fake_run(cmd: list[str], **_kwargs: object) -> Any:
        captured["cmd"] = cmd
        return _make_completed(0, stdout="")

    monkeypatch.setattr(subprocess, "run", fake_run)
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    handle = backend.clone_linked(
        template=r"C:\tpl\WinBuild2025\WinBuild2025.vmx",
        snapshot="BaseClean",
        name="job-123",
        destination=r"C:\bvm\job-123\job-123.vmx",
    )
    assert handle.identifier == r"C:\bvm\job-123\job-123.vmx"
    assert handle.name == "job-123"
    assert captured["cmd"][0] == fake_vmrun
    assert captured["cmd"][1:4] == ["-T", "ws", "clone"]
    assert "linked" in captured["cmd"]
    assert "-snapshot=BaseClean" in captured["cmd"]
    assert "-cloneName=job-123" in captured["cmd"]


def test_clone_failure_raises(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(
        subprocess,
        "run",
        lambda *a, **kw: _make_completed(1, stderr="Error: source not found"),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    with pytest.raises(BackendOperationFailed) as exc_info:
        backend.clone_linked("src.vmx", "snap", "name", destination="dst.vmx")
    assert exc_info.value.returncode == 1
    assert "source not found" in exc_info.value.output


def test_stop_tolerates_already_off(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(
        subprocess,
        "run",
        lambda *a, **kw: _make_completed(255, stderr="Error: The virtual machine is not powered on"),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    # Should NOT raise — VM already off is treated as success.
    backend.stop(VmHandle("x.vmx"))


def test_get_ip_returns_none_on_failure(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(
        subprocess, "run", lambda *a, **kw: _make_completed(1, stderr="Tools not running")
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.get_ip(VmHandle("x.vmx")) is None


def test_get_ip_returns_none_when_tools_not_running(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    """vmrun exits 0 but writes 'Error: ...' to stdout — must return None, not the error string."""
    monkeypatch.setattr(
        subprocess,
        "run",
        lambda *a, **kw: _make_completed(
            0, stdout="Error: The VMware Tools are not running in the virtual machine: x.vmx\n"
        ),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.get_ip(VmHandle("x.vmx")) is None


def test_get_ip_returns_address(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(0, stdout="192.168.79.42\n"))
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.get_ip(VmHandle("x.vmx")) == "192.168.79.42"


def test_list_snapshots_strips_header(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(
        subprocess,
        "run",
        lambda *a, **kw: _make_completed(0, stdout="Total snapshots: 2\nBaseClean\nDirty\n"),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.list_snapshots(VmHandle("x.vmx")) == ["BaseClean", "Dirty"]


def test_is_running_uses_list_not_getip(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str
) -> None:
    """Regression for AGENTS.md error #10."""
    vmx = tmp_path / "job1" / "job1.vmx"
    vmx.parent.mkdir()
    vmx.write_text("")

    calls: list[str] = []

    def fake_run(cmd: list[str], **_kwargs: object) -> Any:
        calls.append(cmd[3])  # operation name
        if cmd[3] == "list":
            return _make_completed(
                0, stdout=f"Total running VMs: 1\n{vmx}\n"
            )
        raise AssertionError(f"Unexpected vmrun op: {cmd[3]}")

    monkeypatch.setattr(subprocess, "run", fake_run)
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.is_running(VmHandle(str(vmx))) is True
    assert calls == ["list"]
    assert "getGuestIPAddress" not in calls


def test_is_running_returns_false_when_absent(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str
) -> None:
    vmx = tmp_path / "job1.vmx"
    vmx.write_text("")
    monkeypatch.setattr(
        subprocess, "run", lambda *a, **kw: _make_completed(0, stdout="Total running VMs: 0\n")
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.is_running(VmHandle(str(vmx))) is False


# ── _resolve_vmrun(explicit=None) paths ───────────────────────────────────────

def test_resolve_vmrun_uses_which(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    import shutil as _shutil
    monkeypatch.setattr(_shutil, "which", lambda _name: fake_vmrun)
    backend = WorkstationVmrunBackend()  # no explicit path
    assert backend.vmrun_path == fake_vmrun


def test_resolve_vmrun_uses_default_win_path(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    import shutil as _shutil
    from ci_orchestrator.backends import workstation as ws_mod

    fake_default = tmp_path / "vmrun.exe"
    fake_default.write_bytes(b"")
    monkeypatch.setattr(_shutil, "which", lambda _name: None)
    monkeypatch.setattr(ws_mod, "_DEFAULT_WIN_VMRUN", str(fake_default))
    backend = WorkstationVmrunBackend()
    assert backend.vmrun_path == str(fake_default)


def test_resolve_vmrun_raises_when_nothing_found(monkeypatch: pytest.MonkeyPatch) -> None:
    import shutil as _shutil
    from ci_orchestrator.backends import workstation as ws_mod

    monkeypatch.setattr(_shutil, "which", lambda _name: None)
    monkeypatch.setattr(ws_mod, "_DEFAULT_WIN_VMRUN", "/nonexistent/vmrun.exe")
    with pytest.raises(BackendNotAvailable):
        WorkstationVmrunBackend()


# ── clone_linked destination=None ────────────────────────────────────────────

def test_clone_linked_auto_destination(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    captured: dict[str, list[str]] = {}

    def fake_run(cmd: list[str], **_kw: object) -> Any:
        captured["cmd"] = cmd
        return _make_completed(0)

    monkeypatch.setattr(subprocess, "run", fake_run)
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    handle = backend.clone_linked(
        template=r"/ci/tpl/WinBuild2025/WinBuild2025.vmx",
        snapshot="BaseClean",
        name="job-auto",
    )
    # destination auto-computed as <template_parent.parent>/<name>/<name>.vmx
    assert handle.name == "job-auto"
    assert handle.identifier.endswith("job-auto.vmx")


# ── start headless=False ─────────────────────────────────────────────────────

def test_start_gui_mode(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    captured: list[str] = []
    monkeypatch.setattr(
        subprocess, "run",
        lambda cmd, **_kw: (captured.append(cmd[-1]) or _make_completed(0)),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    backend.start(VmHandle("x.vmx"), headless=False)
    assert captured[-1] == "gui"


# ── stop edge cases ───────────────────────────────────────────────────────────

def test_stop_succeeds_on_rc_zero(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(0))
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    backend.stop(VmHandle("x.vmx"))  # must not raise


def test_stop_raises_on_unrecognized_error(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(
        subprocess, "run",
        lambda *a, **kw: _make_completed(1, stderr="Error: something unexpected"),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    with pytest.raises(BackendOperationFailed):
        backend.stop(VmHandle("x.vmx"))


def test_stop_tolerates_is_not_running(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(
        subprocess, "run",
        lambda *a, **kw: _make_completed(1, stderr="Error: vm is not running"),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    backend.stop(VmHandle("x.vmx"))  # must not raise


# ── delete ────────────────────────────────────────────────────────────────────

def test_delete_invokes_delete_vm(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    ops: list[str] = []
    monkeypatch.setattr(
        subprocess, "run",
        lambda cmd, **_kw: (ops.append(cmd[3]) or _make_completed(0)),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    backend.delete(VmHandle("x.vmx"))
    assert ops == ["deleteVM"]


# ── get_ip timeout / TimeoutExpired / empty stdout ───────────────────────────

def test_get_ip_passes_wait_when_timeout_positive(
    monkeypatch: pytest.MonkeyPatch, fake_vmrun: str
) -> None:
    captured: list[list[str]] = []
    monkeypatch.setattr(
        subprocess, "run",
        lambda cmd, **_kw: (captured.append(cmd) or _make_completed(0, stdout="10.0.0.1\n")),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    ip = backend.get_ip(VmHandle("x.vmx"), timeout=5.0)
    assert ip == "10.0.0.1"
    assert "-wait" in captured[0]


def test_get_ip_returns_none_on_timeout_expired(
    monkeypatch: pytest.MonkeyPatch, fake_vmrun: str
) -> None:
    def boom(*_a: object, **_kw: object) -> Any:
        raise subprocess.TimeoutExpired(cmd="vmrun", timeout=1)

    monkeypatch.setattr(subprocess, "run", boom)
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.get_ip(VmHandle("x.vmx"), timeout=1.0) is None


def test_get_ip_returns_none_on_empty_stdout(
    monkeypatch: pytest.MonkeyPatch, fake_vmrun: str
) -> None:
    monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(0, stdout="   \n"))
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.get_ip(VmHandle("x.vmx")) is None


# ── list_snapshots without header ─────────────────────────────────────────────

def test_list_snapshots_without_header(monkeypatch: pytest.MonkeyPatch, fake_vmrun: str) -> None:
    monkeypatch.setattr(
        subprocess, "run",
        lambda *a, **kw: _make_completed(0, stdout="BaseClean\n"),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.list_snapshots(VmHandle("x.vmx")) == ["BaseClean"]


# ── is_running edge cases ─────────────────────────────────────────────────────

def test_is_running_returns_false_on_list_failure(
    monkeypatch: pytest.MonkeyPatch, fake_vmrun: str
) -> None:
    monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _make_completed(1, stderr="err"))
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.is_running(VmHandle("x.vmx")) is False


def test_is_running_skips_non_matching_line(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fake_vmrun: str
) -> None:
    target = tmp_path / "job1.vmx"
    target.write_text("")
    other = tmp_path / "other.vmx"
    other.write_text("")
    monkeypatch.setattr(
        subprocess, "run",
        lambda *a, **kw: _make_completed(0, stdout=f"Total running VMs: 1\n{other}\n"),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    assert backend.is_running(VmHandle(str(target))) is False


def test_is_running_handles_oserror_on_bad_path(
    monkeypatch: pytest.MonkeyPatch, fake_vmrun: str
) -> None:
    import pathlib

    _real_resolve = pathlib.Path.resolve

    def _patched_resolve(self: Path, *args: object, **kw: object) -> Path:
        if "trigger_oserror" in str(self):
            raise OSError("EACCES: permission denied")
        return _real_resolve(self, *args, **kw)  # type: ignore[return-value]

    monkeypatch.setattr(pathlib.Path, "resolve", _patched_resolve)
    monkeypatch.setattr(
        subprocess, "run",
        lambda *a, **kw: _make_completed(
            0, stdout="Total running VMs: 1\n/some/trigger_oserror.vmx\n"
        ),
    )
    backend = WorkstationVmrunBackend(vmrun_path=fake_vmrun)
    # OSError from Path.resolve() must be swallowed; VM not found → False.
    assert backend.is_running(VmHandle("x.vmx")) is False