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

"""Tests for the pure ``operations.vms`` layer (W1 of the web-UI plan).

Migrated from the direct-call tests that used to live in
``test_commands_vm.py`` (``_list_orphans`` / ``_find_vmx`` /
``cleanup_orphans``), adapted to the new dataclass return types. Adds
coverage for ``list_live_vms`` and ``remove_vm``.
"""

from __future__ import annotations

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

import pytest

import ci_orchestrator.operations.vms as vms_module
from ci_orchestrator.backends.errors import BackendOperationFailed
from ci_orchestrator.backends.protocol import VmHandle
from ci_orchestrator.commands.job import _make_clone_name
from ci_orchestrator.config import BackendConfig, Config, Paths
from ci_orchestrator.operations.vms import (
    Orphan,
    ParsedClone,
    RemovalResult,
    cleanup_orphans,
    list_live_vms,
    list_orphans,
    parse_clone_name,
    remove_vm,
)


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


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

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


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


def _make_config(build_vms: Path) -> Config:
    return Config(
        paths=Paths(
            root=build_vms.parent,
            templates=build_vms.parent / "templates",
            build_vms=build_vms,
            artifacts=build_vms.parent / "artifacts",
            keys=build_vms.parent / "keys",
        ),
        backend=BackendConfig(),
    )


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


# ── list_orphans ────────────────────────────────────────────────────────────


def test_list_orphans_non_dir_base(tmp_path: Path) -> None:
    f = tmp_path / "file.txt"
    f.write_text("x")
    assert 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 list_orphans(base, 4) == []


def test_list_orphans_returns_orphan_dataclass(tmp_path: Path) -> None:
    base = tmp_path / "b"
    base.mkdir()
    old, vmx = _make_clone_dir(base, "old")
    _age_dir(old, hours=10)

    orphans = list_orphans(base, 4)
    assert orphans == [Orphan(path=old, vmx=vmx, age_hours=10)]


# ── _find_vmx ────────────────────────────────────────────────────────────


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


# ── cleanup_orphans ────────────────────────────────────────────────────────


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)
    results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=None)
    assert sum(r.dir_removed for r in results) == 1
    assert not no_vmx.exists()


def test_cleanup_orphans_backend_stop_error_swallowed(tmp_path: Path) -> None:
    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")

    results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=_B())
    assert sum(r.dir_removed for r in results) == 1
    assert results[0].stopped is False
    assert results[0].deleted is False
    assert results[0].error is not None


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(vms_module, "_try_remove_dir", lambda _p: False)
    results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=None)
    assert sum(r.dir_removed for r in results) == 0
    assert len(results) == 1


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)
    # Must not raise even though unlink() fails.
    cleanup_orphans(clone_base=base, max_age_hours=4, backend=None, lock_file=lock)


def test_cleanup_orphans_dry_run_no_destroy(tmp_path: Path) -> None:
    base = tmp_path / "b"
    base.mkdir()
    old, vmx = _make_clone_dir(base, "old")
    _age_dir(old, hours=10)
    backend = _FakeBackend()
    results = cleanup_orphans(clone_base=base, max_age_hours=4, backend=backend, dry_run=True)
    assert results == [RemovalResult(target=str(vmx), stopped=False, deleted=False, dir_removed=False, error=None)]
    assert backend.calls == []
    assert old.exists()


def test_cleanup_orphans_no_orphans_returns_empty(tmp_path: Path) -> None:
    base = tmp_path / "b"
    base.mkdir()
    assert cleanup_orphans(clone_base=base, max_age_hours=4, backend=None) == []


def test_cleanup_orphans_lock_respects_dry_run(tmp_path: Path) -> None:
    base = tmp_path / "b"
    base.mkdir()
    lock = tmp_path / "stale.lock"
    lock.write_text("x")
    _age_dir(lock, hours=1)
    cleanup_orphans(clone_base=base, max_age_hours=4, backend=None, dry_run=True, lock_file=lock)
    assert lock.exists()  # dry run never touches the lock file


# ── list_live_vms ────────────────────────────────────────────────────────


class _FakeLiveBackend:
    def __init__(
        self,
        handles: list[VmHandle],
        *,
        ip: str | None = "192.168.1.50",
        ip_raises: bool = False,
        list_running_raises: bool = False,
    ) -> None:
        self._handles = handles
        self._ip = ip
        self._ip_raises = ip_raises
        self._list_running_raises = list_running_raises

    def list_running(self) -> list[VmHandle]:
        if self._list_running_raises:
            raise BackendOperationFailed("list", 1, "boom")
        return self._handles

    def get_ip(self, _h: VmHandle, timeout: float = 0.0) -> str | None:
        if self._ip_raises:
            raise BackendOperationFailed("getGuestIPAddress", 1, "no tools")
        return self._ip


def test_list_live_vms_orphan_flag_on_old_clone(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    clone_dir, vmx = _make_clone_dir(build_vms, "old")
    vmx.write_text('guestOS = "ubuntu-64"', encoding="utf-8")
    _age_dir(clone_dir, hours=10)

    backend = _FakeLiveBackend([VmHandle(identifier=str(vmx), name="old")])
    config = _make_config(build_vms)

    result = list_live_vms(backend, config)  # type: ignore[arg-type]
    assert len(result) == 1
    vm = result[0]
    assert vm.name == "old"
    assert vm.vmx == str(vmx)
    assert vm.ip == "192.168.1.50"
    assert vm.guest_os == "ubuntu-64"
    assert vm.uptime_sec is not None and vm.uptime_sec >= 9 * 3600
    assert vm.is_orphan is True


def test_list_live_vms_fresh_clone_not_orphan(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    clone_dir, vmx = _make_clone_dir(build_vms, "fresh")
    _age_dir(clone_dir, hours=0)

    backend = _FakeLiveBackend([VmHandle(identifier=str(vmx), name=None)])
    config = _make_config(build_vms)

    result = list_live_vms(backend, config)  # type: ignore[arg-type]
    assert len(result) == 1
    vm = result[0]
    assert vm.name == vmx.stem  # name fell back to vmx stem
    assert vm.is_orphan is False


def test_list_live_vms_outside_build_vms_not_orphan(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    (tmp_path / "elsewhere").mkdir()
    elsewhere, vmx = _make_clone_dir(tmp_path / "elsewhere", "other")
    _age_dir(elsewhere, hours=10)

    backend = _FakeLiveBackend([VmHandle(identifier=str(vmx), name="other")])
    config = _make_config(build_vms)

    result = list_live_vms(backend, config)  # type: ignore[arg-type]
    assert len(result) == 1
    vm = result[0]
    assert vm.is_orphan is False
    assert vm.uptime_sec is None


def test_list_live_vms_get_ip_none(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    _, vmx = _make_clone_dir(build_vms, "noip")

    backend = _FakeLiveBackend([VmHandle(identifier=str(vmx))], ip=None)
    config = _make_config(build_vms)

    result = list_live_vms(backend, config)  # type: ignore[arg-type]
    assert result[0].ip is None


def test_list_live_vms_get_ip_raises_is_tolerated(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    _, vmx = _make_clone_dir(build_vms, "iperr")

    backend = _FakeLiveBackend([VmHandle(identifier=str(vmx))], ip_raises=True)
    config = _make_config(build_vms)

    result = list_live_vms(backend, config)  # type: ignore[arg-type]
    assert result[0].ip is None


def test_list_live_vms_list_running_raises_returns_empty(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    backend = _FakeLiveBackend([], list_running_raises=True)
    config = _make_config(build_vms)
    assert list_live_vms(backend, config) == []  # type: ignore[arg-type]


def test_list_live_vms_missing_vmx_file_guest_os_empty(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    missing_vmx = build_vms / "ghost" / "ghost.vmx"  # parent dir doesn't even exist

    backend = _FakeLiveBackend([VmHandle(identifier=str(missing_vmx))])
    config = _make_config(build_vms)

    result = list_live_vms(backend, config)  # type: ignore[arg-type]
    assert result[0].guest_os == ""


def test_list_live_vms_empty_when_nothing_running(tmp_path: Path) -> None:
    build_vms = tmp_path / "build-vms"
    build_vms.mkdir()
    backend = _FakeLiveBackend([])
    config = _make_config(build_vms)
    assert list_live_vms(backend, config) == []  # type: ignore[arg-type]


# ── remove_vm ────────────────────────────────────────────────────────────


def test_remove_vm_happy_path(tmp_path: Path) -> None:
    clone_dir, vmx = _make_clone_dir(tmp_path, "live")
    backend = _FakeBackend()
    result = remove_vm(backend, vmx)  # type: ignore[arg-type]
    assert result == RemovalResult(
        target=str(vmx), stopped=True, deleted=True, dir_removed=True, error=None
    )
    assert not clone_dir.exists()
    assert ("delete", None) in backend.calls


def test_remove_vm_dry_run_no_op(tmp_path: Path) -> None:
    clone_dir, vmx = _make_clone_dir(tmp_path, "dry")
    backend = _FakeBackend()
    result = remove_vm(backend, vmx, dry_run=True)  # type: ignore[arg-type]
    assert result == RemovalResult(
        target=str(vmx), stopped=False, deleted=False, dir_removed=False, error=None
    )
    assert backend.calls == []
    assert clone_dir.exists()


def test_remove_vm_delete_retry_succeeds(tmp_path: Path) -> None:
    clone_dir, vmx = _make_clone_dir(tmp_path, "retry")
    backend = _FakeBackend(delete_fails_first=2)
    result = remove_vm(backend, vmx)  # type: ignore[arg-type]
    delete_calls = [c for c in backend.calls if c[0] == "delete"]
    assert len(delete_calls) == 3  # 2 failures + 1 success
    assert result.deleted is True
    assert not clone_dir.exists()


def test_remove_vm_delete_fails_all_retries(tmp_path: Path) -> None:
    clone_dir, vmx = _make_clone_dir(tmp_path, "neverdeletes")

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

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

    result = remove_vm(_B(), vmx)  # type: ignore[arg-type]
    assert result.deleted is False
    assert result.error is not None
    # Directory removal is still attempted even when delete fails.
    assert not clone_dir.exists()
    assert result.dir_removed is True


def test_remove_vm_stop_error_is_swallowed(tmp_path: Path) -> None:
    _, vmx = _make_clone_dir(tmp_path, "stoperr")

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

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

    result = remove_vm(_B(), vmx)  # type: ignore[arg-type]
    assert result.stopped is False
    assert result.deleted is True
    assert result.error is not None


def test_remove_vm_dir_removal_failure(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    _, vmx = _make_clone_dir(tmp_path, "stubborn")
    backend = _FakeBackend()
    monkeypatch.setattr(vms_module, "_try_remove_dir", lambda _p: False)
    result = remove_vm(backend, vmx)  # type: ignore[arg-type]
    assert result.dir_removed is False


# ── reconcile_ip_leases (TODO §2.8) ───────────────────────────────────────────


def _make_pool(tmp_path: Path) -> Any:
    from ci_orchestrator.ip_pool import IpSlotPool

    return IpSlotPool(
        addresses=["10.0.0.1", "10.0.0.2"],
        netmask="255.255.255.0",
        gateway="10.0.0.254",
        lease_file=tmp_path / "ip-pool.json",
    )


def test_reconcile_ip_leases_releases_dead_clone(tmp_path: Path) -> None:
    from ci_orchestrator.operations.vms import reconcile_ip_leases

    base = tmp_path / "build-vms"
    base.mkdir()
    (base / "clone-alive").mkdir()          # survives on disk
    pool = _make_pool(tmp_path)
    pool.acquire("clone-alive")             # 10.0.0.1
    pool.acquire("clone-dead")              # 10.0.0.2 — no dir on disk

    released = reconcile_ip_leases(base, pool)

    assert released == ["10.0.0.2"]


def test_reconcile_ip_leases_dry_run_no_write(tmp_path: Path) -> None:
    from ci_orchestrator.operations.vms import reconcile_ip_leases

    base = tmp_path / "build-vms"
    base.mkdir()
    pool = _make_pool(tmp_path)
    pool.acquire("clone-dead")              # 10.0.0.1, dir absent

    released = reconcile_ip_leases(base, pool, dry_run=True)

    assert released == ["10.0.0.1"]
    import json

    leases = json.loads((tmp_path / "ip-pool.json").read_text())
    assert leases["10.0.0.1"] == "clone-dead"  # not written


def test_reconcile_ip_leases_unreadable_base_is_noop(tmp_path: Path) -> None:
    from ci_orchestrator.operations.vms import reconcile_ip_leases

    missing = tmp_path / "does-not-exist"
    pool = _make_pool(tmp_path)
    pool.acquire("clone-dead")

    # clone_base cannot be enumerated → must NOT release anything.
    assert reconcile_ip_leases(missing, pool) == []
    import json

    leases = json.loads((tmp_path / "ip-pool.json").read_text())
    assert leases["10.0.0.1"] == "clone-dead"


# ── parse_clone_name (Stage C1) ───────────────────────────────────────────


def test_parse_clone_name_round_trips_real_make_clone_name() -> None:
    """A real ``_make_clone_name`` output must invert cleanly."""
    before = datetime.now(tz=UTC).replace(microsecond=0)
    name = _make_clone_name("build-42")
    after = datetime.now(tz=UTC).replace(microsecond=0)

    parsed = parse_clone_name(name)

    assert parsed.job_id == "build-42"
    assert parsed.started is not None
    # started is a naive datetime built from UTC wall-clock digits -- compare
    # against naive bounds taken from an aware UTC "now" either side of the call.
    assert before.replace(tzinfo=None) <= parsed.started <= after.replace(tzinfo=None) + timedelta(seconds=1)


def test_parse_clone_name_job_id_containing_underscore() -> None:
    parsed = parse_clone_name("Clone_my_job_42_20250101_123456_abc123")

    assert parsed.job_id == "my_job_42"
    assert parsed.started == datetime(2025, 1, 1, 12, 34, 56)


def test_parse_clone_name_non_clone_prefix_degrades_both_to_none() -> None:
    assert parse_clone_name("SomeOtherVMName") == ParsedClone(job_id=None, started=None)


def test_parse_clone_name_too_few_parts_degrades_both_to_none() -> None:
    # No room for date + time + suffix at all.
    assert parse_clone_name("Clone_short") == ParsedClone(job_id=None, started=None)


def test_parse_clone_name_garbage_timestamp_degrades_started_only() -> None:
    parsed = parse_clone_name("Clone_myjob_notadate_notatime_abc123")

    assert parsed.job_id == "myjob"
    assert parsed.started is None


def test_parse_clone_name_empty_job_id_degrades_to_none() -> None:
    # No text between "Clone_" and the timestamp.
    parsed = parse_clone_name("Clone_20250101_123456_abc123")

    assert parsed.job_id is None
    assert parsed.started == datetime(2025, 1, 1, 12, 34, 56)


def test_parse_clone_name_wrong_length_date_degrades_started_only() -> None:
    parsed = parse_clone_name("Clone_job1_2025010_123456_abc123")

    assert parsed.started is None