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

"""Tests for ``ci_orchestrator vm remove`` and ``vm cleanup``.

Migrated from Pester ``tests/Remove-BuildVM.Tests.ps1`` (now removed).
"""

from __future__ import annotations

import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

import pytest
from click.testing import CliRunner

import ci_orchestrator.commands.vm as vm_module
from ci_orchestrator.__main__ import cli
from ci_orchestrator.backends.errors import BackendOperationFailed
from ci_orchestrator.backends.protocol import VmHandle


class _FakeBackend:
    def __init__(
        self,
        *,
        running_after_stop: bool = False,
        delete_fails_first: int = 0,
    ) -> None:
        self.calls: list[tuple[str, Any]] = []
        self._running = True
        self._running_after_stop = running_after_stop
        self._delete_fails_remaining = delete_fails_first

    def stop(self, handle: VmHandle, hard: bool = False) -> None:
        self.calls.append(("stop", hard))
        if not hard or not self._running_after_stop:
            self._running = False

    def is_running(self, _h: VmHandle) -> bool:
        return self._running

    def delete(self, _h: VmHandle) -> None:
        self.calls.append(("delete", None))
        if self._delete_fails_remaining > 0:
            self._delete_fails_remaining -= 1
            raise BackendOperationFailed("deleteVM", 1, "transient")


@pytest.fixture(autouse=True)
def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(time, "sleep", lambda _s: None)
    monkeypatch.setattr(vm_module.time, "sleep", lambda _s: None)


def _make_clone_dir(tmp_path: Path, name: str = "clone") -> tuple[Path, Path]:
    clone_dir = tmp_path / name
    clone_dir.mkdir()
    vmx = clone_dir / f"{name}.vmx"
    vmx.write_text('config.version = "8"', encoding="utf-8")
    return clone_dir, vmx


# ── vm remove ──────────────────────────────────────────────────────────────


def test_vm_remove_missing_vmx_returns_without_error(tmp_path: Path) -> None:
    """Pester migration: returns without error when VMX does not exist."""
    missing = tmp_path / "ghost" / "ghost.vmx"
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(missing)])
    assert result.exit_code == 0, result.output
    assert "nothing to remove" in result.output


def test_vm_remove_partial_clone_dir_is_removed(tmp_path: Path) -> None:
    """Pester migration: partial clone dir without VMX is still removed."""
    clone_dir, vmx = _make_clone_dir(tmp_path, "partial")
    vmx.unlink()  # leave the directory but no VMX
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
    assert result.exit_code == 0
    assert not clone_dir.exists()


def test_vm_remove_happy_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    """Pester migration: removes the clone dir when VMX exists and stop+delete succeed."""
    clone_dir, vmx = _make_clone_dir(tmp_path, "live")
    backend = _FakeBackend()
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
    result = CliRunner().invoke(
        cli, ["vm", "remove", "--vmx", str(vmx), "--graceful-timeout", "1"]
    )
    assert result.exit_code == 0, result.output
    assert ("delete", None) in backend.calls
    assert not clone_dir.exists()


def test_vm_remove_force_skips_soft_stop(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    _, vmx = _make_clone_dir(tmp_path, "force")
    backend = _FakeBackend()
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx), "--force"])
    assert result.exit_code == 0, result.output
    # No soft stop was issued.
    soft_stops = [c for c in backend.calls if c == ("stop", False)]
    assert soft_stops == []
    hard_stops = [c for c in backend.calls if c == ("stop", True)]
    assert hard_stops != []


def test_vm_remove_delete_retry_succeeds(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    clone_dir, vmx = _make_clone_dir(tmp_path, "retry")
    backend = _FakeBackend(delete_fails_first=2)
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
    assert result.exit_code == 0
    delete_calls = [c for c in backend.calls if c[0] == "delete"]
    assert len(delete_calls) == 3  # 2 failures + 1 success
    assert not clone_dir.exists()


# ── vm cleanup ─────────────────────────────────────────────────────────────


def _age_dir(path: Path, hours: float) -> None:
    ts = (datetime.now() - timedelta(hours=hours)).timestamp()
    import os

    os.utime(path, (ts, ts))


def test_vm_cleanup_dry_run_lists_orphans(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    base = tmp_path / "build-vms"
    base.mkdir()
    fresh, _ = _make_clone_dir(base, "fresh")
    old, _ = _make_clone_dir(base, "old")
    _age_dir(old, hours=10)
    _age_dir(fresh, hours=0)
    # Make sure backend isn't constructed (vmrun absent on test host).
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: None)
    result = CliRunner().invoke(
        cli,
        [
            "vm",
            "cleanup",
            "--clone-base-dir",
            str(base),
            "--max-age-hours",
            "4",
            "--what-if",
        ],
    )
    assert result.exit_code == 0, result.output
    assert "would destroy" in result.output
    assert old.exists()  # dry run did NOT delete
    assert fresh.exists()


def test_vm_cleanup_destroys_old_dirs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    base = tmp_path / "build-vms"
    base.mkdir()
    old, _ = _make_clone_dir(base, "old")
    _age_dir(old, hours=10)

    backend = _FakeBackend()
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)

    result = CliRunner().invoke(
        cli,
        [
            "vm",
            "cleanup",
            "--clone-base-dir",
            str(base),
            "--max-age-hours",
            "4",
        ],
    )
    assert result.exit_code == 0, result.output
    assert not old.exists()
    # Backend was used to stop+delete.
    assert any(c[0] == "delete" for c in backend.calls)


def test_vm_cleanup_handles_missing_base_dir(tmp_path: Path) -> None:
    missing = tmp_path / "does-not-exist"
    result = CliRunner().invoke(
        cli, ["vm", "cleanup", "--clone-base-dir", str(missing)]
    )
    assert result.exit_code == 0
    assert "nothing to do" in result.output


def test_vm_cleanup_removes_stale_lock(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    base = tmp_path / "bv"
    base.mkdir()
    lock = tmp_path / "vm-start.lock"
    lock.write_text("x")
    _age_dir(lock, hours=1)
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: None)
    result = CliRunner().invoke(
        cli,
        [
            "vm",
            "cleanup",
            "--clone-base-dir",
            str(base),
            "--lock-file",
            str(lock),
        ],
    )
    assert result.exit_code == 0
    assert not lock.exists()


# ── helpers + uncovered branches ──────────────────────────────────────────


def test_make_backend_with_vmrun_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    """Override path returns a WorkstationVmrunBackend honouring vmrun_path."""
    from ci_orchestrator.backends.workstation import WorkstationVmrunBackend

    class _StubConfig:
        pass

    fake_vmrun = tmp_path / "vmrun.exe"
    fake_vmrun.write_text("x")
    monkeypatch.setattr(vm_module, "load_config", lambda: _StubConfig())
    backend = vm_module._make_backend(str(fake_vmrun))
    assert isinstance(backend, WorkstationVmrunBackend)


def test_make_backend_default_uses_load_backend(monkeypatch: pytest.MonkeyPatch) -> None:
    sentinel = object()
    monkeypatch.setattr(vm_module, "load_config", lambda: "cfg")
    monkeypatch.setattr(vm_module, "load_backend", lambda _c: sentinel)
    assert vm_module._make_backend(None) is sentinel


def test_try_remove_dir_returns_true_when_missing(tmp_path: Path) -> None:
    assert vm_module._try_remove_dir(tmp_path / "ghost") is True


def test_try_remove_dir_real_removal(tmp_path: Path) -> None:
    d = tmp_path / "x"
    (d / "sub").mkdir(parents=True)
    (d / "sub" / "f.txt").write_text("a")
    assert vm_module._try_remove_dir(d) is True
    assert not d.exists()


def test_stop_with_fallback_soft_error_then_running_with_hard_error() -> None:
    from ci_orchestrator.backends.errors import BackendOperationFailed

    class _B:
        def __init__(self) -> None:
            self.calls: list[tuple[str, bool]] = []

        def stop(self, _h: VmHandle, hard: bool = False) -> None:
            self.calls.append(("stop", hard))
            raise BackendOperationFailed("stop", 1, "boom")

        def is_running(self, _h: VmHandle) -> bool:
            return True

    b = _B()
    # Should swallow both errors.
    vm_module._stop_with_fallback(b, VmHandle(identifier="x.vmx"))
    assert ("stop", False) in b.calls
    assert ("stop", True) in b.calls


def test_vm_remove_backend_unavailable(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    from ci_orchestrator.backends.errors import BackendNotAvailable

    clone_dir, vmx = _make_clone_dir(tmp_path, "noback")

    def _raise(_v: Any) -> Any:
        raise BackendNotAvailable("vmrun missing")

    monkeypatch.setattr(vm_module, "_make_backend", _raise)
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
    assert result.exit_code == 0, result.output
    assert "vmrun unavailable" in result.output
    assert not clone_dir.exists()


def test_vm_remove_force_hard_stop_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    from ci_orchestrator.backends.errors import BackendOperationFailed

    _, vmx = _make_clone_dir(tmp_path, "forcefail")

    class _B:
        def stop(self, _h: VmHandle, hard: bool = False) -> None:
            raise BackendOperationFailed("stop", 1, "no")

        def is_running(self, _h: VmHandle) -> bool:
            return False

        def delete(self, _h: VmHandle) -> None:
            pass

    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: _B())
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx), "--force"])
    assert result.exit_code == 0
    assert "hard stop failed" in result.output


def test_vm_remove_deadline_loop_with_running_then_error(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Cover the deadline loop that breaks on BackendError."""
    from ci_orchestrator.backends.errors import BackendOperationFailed

    _, vmx = _make_clone_dir(tmp_path, "loop")

    class _B:
        def __init__(self) -> None:
            self._calls = 0

        def stop(self, _h: VmHandle, hard: bool = False) -> None:
            pass

        def is_running(self, _h: VmHandle) -> bool:
            self._calls += 1
            if self._calls == 1:
                return True
            raise BackendOperationFailed("isrunning", 1, "lost")

        def delete(self, _h: VmHandle) -> None:
            pass

    # Make monotonic strictly advance so the loop runs at least twice but exits.
    base_t = [0.0]

    def _mono() -> float:
        base_t[0] += 1.0
        return base_t[0]

    monkeypatch.setattr(vm_module.time, "monotonic", _mono)
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: _B())
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
    assert result.exit_code == 0


def test_vm_remove_delete_fails_all_retries(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Delete that fails on every retry triggers fallback dir-removal log."""
    clone_dir, vmx = _make_clone_dir(tmp_path, "neverdeletes")

    class _B:
        def stop(self, _h: VmHandle, hard: bool = False) -> None:
            pass

        def is_running(self, _h: VmHandle) -> bool:
            return False

        def delete(self, _h: VmHandle) -> None:
            from ci_orchestrator.backends.errors import BackendOperationFailed
            raise BackendOperationFailed("delete", 1, "always")

    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: _B())
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
    assert result.exit_code == 0
    assert "deleteVM failed after retries" in result.output
    assert not clone_dir.exists()


def test_vm_remove_clone_dir_removal_failure(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """When _try_remove_dir reports failure, the manual-cleanup notice fires."""
    _, vmx = _make_clone_dir(tmp_path, "stubborn")
    backend = _FakeBackend()
    monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
    monkeypatch.setattr(vm_module, "_try_remove_dir", lambda _p: False)
    result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
    assert result.exit_code == 0
    assert "manual cleanup needed" in result.output


def test_list_orphans_non_dir_base(tmp_path: Path) -> None:
    f = tmp_path / "file.txt"
    f.write_text("x")
    assert vm_module._list_orphans(f, 4) == []


def test_list_orphans_skips_non_dir_and_oserror(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    base = tmp_path / "b"
    base.mkdir()
    (base / "a-file").write_text("not a dir")
    bad_dir = base / "bad"
    bad_dir.mkdir()
    _age_dir(bad_dir, hours=10)

    real_stat = Path.stat

    def _stat(self: Path, *a: Any, **kw: Any) -> Any:
        if self == bad_dir:
            raise OSError("nope")
        return real_stat(self, *a, **kw)

    monkeypatch.setattr(Path, "stat", _stat)
    assert vm_module._list_orphans(base, 4) == []


def test_find_vmx_returns_none(tmp_path: Path) -> None:
    d = tmp_path / "novmx"
    d.mkdir()
    (d / "readme.txt").write_text("x")
    assert vm_module._find_vmx(d) is None


def test_cleanup_orphans_dir_without_vmx_logs_and_removes(tmp_path: Path) -> None:
    base = tmp_path / "b"
    base.mkdir()
    no_vmx = base / "no-vmx"
    no_vmx.mkdir()
    _age_dir(no_vmx, hours=10)
    runner_count = vm_module.cleanup_orphans(
        clone_base=base, max_age_hours=4, backend=None
    )
    assert runner_count == 1
    assert not no_vmx.exists()


def test_cleanup_orphans_backend_stop_error_swallowed(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    from ci_orchestrator.backends.errors import BackendOperationFailed

    base = tmp_path / "b"
    base.mkdir()
    old, _ = _make_clone_dir(base, "old1")
    _age_dir(old, hours=10)

    class _B:
        def stop(self, _h: VmHandle, hard: bool = False) -> None:
            raise BackendOperationFailed("stop", 1, "no")

        def delete(self, _h: VmHandle) -> None:
            raise BackendOperationFailed("delete", 1, "no")

    count = vm_module.cleanup_orphans(
        clone_base=base, max_age_hours=4, backend=_B()
    )
    assert count == 1


def test_cleanup_orphans_dir_removal_failure_logs(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    base = tmp_path / "b"
    base.mkdir()
    old, _ = _make_clone_dir(base, "stuck")
    _age_dir(old, hours=10)
    monkeypatch.setattr(vm_module, "_try_remove_dir", lambda _p: False)
    count = vm_module.cleanup_orphans(
        clone_base=base, max_age_hours=4, backend=None
    )
    assert count == 0


def test_cleanup_orphans_lock_unlink_oserror(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    base = tmp_path / "b"
    base.mkdir()
    lock = tmp_path / "stale.lock"
    lock.write_text("x")
    _age_dir(lock, hours=1)

    real_unlink = Path.unlink

    def _boom(self: Path, *a: Any, **kw: Any) -> Any:
        if self == lock:
            raise OSError("locked")
        return real_unlink(self, *a, **kw)

    monkeypatch.setattr(Path, "unlink", _boom)
    vm_module.cleanup_orphans(
        clone_base=base, max_age_hours=4, backend=None, lock_file=lock
    )


def test_vm_cleanup_backend_unavailable(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    from ci_orchestrator.backends.errors import BackendNotAvailable

    base = tmp_path / "build-vms"
    base.mkdir()

    def _raise(_v: Any) -> Any:
        raise BackendNotAvailable("no vmrun")

    monkeypatch.setattr(vm_module, "_make_backend", _raise)
    result = CliRunner().invoke(
        cli, ["vm", "cleanup", "--clone-base-dir", str(base)]
    )
    assert result.exit_code == 0
    assert "vmrun not available" in result.output


# ── vmgui ───────────────────────────────────────────────────────────────────


def _capture_launch(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
    """Patch _launch_gui to record argv/env instead of spawning a process."""
    captured: dict[str, Any] = {}

    def _fake(argv: list[str], env: dict[str, str]) -> int:
        captured["argv"] = argv
        captured["env"] = env
        return 4242

    monkeypatch.setattr(vm_module, "_launch_gui", _fake)
    return captured


def test_vm_open_missing_vmx_errors(tmp_path: Path) -> None:
    result = CliRunner().invoke(cli, ["vmgui", "--vmx", str(tmp_path / "nope.vmx")])
    assert result.exit_code != 0
    assert "VMX file not found" in result.output


def test_vm_open_happy_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    vmx = tmp_path / "clone.vmx"
    vmx.write_text("config")
    cap = _capture_launch(monkeypatch)
    monkeypatch.setenv("DISPLAY", ":0")
    result = CliRunner().invoke(cli, ["vmgui", "--vmx", str(vmx)])
    assert result.exit_code == 0, result.output
    assert cap["argv"][-1] == str(vmx)
    assert "-x" not in cap["argv"] and "-X" not in cap["argv"]
    assert "pid 4242" in result.output


def test_vm_open_power_on_adds_x(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    vmx = tmp_path / "clone.vmx"
    vmx.write_text("config")
    cap = _capture_launch(monkeypatch)
    monkeypatch.setenv("DISPLAY", ":0")
    result = CliRunner().invoke(cli, ["vmgui", "--vmx", str(vmx), "--power-on"])
    assert result.exit_code == 0, result.output
    assert "-x" in cap["argv"]


def test_vm_open_fullscreen_adds_capital_x(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    vmx = tmp_path / "clone.vmx"
    vmx.write_text("config")
    cap = _capture_launch(monkeypatch)
    monkeypatch.setenv("DISPLAY", ":0")
    result = CliRunner().invoke(
        cli, ["vmgui", "--vmx", str(vmx), "--fullscreen", "--new-window"]
    )
    assert result.exit_code == 0, result.output
    assert "-X" in cap["argv"] and "-n" in cap["argv"]


def test_vm_open_display_option_overrides_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    vmx = tmp_path / "clone.vmx"
    vmx.write_text("config")
    cap = _capture_launch(monkeypatch)
    monkeypatch.delenv("DISPLAY", raising=False)
    result = CliRunner().invoke(cli, ["vmgui", "--vmx", str(vmx), "--display", ":1"])
    assert result.exit_code == 0, result.output
    assert cap["env"]["DISPLAY"] == ":1"
    assert "display=:1" in result.output


def test_vm_open_warns_without_display(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    vmx = tmp_path / "clone.vmx"
    vmx.write_text("config")
    _capture_launch(monkeypatch)
    monkeypatch.delenv("DISPLAY", raising=False)
    result = CliRunner().invoke(cli, ["vmgui", "--vmx", str(vmx)])
    assert result.exit_code == 0, result.output
    assert "no X display" in result.output


def test_vm_open_launch_failure_is_reported(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    vmx = tmp_path / "clone.vmx"
    vmx.write_text("config")

    def _boom(argv: list[str], env: dict[str, str]) -> int:
        raise OSError("vmware not found")

    monkeypatch.setattr(vm_module, "_launch_gui", _boom)
    monkeypatch.setenv("DISPLAY", ":0")
    result = CliRunner().invoke(cli, ["vmgui", "--vmx", str(vmx)])
    assert result.exit_code != 0
    assert "failed to launch VMware GUI" in result.output


def test_vm_open_without_vmx_launches_bare_gui(monkeypatch: pytest.MonkeyPatch) -> None:
    cap = _capture_launch(monkeypatch)
    monkeypatch.setenv("DISPLAY", ":0")
    result = CliRunner().invoke(cli, ["vmgui"])
    assert result.exit_code == 0, result.output
    # Only the binary (no vmx path, no power-on flags).
    assert cap["argv"][1:] == [] or cap["argv"][1:] == ["-n"]
    assert "-x" not in cap["argv"] and "-X" not in cap["argv"]
    assert "pid 4242" in result.output


def test_vm_open_power_on_ignored_without_vmx(monkeypatch: pytest.MonkeyPatch) -> None:
    cap = _capture_launch(monkeypatch)
    monkeypatch.setenv("DISPLAY", ":0")
    result = CliRunner().invoke(cli, ["vmgui", "--power-on"])
    assert result.exit_code == 0, result.output
    assert "-x" not in cap["argv"]
    assert "ignored without --vmx" in result.output