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

"""Tests for the ``bench`` sub-commands (Phase C2 burn-in + C3 measure).

The C1 ``--help`` smoke tests are kept; behavioural tests mock the backend
and the per-job runner so no real VMs are spun up.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import pytest
from click.testing import CliRunner

import ci_orchestrator.commands.bench as bench_module
from ci_orchestrator.__main__ import cli
from ci_orchestrator.backends.errors import BackendNotAvailable, BackendOperationFailed
from ci_orchestrator.backends.protocol import VmHandle
from ci_orchestrator.commands.bench import (
    JobResult,
    PhaseTiming,
    _find_orphans,
    _stale_lock,
    bench,
)

# ── C1 smoke tests (kept) ───────────────────────────────────────────────────


def test_bench_help_lists_subcommands() -> None:
    result = CliRunner().invoke(cli, ["bench", "--help"])
    assert result.exit_code == 0, result.output
    assert "run" in result.output
    assert "measure" in result.output


def test_bench_run_help() -> None:
    result = CliRunner().invoke(cli, ["bench", "run", "--help"])
    assert result.exit_code == 0, result.output
    assert "--concurrency" in result.output
    assert "--rounds" in result.output


def test_bench_measure_help() -> None:
    result = CliRunner().invoke(cli, ["bench", "measure", "--help"])
    assert result.exit_code == 0, result.output
    assert "--iterations" in result.output


# ── helpers ─────────────────────────────────────────────────────────────────


def test_find_orphans_matches_job_prefix(tmp_path: Path) -> None:
    (tmp_path / "Clone_burnin-r1-j1-x_20260101_aaa").mkdir()
    (tmp_path / "Clone_other-job_20260101_bbb").mkdir()
    (tmp_path / "not-a-clone").mkdir()
    orphans = _find_orphans(tmp_path, ["burnin-r1-j1-x"])
    assert len(orphans) == 1
    assert orphans[0].name.startswith("Clone_burnin-r1-j1-x_")


def test_find_orphans_missing_base_dir(tmp_path: Path) -> None:
    assert _find_orphans(tmp_path / "ghost", ["j1"]) == []


def test_stale_lock_detects_old_file(tmp_path: Path) -> None:
    import os

    lock = tmp_path / "vm-start.lock"
    lock.write_text("x")
    old = lock.stat().st_mtime - (10 * 60)
    os.utime(lock, (old, old))
    assert _stale_lock(lock) is True


def test_stale_lock_fresh_file_is_ok(tmp_path: Path) -> None:
    lock = tmp_path / "vm-start.lock"
    lock.write_text("x")
    assert _stale_lock(lock) is False


def test_stale_lock_missing_file(tmp_path: Path) -> None:
    assert _stale_lock(tmp_path / "absent.lock") is False


# ── bench run ───────────────────────────────────────────────────────────────


@pytest.fixture
def _ci_paths(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
    """Point load_config at temp build-vms/artifacts dirs."""
    env = {
        "CI_BUILD_VMS": str(tmp_path / "build-vms"),
        "CI_ARTIFACTS": str(tmp_path / "artifacts"),
    }
    (tmp_path / "build-vms").mkdir()
    (tmp_path / "artifacts").mkdir()
    real = bench_module.load_config
    monkeypatch.setattr(bench_module, "load_config", lambda: real(env=env))
    return tmp_path


def test_bench_run_all_pass(monkeypatch: pytest.MonkeyPatch, _ci_paths: Path) -> None:
    calls: list[str] = []

    def _fake_job(*, job_id: str, **_kw: Any) -> tuple[int, str]:
        calls.append(job_id)
        return 0, ""

    monkeypatch.setattr(bench_module, "_run_one_job", _fake_job)
    result = CliRunner().invoke(
        cli, ["bench", "run", "--concurrency", "3", "--rounds", "2"]
    )
    assert result.exit_code == 0, result.output
    assert "OVERALL: PASS" in result.output
    assert len(calls) == 6  # 3 jobs * 2 rounds
    # Report file written under artifacts/bench.
    reports = list((_ci_paths / "artifacts" / "bench").glob("*.json"))
    assert len(reports) == 1
    data = json.loads(reports[0].read_text())
    assert data["overall"] == "PASS"
    assert data["totalJobs"] == 6
    assert len(data["results"]) == 2


def test_bench_run_failing_job_marks_round_fail(
    monkeypatch: pytest.MonkeyPatch, _ci_paths: Path
) -> None:
    def _fake_job(*, job_id: str, slot: int = 0, **_kw: Any) -> tuple[int, str]:
        return 1, "clone failed: boom"

    monkeypatch.setattr(bench_module, "_run_one_job", _fake_job)
    result = CliRunner().invoke(
        cli, ["bench", "run", "--concurrency", "1", "--rounds", "1"]
    )
    assert result.exit_code == 1
    assert "OVERALL: FAIL" in result.output
    assert "clone failed: boom" in result.output


def test_bench_run_orphan_dir_fails_round(
    monkeypatch: pytest.MonkeyPatch, _ci_paths: Path
) -> None:
    base = _ci_paths / "build-vms"

    def _fake_job(*, job_id: str, **_kw: Any) -> tuple[int, str]:
        # Leave an orphan clone dir behind for this job.
        (base / f"Clone_{job_id}_20260101_orphan").mkdir()
        return 0, ""

    monkeypatch.setattr(bench_module, "_run_one_job", _fake_job)
    result = CliRunner().invoke(
        cli, ["bench", "run", "--concurrency", "1", "--rounds", "1"]
    )
    assert result.exit_code == 1
    assert "orphan" in result.output
    assert "OVERALL: FAIL" in result.output


def test_bench_run_stale_lock_fails_round(
    monkeypatch: pytest.MonkeyPatch, _ci_paths: Path
) -> None:
    base = _ci_paths / "build-vms"
    monkeypatch.setattr(bench_module, "_run_one_job", lambda **_kw: (0, ""))
    # _stale_lock reports a stale lock regardless of file age.
    monkeypatch.setattr(bench_module, "_stale_lock", lambda _p: True)
    result = CliRunner().invoke(
        cli,
        ["bench", "run", "--concurrency", "1", "--rounds", "1", "--clone-base-dir", str(base)],
    )
    assert result.exit_code == 1
    assert "stale lock" in result.output


def test_bench_run_custom_json_out(
    monkeypatch: pytest.MonkeyPatch, _ci_paths: Path
) -> None:
    out = _ci_paths / "custom" / "report.json"
    monkeypatch.setattr(bench_module, "_run_one_job", lambda **_kw: (0, ""))
    result = CliRunner().invoke(
        cli,
        ["bench", "run", "--concurrency", "1", "--rounds", "1", "--json-out", str(out)],
    )
    assert result.exit_code == 0, result.output
    assert out.is_file()


def test_run_one_job_invokes_job_command(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    """_run_one_job drives the real job command and maps exit codes."""
    import ci_orchestrator.commands.job as job_module

    captured: dict[str, Any] = {}

    class _FakeResult:
        exit_code = 0
        output = "ok\n"

    class _FakeRunner:
        def invoke(self, _cmd: Any, args: list[str], **_kw: Any) -> _FakeResult:
            captured["args"] = args
            return _FakeResult()

    monkeypatch.setattr("click.testing.CliRunner", lambda *a, **k: _FakeRunner())
    del job_module  # imported for symmetry; not otherwise needed
    code, err = bench_module._run_one_job(
        job_id="j1",
        guest_os="linux",
        build_command="echo hi",
        clone_base_dir=tmp_path,
    )
    assert code == 0
    assert err == ""
    assert "--job-id" in captured["args"]
    assert "--build-command" in captured["args"]


def test_run_one_job_failure_returns_tail(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    class _FakeResult:
        exit_code = 1
        output = "line one\nError: clone failed\n"

    class _FakeRunner:
        def invoke(self, *_a: Any, **_k: Any) -> _FakeResult:
            return _FakeResult()

    monkeypatch.setattr("click.testing.CliRunner", lambda *a, **k: _FakeRunner())
    code, err = bench_module._run_one_job(
        job_id="j1", guest_os="linux", build_command="", clone_base_dir=tmp_path
    )
    assert code == 1
    assert err == "Error: clone failed"


# ── bench measure ───────────────────────────────────────────────────────────


class _FakeBackend:
    def __init__(self, *, ip: str | None = "10.0.0.5", fail: str = "") -> None:
        self._ip = ip
        self._fail = fail
        self.calls: list[str] = []

    def clone_linked(
        self, template: str, snapshot: str, name: str, destination: str | None = None
    ) -> VmHandle:
        self.calls.append("clone")
        if destination:
            d = Path(destination)
            d.parent.mkdir(parents=True, exist_ok=True)
            d.write_text('guestOS = "ubuntu-64"', encoding="utf-8")
        if self._fail == "clone":
            raise BackendOperationFailed("clone", 1, "no")
        return VmHandle(identifier=destination or name, name=name)

    def start(self, handle: VmHandle, headless: bool = True) -> None:
        self.calls.append("start")

    def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
        self.calls.append("get_ip")
        return self._ip

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

    def delete(self, handle: VmHandle) -> None:
        self.calls.append("delete")


def _make_template(tmp_path: Path) -> Path:
    tpl = tmp_path / "tpl" / "LinuxBuild2404.vmx"
    tpl.parent.mkdir(parents=True)
    tpl.write_text('guestOS = "ubuntu-64"', encoding="utf-8")
    return tpl


@pytest.fixture
def _measure_paths(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
    env = {
        "CI_BUILD_VMS": str(tmp_path / "build-vms"),
        "CI_ARTIFACTS": str(tmp_path / "artifacts"),
    }
    real = bench_module.load_config
    monkeypatch.setattr(bench_module, "load_config", lambda: real(env=env))
    return tmp_path


def test_bench_measure_happy_path(
    monkeypatch: pytest.MonkeyPatch, _measure_paths: Path
) -> None:
    tpl = _make_template(_measure_paths)
    backend = _FakeBackend(ip="10.0.0.5")
    monkeypatch.setattr(bench_module, "load_backend", lambda _c: backend)
    # transport ready immediately
    monkeypatch.setattr(bench_module, "_tcp_open", lambda *a, **k: True)

    result = CliRunner().invoke(
        cli,
        [
            "bench",
            "measure",
            "--guest-os",
            "linux",
            "--iterations",
            "2",
            "--template-path",
            str(tpl),
            "--timeout-seconds",
            "30",
        ],
    )
    assert result.exit_code == 0, result.output
    assert "clone" in backend.calls
    assert "delete" in backend.calls
    jsonl = _measure_paths / "artifacts" / "benchmark.jsonl"
    assert jsonl.is_file()
    records = [json.loads(line) for line in jsonl.read_text().splitlines()]
    assert len(records) == 2
    # Legacy field names preserved.
    assert set(records[0]) >= {
        "ts", "runId", "iteration", "template", "snapshot", "guestOS",
        "cloneSec", "startSec", "ipSec", "readySec", "destroySec",
        "totalBootSec", "deltaKB", "vmIP", "error",
    }
    assert records[0]["vmIP"] == "10.0.0.5"
    assert records[0]["error"] is None
    assert "Average boot-to-ready" in result.output


def test_bench_measure_ip_timeout_records_error(
    monkeypatch: pytest.MonkeyPatch, _measure_paths: Path
) -> None:
    tpl = _make_template(_measure_paths)
    backend = _FakeBackend(ip=None)  # never returns an IP
    monkeypatch.setattr(bench_module, "load_backend", lambda _c: backend)
    # Make the IP-wait loop terminate fast: a monotonic clock that jumps
    # 100s per call, so any deadline is exceeded within a couple of polls.
    counter = {"t": 0.0}

    def _fast_now() -> float:
        counter["t"] += 100.0
        return counter["t"]

    monkeypatch.setattr(bench_module, "_now", _fast_now)

    result = CliRunner().invoke(
        cli,
        ["bench", "measure", "--template-path", str(tpl), "--timeout-seconds", "30"],
    )
    assert result.exit_code == 0, result.output
    jsonl = _measure_paths / "artifacts" / "benchmark.jsonl"
    rec = json.loads(jsonl.read_text().splitlines()[0])
    assert rec["error"] is not None
    assert rec["readySec"] is None
    # destroy still ran (finally block).
    assert "delete" in backend.calls


def test_bench_measure_template_missing(
    monkeypatch: pytest.MonkeyPatch, _measure_paths: Path
) -> None:
    result = CliRunner().invoke(
        cli, ["bench", "measure", "--template-path", str(_measure_paths / "nope.vmx")]
    )
    assert result.exit_code != 0
    assert "Template VMX not found" in result.output


def test_bench_measure_backend_unavailable(
    monkeypatch: pytest.MonkeyPatch, _measure_paths: Path
) -> None:
    tpl = _make_template(_measure_paths)

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

    monkeypatch.setattr(bench_module, "load_backend", _raise)
    result = CliRunner().invoke(
        cli, ["bench", "measure", "--template-path", str(tpl)]
    )
    assert result.exit_code != 0
    assert "backend unavailable" in result.output


def test_bench_measure_clone_failure_records_error(
    monkeypatch: pytest.MonkeyPatch, _measure_paths: Path
) -> None:
    tpl = _make_template(_measure_paths)
    backend = _FakeBackend(fail="clone")
    monkeypatch.setattr(bench_module, "load_backend", lambda _c: backend)
    result = CliRunner().invoke(
        cli, ["bench", "measure", "--template-path", str(tpl), "--timeout-seconds", "30"]
    )
    assert result.exit_code == 0, result.output
    jsonl = _measure_paths / "artifacts" / "benchmark.jsonl"
    rec = json.loads(jsonl.read_text().splitlines()[0])
    assert rec["error"] is not None
    assert rec["cloneSec"] is None


def test_bench_measure_transport_port_timeout(
    monkeypatch: pytest.MonkeyPatch, _measure_paths: Path
) -> None:
    tpl = _make_template(_measure_paths)
    backend = _FakeBackend(ip="10.0.0.9")
    monkeypatch.setattr(bench_module, "load_backend", lambda _c: backend)
    # IP is acquired immediately, but the transport port never opens.
    monkeypatch.setattr(bench_module, "_wait_port", lambda *a, **k: False)
    result = CliRunner().invoke(
        cli, ["bench", "measure", "--template-path", str(tpl), "--timeout-seconds", "30"]
    )
    assert result.exit_code == 0, result.output
    jsonl = _measure_paths / "artifacts" / "benchmark.jsonl"
    rec = json.loads(jsonl.read_text().splitlines()[0])
    assert rec["readySec"] is None
    assert "port" in (rec["error"] or "")


def test_bench_measure_destroy_suppresses_backend_errors(
    monkeypatch: pytest.MonkeyPatch, _measure_paths: Path
) -> None:
    tpl = _make_template(_measure_paths)

    class _FlakyBackend(_FakeBackend):
        def stop(self, handle: VmHandle, hard: bool = False) -> None:
            self.calls.append("stop")
            raise BackendOperationFailed("stop", 1, "no")

        def delete(self, handle: VmHandle) -> None:
            self.calls.append("delete")
            raise BackendOperationFailed("delete", 1, "no")

    backend = _FlakyBackend(ip="10.0.0.5")
    monkeypatch.setattr(bench_module, "load_backend", lambda _c: backend)
    monkeypatch.setattr(bench_module, "_tcp_open", lambda *a, **k: True)
    result = CliRunner().invoke(
        cli, ["bench", "measure", "--template-path", str(tpl), "--timeout-seconds", "30"]
    )
    # stop/delete raising must not crash the command.
    assert result.exit_code == 0, result.output
    assert "stop" in backend.calls
    assert "delete" in backend.calls


def test_bench_run_report_write_failure_is_tolerated(
    monkeypatch: pytest.MonkeyPatch, _ci_paths: Path
) -> None:
    monkeypatch.setattr(bench_module, "_run_one_job", lambda **_kw: (0, ""))

    def _boom(_report: Any, _path: Any) -> None:
        raise OSError("disk full")

    monkeypatch.setattr(bench_module, "_write_run_report", _boom)
    result = CliRunner().invoke(
        cli, ["bench", "run", "--concurrency", "1", "--rounds", "1"]
    )
    assert result.exit_code == 0, result.output
    assert "could not write report" in result.output


def test_bench_measure_default_template_windows(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    """--guest-os windows resolves the WinBuild default template + snapshot."""
    from ci_orchestrator.config import load_config as real_load

    cfg = real_load(env={"CI_ROOT": str(tmp_path)})
    tpl, snap = bench_module._resolve_template(cfg, "windows")
    assert tpl.name == "WinBuild2025.vmx"
    assert snap == "BaseClean"
    tpl_l, snap_l = bench_module._resolve_template(cfg, "linux")
    assert tpl_l.name == "LinuxBuild2404.vmx"
    assert snap_l == "BaseClean-Linux"


# ── dataclass behaviour ─────────────────────────────────────────────────────


def test_phase_timing_total_boot_none_when_incomplete() -> None:
    t = PhaseTiming(iteration=1, run_id="r", template="t", snapshot="s", guest_os="linux")
    assert t.total_boot_sec is None
    t.clone_sec, t.start_sec, t.ip_sec, t.ready_sec = 1.0, 2.0, 3.0, 4.0
    assert t.total_boot_sec == 10.0


def test_job_result_dataclass_defaults() -> None:
    jr = JobResult(job_id="j", slot=1, exit_code=0, status="PASS", elapsed_sec=1.0)
    assert jr.error == ""
    assert bench is not None