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