feat(cli): port burn-in/smoke/validate/creds to Python (Phase C1-C6)
Add four native command groups so the Linux host no longer needs pwsh for
manual ops. __main__ registers all four at once, so they ship together.
- bench run (C2): N concurrent jobs in-process via concurrent.futures,
per-round orphan-clone + stale-lock assertions. Replaces
Test-CapacityBurnIn.ps1 + Start-BurnInTest*.ps1. No
Start-Job/pwsh/$IsWindows.
- bench measure (C3): clone/start/IP/transport/destroy timings appended to
benchmark.jsonl (legacy field names kept). Replaces
Measure-CIBenchmark.ps1.
- smoke run (C4): one E2E job; asserts exit 0 + artifact dir + job/success
event. ns7zip/nsinnounp presets. Replaces Test-Smoke.ps1
+ Test-*Build-Linux.ps1.
- validate host (C5): vmrun/snapshot/keyring/permissions/systemd checks.
- validate guest(C6): transport probe with explicit failure cause.
- creds set (C6): keyring writer matching credentials.py read scheme;
password via stdin/prompt only. Replaces
Set-CIGuestCredential.ps1.
All groups import backends.load_backend only (Phase D ESXi path stays open).
Suite green, coverage 95.10% (gate 90%); new modules at 100%.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,487 @@
|
||||
"""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
|
||||
@@ -0,0 +1,184 @@
|
||||
"""Tests for ``creds set`` (C6).
|
||||
|
||||
CLI registration smoke tests (``--help``) are kept; behavioural tests mock
|
||||
the ``keyring`` module and include a round-trip via ``KeyringCredentialStore``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import types
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from ci_orchestrator.__main__ import cli
|
||||
from ci_orchestrator.credentials import KeyringCredentialStore
|
||||
|
||||
# ── --help smoke tests (kept) ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_creds_help_lists_set() -> None:
|
||||
result = CliRunner().invoke(cli, ["creds", "--help"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "set" in result.output
|
||||
|
||||
|
||||
def test_creds_set_help() -> None:
|
||||
result = CliRunner().invoke(cli, ["creds", "set", "--help"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "--target" in result.output
|
||||
assert "--user" in result.output
|
||||
|
||||
|
||||
# ── fake keyring ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _install_fake_keyring(monkeypatch: pytest.MonkeyPatch) -> dict[tuple[str, str], str]:
|
||||
"""Install a fake ``keyring`` module with a two-key password store.
|
||||
|
||||
Returns the backing dict so tests can inspect what was written. The fake
|
||||
implements both ``set_password`` (writer) and ``get_password`` /
|
||||
``get_credential`` (reader), so round-trips work end to end.
|
||||
"""
|
||||
store: dict[tuple[str, str], str] = {}
|
||||
fake = types.ModuleType("keyring")
|
||||
|
||||
def set_password(service: str, key: str, password: str) -> None:
|
||||
store[(service, key)] = password
|
||||
|
||||
def get_password(service: str, key: str) -> str | None:
|
||||
return store.get((service, key))
|
||||
|
||||
def get_credential(service: str, _username: str | None) -> Any:
|
||||
# File backends are a no-op here (mirrors PlaintextKeyring); the reader
|
||||
# falls through to the two-entry scheme.
|
||||
return None
|
||||
|
||||
fake.set_password = set_password # type: ignore[attr-defined]
|
||||
fake.get_password = get_password # type: ignore[attr-defined]
|
||||
fake.get_credential = get_credential # type: ignore[attr-defined]
|
||||
monkeypatch.setitem(sys.modules, "keyring", fake)
|
||||
return store
|
||||
|
||||
|
||||
# ── write scheme ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_creds_set_writes_two_entries(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
store = _install_fake_keyring(monkeypatch)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["creds", "set", "--target", "BuildVMGuest", "--user", "ci_build",
|
||||
"--password-stdin"],
|
||||
input="s3cret\n",
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
# username under "{target}:meta"
|
||||
assert store[("BuildVMGuest:meta", "username")] == "ci_build"
|
||||
# password under target keyed by username
|
||||
assert store[("BuildVMGuest", "ci_build")] == "s3cret"
|
||||
|
||||
|
||||
def test_creds_set_round_trips_via_store(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Round-trip: write with `creds set`, read back with the reader."""
|
||||
_install_fake_keyring(monkeypatch)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["creds", "set", "--target", "BuildVMGuest", "--user", "WINBUILD\\ci",
|
||||
"--password-stdin"],
|
||||
input="p@ss word!\n",
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
cred = KeyringCredentialStore().get("BuildVMGuest")
|
||||
assert cred.username == "WINBUILD\\ci"
|
||||
assert cred.password == "p@ss word!"
|
||||
|
||||
|
||||
def test_creds_set_default_target(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
store = _install_fake_keyring(monkeypatch)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["creds", "set", "--user", "ci_build", "--password-stdin"],
|
||||
input="abc\n",
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert store[("BuildVMGuest:meta", "username")] == "ci_build"
|
||||
|
||||
|
||||
# ── password sourcing ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_creds_set_prompts_when_no_stdin_flag(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
store = _install_fake_keyring(monkeypatch)
|
||||
# click.prompt with confirmation_prompt reads the value twice.
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["creds", "set", "--target", "T", "--user", "u"],
|
||||
input="hunter2\nhunter2\n",
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert store[("T", "u")] == "hunter2"
|
||||
|
||||
|
||||
def test_creds_set_empty_password_rejected(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_install_fake_keyring(monkeypatch)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["creds", "set", "--target", "T", "--user", "u", "--password-stdin"],
|
||||
input="\n",
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "empty" in result.output.lower()
|
||||
|
||||
|
||||
def test_creds_set_strips_trailing_newline(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
store = _install_fake_keyring(monkeypatch)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["creds", "set", "--target", "T", "--user", "u", "--password-stdin"],
|
||||
input="secret\r\n",
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert store[("T", "u")] == "secret"
|
||||
|
||||
|
||||
def test_creds_set_never_accepts_password_option() -> None:
|
||||
"""The CLI must NOT expose a --password option."""
|
||||
result = CliRunner().invoke(cli, ["creds", "set", "--help"])
|
||||
assert "--password " not in result.output
|
||||
assert "--password=" not in result.output
|
||||
|
||||
|
||||
def test_creds_set_keyring_missing(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""When keyring import fails, surface a clean ClickException."""
|
||||
import ci_orchestrator.commands.creds as creds_module
|
||||
|
||||
def _raise(_t: str, _u: str, _p: str) -> None:
|
||||
raise RuntimeError("keyring is not installed; run `pip install keyring`.")
|
||||
|
||||
monkeypatch.setattr(creds_module, "_store_credential", _raise)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["creds", "set", "--target", "T", "--user", "u", "--password-stdin"],
|
||||
input="x\n",
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "keyring is not installed" in result.output
|
||||
|
||||
|
||||
def test_store_credential_invokes_keyring(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Unit-level: _store_credential writes both entries to the module."""
|
||||
import ci_orchestrator.commands.creds as creds_module
|
||||
|
||||
store = _install_fake_keyring(monkeypatch)
|
||||
creds_module._store_credential("Tgt", "user1", "pw1")
|
||||
assert store[("Tgt:meta", "username")] == "user1"
|
||||
assert store[("Tgt", "user1")] == "pw1"
|
||||
@@ -0,0 +1,286 @@
|
||||
"""Tests for ``ci_orchestrator smoke`` (Phase C1 scaffolding + C4 behaviour)."""
|
||||
|
||||
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.smoke as smoke_module
|
||||
from ci_orchestrator.__main__ import cli
|
||||
from ci_orchestrator.config import BackendConfig, Config, Paths
|
||||
|
||||
# ──────────────────────────────────────────────────────── C1 --help smoke
|
||||
|
||||
|
||||
def test_smoke_help_lists_run() -> None:
|
||||
result = CliRunner().invoke(cli, ["smoke", "--help"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "run" in result.output
|
||||
|
||||
|
||||
def test_smoke_run_help() -> None:
|
||||
result = CliRunner().invoke(cli, ["smoke", "run", "--help"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "--guest-os" in result.output
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────── helpers
|
||||
|
||||
|
||||
def _patch_config(monkeypatch: pytest.MonkeyPatch, artifacts: Path) -> None:
|
||||
paths = Paths(
|
||||
root=artifacts.parent,
|
||||
templates=artifacts.parent / "templates",
|
||||
build_vms=artifacts.parent / "build-vms",
|
||||
artifacts=artifacts,
|
||||
keys=artifacts.parent / "keys",
|
||||
)
|
||||
cfg = Config(paths=paths, backend=BackendConfig(), ip_pool=None)
|
||||
monkeypatch.setattr(smoke_module, "load_config", lambda: cfg)
|
||||
|
||||
|
||||
def _patch_job(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
*,
|
||||
artifacts: Path,
|
||||
make_artifact: bool = True,
|
||||
raises: bool = False,
|
||||
) -> dict[str, list[Any]]:
|
||||
"""Replace the in-process ``job`` command with a recording fake."""
|
||||
calls: dict[str, list[Any]] = {"job": []}
|
||||
|
||||
def _fake_callback(**kwargs: Any) -> None:
|
||||
calls["job"].append(kwargs)
|
||||
if make_artifact:
|
||||
job_dir = artifacts / kwargs["job_id"]
|
||||
job_dir.mkdir(parents=True, exist_ok=True)
|
||||
(job_dir / "smoke.txt").write_text("smoke-ok", encoding="utf-8")
|
||||
if raises:
|
||||
import click
|
||||
|
||||
raise click.ClickException("clone failed: boom")
|
||||
|
||||
# ``job_command`` is a click.Command; patch its callback.
|
||||
monkeypatch.setattr(smoke_module.job_command, "callback", _fake_callback)
|
||||
return calls
|
||||
|
||||
|
||||
def _read_events(jsonl_path: Path) -> list[dict[str, Any]]:
|
||||
out: list[dict[str, Any]] = []
|
||||
for line in jsonl_path.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if line:
|
||||
out.append(json.loads(line))
|
||||
return out
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────── C4 behaviour
|
||||
|
||||
|
||||
def test_smoke_run_linux_happy_path(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
calls = _patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["smoke", "run", "--guest-os", "linux", "--job-id", "smoke-1"]
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "PASSED" in result.output
|
||||
|
||||
# The job ran once with the Linux no-op marker preset.
|
||||
assert len(calls["job"]) == 1
|
||||
kw = calls["job"][0]
|
||||
assert kw["guest_os"] == "linux"
|
||||
assert kw["job_id"] == "smoke-1"
|
||||
assert "smoke.txt" in kw["build_command"]
|
||||
assert kw["guest_artifact_source"] == "/opt/ci/output"
|
||||
|
||||
# Artifact dir was created and the success event is present.
|
||||
assert (artifacts / "smoke-1").is_dir()
|
||||
events = _read_events(tmp_path / "logs" / "smoke-1" / "invoke-ci.jsonl")
|
||||
assert any(e["phase"] == "job" and e["status"] == "start" for e in events)
|
||||
assert any(e["phase"] == "job" and e["status"] == "success" for e in events)
|
||||
|
||||
|
||||
def test_smoke_run_windows_uses_windows_preset(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
calls = _patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["smoke", "run", "--guest-os", "windows", "--job-id", "smoke-win"]
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
kw = calls["job"][0]
|
||||
assert kw["guest_os"] == "windows"
|
||||
assert kw["guest_artifact_source"] == "C:\\CI\\output"
|
||||
assert "C:\\CI\\output" in kw["build_command"]
|
||||
|
||||
|
||||
def test_smoke_run_failure_when_job_raises(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
_patch_job(monkeypatch, artifacts=artifacts, make_artifact=False, raises=True)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["smoke", "run", "--guest-os", "linux", "--job-id", "smoke-fail"]
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "FAILED" in result.output
|
||||
# A failure event was logged and there is no success event.
|
||||
events = _read_events(tmp_path / "logs" / "smoke-fail" / "invoke-ci.jsonl")
|
||||
assert any(e["status"] == "failure" for e in events)
|
||||
assert not any(e["status"] == "success" for e in events)
|
||||
|
||||
|
||||
def test_smoke_run_fails_when_artifact_missing(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
# Job "succeeds" but produces no artifact dir → smoke must still fail.
|
||||
_patch_job(monkeypatch, artifacts=artifacts, make_artifact=False)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["smoke", "run", "--guest-os", "linux", "--job-id", "smoke-noart"]
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "artifact dir not found" in result.output
|
||||
|
||||
|
||||
def test_smoke_run_preset_ns7zip(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
calls = _patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["smoke", "run", "--guest-os", "linux", "--preset", "ns7zip", "--job-id", "p1"],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
kw = calls["job"][0]
|
||||
assert "build_plugin.py" in kw["build_command"]
|
||||
assert "7zip-version" in kw["build_command"]
|
||||
assert kw["guest_artifact_source"] == "plugins"
|
||||
|
||||
|
||||
def test_smoke_run_preset_nsinnounp(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
calls = _patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["smoke", "run", "--preset", "nsinnounp", "--job-id", "p2"],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
kw = calls["job"][0]
|
||||
assert kw["build_command"] == "python3 build_plugin.py --final --dist-dir dist"
|
||||
assert kw["guest_artifact_source"] == "dist"
|
||||
|
||||
|
||||
def test_smoke_run_preset_rejected_on_windows(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
_patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["smoke", "run", "--guest-os", "windows", "--preset", "ns7zip"]
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "only supported with --guest-os linux" in result.output
|
||||
|
||||
|
||||
def test_smoke_run_preset_and_build_command_mutually_exclusive(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
_patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"smoke", "run", "--preset", "ns7zip",
|
||||
"--build-command", "echo hi",
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "mutually exclusive" in result.output
|
||||
|
||||
|
||||
def test_smoke_run_custom_build_command(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
calls = _patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"smoke", "run", "--guest-os", "linux",
|
||||
"--build-command", "make all",
|
||||
"--job-id", "custom-1",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
kw = calls["job"][0]
|
||||
assert kw["build_command"] == "make all"
|
||||
# Custom command keeps the OS-default artifact source.
|
||||
assert kw["guest_artifact_source"] == "/opt/ci/output"
|
||||
|
||||
|
||||
def test_has_job_success_tolerates_malformed_jsonl(tmp_path: Path) -> None:
|
||||
"""Garbage/missing log lines must not raise and must report no success."""
|
||||
missing = tmp_path / "nope.jsonl"
|
||||
assert smoke_module._has_job_success(missing) is False
|
||||
|
||||
jsonl = tmp_path / "log.jsonl"
|
||||
jsonl.write_text(
|
||||
"\n" # blank line
|
||||
"not json\n" # JSONDecodeError
|
||||
"[1, 2, 3]\n" # valid JSON but not a dict
|
||||
'{"phase": "job", "status": "start"}\n', # dict, no success
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert smoke_module._has_job_success(jsonl) is False
|
||||
|
||||
jsonl.write_text('{"phase": "job", "status": "success"}\n', encoding="utf-8")
|
||||
assert smoke_module._has_job_success(jsonl) is True
|
||||
|
||||
|
||||
def test_smoke_run_default_job_id_and_separate_log_dir(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
artifacts = tmp_path / "artifacts"
|
||||
logs = tmp_path / "logs"
|
||||
_patch_config(monkeypatch, artifacts)
|
||||
calls = _patch_job(monkeypatch, artifacts=artifacts)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["smoke", "run", "--log-dir", str(logs)]
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
generated_id = calls["job"][0]["job_id"]
|
||||
assert generated_id.startswith("smoke-")
|
||||
# JSONL written under the override log dir, artifacts under config dir.
|
||||
assert (logs / generated_id / "invoke-ci.jsonl").is_file()
|
||||
assert (artifacts / generated_id).is_dir()
|
||||
@@ -0,0 +1,542 @@
|
||||
"""Tests for ``validate host`` (C5) and ``validate guest`` (C6).
|
||||
|
||||
CLI registration smoke tests (``--help``) are kept; behavioural tests mock
|
||||
the backend, keyring, systemctl, and transports.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
import ci_orchestrator.commands.validate as validate_module
|
||||
from ci_orchestrator.__main__ import cli
|
||||
from ci_orchestrator.backends.errors import (
|
||||
BackendNotAvailable,
|
||||
BackendOperationFailed,
|
||||
)
|
||||
from ci_orchestrator.config import BackendConfig, Config, Paths
|
||||
from ci_orchestrator.credentials import Credential
|
||||
from ci_orchestrator.transport.errors import (
|
||||
TransportCommandError,
|
||||
TransportConnectError,
|
||||
)
|
||||
|
||||
# ── --help smoke tests (kept) ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_validate_help_lists_subcommands() -> None:
|
||||
result = CliRunner().invoke(cli, ["validate", "--help"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "host" in result.output
|
||||
assert "guest" in result.output
|
||||
|
||||
|
||||
def test_validate_guest_help() -> None:
|
||||
result = CliRunner().invoke(cli, ["validate", "guest", "--help"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "--host" in result.output
|
||||
|
||||
|
||||
# ── helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_config(tmp_path: Path) -> Config:
|
||||
root = tmp_path / "ci"
|
||||
templates = root / "templates"
|
||||
build_vms = root / "build-vms"
|
||||
artifacts = root / "artifacts"
|
||||
keys = root / "keys"
|
||||
for p in (root, templates, build_vms, artifacts, keys):
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
return Config(
|
||||
paths=Paths(
|
||||
root=root,
|
||||
templates=templates,
|
||||
build_vms=build_vms,
|
||||
artifacts=artifacts,
|
||||
keys=keys,
|
||||
),
|
||||
backend=BackendConfig(type="workstation"),
|
||||
guest_cred_target="BuildVMGuest",
|
||||
)
|
||||
|
||||
|
||||
class _FakeBackend:
|
||||
def __init__(self, snapshots: dict[str, list[str]] | None = None) -> None:
|
||||
self.vmrun_path = "/usr/bin/vmrun"
|
||||
self._snapshots = snapshots or {}
|
||||
|
||||
def list_snapshots(self, handle: Any) -> list[str]:
|
||||
return self._snapshots.get(handle.identifier, [])
|
||||
|
||||
|
||||
def _make_template_vmx(config: Config, name: str) -> str:
|
||||
vmx_dir = config.paths.templates / name
|
||||
vmx_dir.mkdir(parents=True, exist_ok=True)
|
||||
vmx = vmx_dir / f"{name}.vmx"
|
||||
vmx.write_text('config.version = "8"', encoding="utf-8")
|
||||
return str(vmx)
|
||||
|
||||
|
||||
# ── validate host: individual checks ────────────────────────────────────────
|
||||
|
||||
|
||||
def test_check_vmrun_ok(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
monkeypatch.setattr(validate_module, "load_backend", lambda _c: _FakeBackend())
|
||||
result = validate_module._check_vmrun(config)
|
||||
assert result.ok
|
||||
assert "vmrun" in result.detail
|
||||
|
||||
|
||||
def test_check_vmrun_no_path_attr(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
class _NoPath:
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(validate_module, "load_backend", lambda _c: _NoPath())
|
||||
result = validate_module._check_vmrun(config)
|
||||
assert result.ok
|
||||
assert "backend loaded" in result.detail
|
||||
|
||||
|
||||
def test_check_vmrun_unavailable(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
def _raise(_c: Any) -> Any:
|
||||
raise BackendNotAvailable("vmrun missing")
|
||||
|
||||
monkeypatch.setattr(validate_module, "load_backend", _raise)
|
||||
result = validate_module._check_vmrun(config)
|
||||
assert not result.ok
|
||||
assert "missing" in result.detail
|
||||
|
||||
|
||||
def test_check_snapshots_backend_unavailable(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
def _raise(_c: Any) -> Any:
|
||||
raise BackendNotAvailable("no vmrun")
|
||||
|
||||
monkeypatch.setattr(validate_module, "load_backend", _raise)
|
||||
result = validate_module._check_snapshots(config)
|
||||
assert not result.ok
|
||||
assert "backend unavailable" in result.detail
|
||||
|
||||
|
||||
def test_check_snapshots_list_raises(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
_make_template_vmx(config, "LinuxBuild2404")
|
||||
|
||||
class _Backend:
|
||||
vmrun_path = "/usr/bin/vmrun"
|
||||
|
||||
def list_snapshots(self, _handle: Any) -> list[str]:
|
||||
raise BackendOperationFailed("listSnapshots", 1, "boom")
|
||||
|
||||
monkeypatch.setattr(validate_module, "load_backend", lambda _c: _Backend())
|
||||
result = validate_module._check_snapshots(config)
|
||||
assert not result.ok
|
||||
assert "listSnapshots failed" in result.detail
|
||||
|
||||
|
||||
def test_check_keyring_runtime_error(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
class _Store:
|
||||
def get(self, target: str) -> Credential:
|
||||
raise RuntimeError("keyring is not installed")
|
||||
|
||||
monkeypatch.setattr(validate_module, "KeyringCredentialStore", _Store)
|
||||
result = validate_module._check_keyring(config)
|
||||
assert not result.ok
|
||||
assert "keyring is not installed" in result.detail
|
||||
|
||||
|
||||
def test_check_permissions_not_writable(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
def _no_access(_path: Any, _mode: int) -> bool:
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(validate_module.os, "access", _no_access)
|
||||
result = validate_module._check_permissions(config)
|
||||
assert not result.ok
|
||||
assert "not read/write" in result.detail
|
||||
|
||||
|
||||
def test_check_snapshots_ok(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
win_vmx = _make_template_vmx(config, "WinBuild2025")
|
||||
backend = _FakeBackend({win_vmx: ["BaseClean"]})
|
||||
monkeypatch.setattr(validate_module, "load_backend", lambda _c: backend)
|
||||
result = validate_module._check_snapshots(config)
|
||||
assert result.ok, result.detail
|
||||
assert "WinBuild2025->BaseClean" in result.detail
|
||||
|
||||
|
||||
def test_check_snapshots_missing_snapshot(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
lin_vmx = _make_template_vmx(config, "LinuxBuild2404")
|
||||
backend = _FakeBackend({lin_vmx: ["SomethingElse"]})
|
||||
monkeypatch.setattr(validate_module, "load_backend", lambda _c: backend)
|
||||
result = validate_module._check_snapshots(config)
|
||||
assert not result.ok
|
||||
assert "BaseClean-Linux" in result.detail
|
||||
|
||||
|
||||
def test_check_snapshots_no_templates(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
backend = _FakeBackend({})
|
||||
monkeypatch.setattr(validate_module, "load_backend", lambda _c: backend)
|
||||
result = validate_module._check_snapshots(config)
|
||||
assert not result.ok
|
||||
assert "no known template VMX found" in result.detail
|
||||
|
||||
|
||||
def test_check_keyring_ok(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
class _Store:
|
||||
def get(self, target: str) -> Credential:
|
||||
return Credential(username="ci_build", password="x")
|
||||
|
||||
monkeypatch.setattr(validate_module, "KeyringCredentialStore", _Store)
|
||||
result = validate_module._check_keyring(config)
|
||||
assert result.ok
|
||||
assert "ci_build" in result.detail
|
||||
|
||||
|
||||
def test_check_keyring_missing(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
class _Store:
|
||||
def get(self, target: str) -> Credential:
|
||||
raise KeyError(target)
|
||||
|
||||
monkeypatch.setattr(validate_module, "KeyringCredentialStore", _Store)
|
||||
result = validate_module._check_keyring(config)
|
||||
assert not result.ok
|
||||
|
||||
|
||||
def test_check_permissions_ok(tmp_path: Path) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
result = validate_module._check_permissions(config)
|
||||
assert result.ok, result.detail
|
||||
|
||||
|
||||
def test_check_permissions_missing_dir(tmp_path: Path) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
# Remove a required dir.
|
||||
config.paths.build_vms.rmdir()
|
||||
result = validate_module._check_permissions(config)
|
||||
assert not result.ok
|
||||
assert "missing" in result.detail
|
||||
|
||||
|
||||
def test_check_units_no_systemctl(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: None)
|
||||
result = validate_module._check_units()
|
||||
assert not result.ok
|
||||
assert "systemctl" in result.detail
|
||||
|
||||
|
||||
def test_check_units_active(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: "/bin/systemctl")
|
||||
monkeypatch.setattr(validate_module, "_systemctl_is_active", lambda _u: "active")
|
||||
result = validate_module._check_units()
|
||||
assert result.ok
|
||||
|
||||
|
||||
def test_check_units_inactive(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: "/bin/systemctl")
|
||||
monkeypatch.setattr(validate_module, "_systemctl_is_active", lambda _u: "inactive")
|
||||
result = validate_module._check_units()
|
||||
assert not result.ok
|
||||
assert "inactive" in result.detail
|
||||
|
||||
|
||||
def test_systemctl_is_active_invokes_subprocess(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: "/bin/systemctl")
|
||||
|
||||
class _CP:
|
||||
stdout = "active\n"
|
||||
|
||||
monkeypatch.setattr(
|
||||
validate_module.subprocess, "run", lambda *a, **k: _CP()
|
||||
)
|
||||
assert validate_module._systemctl_is_active("act-runner.service") == "active"
|
||||
|
||||
|
||||
def test_systemctl_is_active_no_binary(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: None)
|
||||
assert validate_module._systemctl_is_active("x") is None
|
||||
|
||||
|
||||
def test_systemctl_is_active_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
import subprocess as _sp
|
||||
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: "/bin/systemctl")
|
||||
|
||||
def _boom(*_a: Any, **_k: Any) -> Any:
|
||||
raise _sp.TimeoutExpired(cmd="systemctl", timeout=10.0)
|
||||
|
||||
monkeypatch.setattr(validate_module.subprocess, "run", _boom)
|
||||
assert validate_module._systemctl_is_active("x") is None
|
||||
|
||||
|
||||
# ── validate host: end-to-end command ───────────────────────────────────────
|
||||
|
||||
|
||||
def test_validate_host_all_pass(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
win_vmx = _make_template_vmx(config, "WinBuild2025")
|
||||
backend = _FakeBackend({win_vmx: ["BaseClean"]})
|
||||
|
||||
class _Store:
|
||||
def get(self, target: str) -> Credential:
|
||||
return Credential("ci_build", "x")
|
||||
|
||||
monkeypatch.setattr(validate_module, "load_config", lambda: config)
|
||||
monkeypatch.setattr(validate_module, "load_backend", lambda _c: backend)
|
||||
monkeypatch.setattr(validate_module, "KeyringCredentialStore", _Store)
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: "/bin/systemctl")
|
||||
monkeypatch.setattr(validate_module, "_systemctl_is_active", lambda _u: "active")
|
||||
|
||||
result = CliRunner().invoke(cli, ["validate", "host"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "All host checks passed" in result.output
|
||||
|
||||
|
||||
def test_validate_host_reports_failures_and_exits_nonzero(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
def _raise(_c: Any) -> Any:
|
||||
raise BackendNotAvailable("vmrun missing")
|
||||
|
||||
class _Store:
|
||||
def get(self, target: str) -> Credential:
|
||||
raise KeyError(target)
|
||||
|
||||
monkeypatch.setattr(validate_module, "load_config", lambda: config)
|
||||
monkeypatch.setattr(validate_module, "load_backend", _raise)
|
||||
monkeypatch.setattr(validate_module, "KeyringCredentialStore", _Store)
|
||||
monkeypatch.setattr(validate_module.shutil, "which", lambda _x: None)
|
||||
|
||||
result = CliRunner().invoke(cli, ["validate", "host"])
|
||||
assert result.exit_code == 1, result.output
|
||||
assert "FAIL" in result.output
|
||||
assert "check(s) failed" in result.output
|
||||
|
||||
|
||||
# ── validate guest ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_resolve_use_winrm_explicit() -> None:
|
||||
assert validate_module._resolve_use_winrm(True, None) is True
|
||||
assert validate_module._resolve_use_winrm(False, "windows") is False
|
||||
|
||||
|
||||
def test_resolve_use_winrm_inferred() -> None:
|
||||
assert validate_module._resolve_use_winrm(None, "linux") is False
|
||||
assert validate_module._resolve_use_winrm(None, "windows") is True
|
||||
|
||||
|
||||
def test_resolve_use_winrm_requires_hint() -> None:
|
||||
import click as _click
|
||||
|
||||
with pytest.raises(_click.UsageError):
|
||||
validate_module._resolve_use_winrm(None, None)
|
||||
|
||||
|
||||
class _FakeTransport:
|
||||
"""Records connect/run/close and raises a configured error on run."""
|
||||
|
||||
def __init__(self, *args: Any, raise_on_run: Exception | None = None, **kw: Any) -> None:
|
||||
self._raise = raise_on_run
|
||||
self.closed = False
|
||||
|
||||
def run(self, *_a: Any, **_k: Any) -> Any:
|
||||
if self._raise is not None:
|
||||
raise self._raise
|
||||
|
||||
class _R:
|
||||
returncode = 0
|
||||
|
||||
return _R()
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
|
||||
def _install_keyring_store(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
class _Store:
|
||||
def get(self, target: str) -> Credential:
|
||||
return Credential("Administrator", "pw")
|
||||
|
||||
monkeypatch.setattr(validate_module, "KeyringCredentialStore", _Store)
|
||||
|
||||
|
||||
def test_probe_winrm_ok(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
_install_keyring_store(monkeypatch)
|
||||
transport = _FakeTransport()
|
||||
monkeypatch.setattr(validate_module, "WinRmTransport", lambda *a, **k: transport)
|
||||
result = validate_module._probe_winrm("10.0.0.5", config)
|
||||
assert result.ok
|
||||
assert transport.closed
|
||||
|
||||
|
||||
def test_probe_winrm_auth_fail(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
_install_keyring_store(monkeypatch)
|
||||
transport = _FakeTransport(raise_on_run=TransportConnectError("auth denied"))
|
||||
monkeypatch.setattr(validate_module, "WinRmTransport", lambda *a, **k: transport)
|
||||
result = validate_module._probe_winrm("10.0.0.5", config)
|
||||
assert not result.ok
|
||||
assert "auth denied" in result.detail
|
||||
|
||||
|
||||
def test_probe_winrm_command_fail(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
_install_keyring_store(monkeypatch)
|
||||
transport = _FakeTransport(raise_on_run=TransportCommandError(1, "", "boom"))
|
||||
monkeypatch.setattr(validate_module, "WinRmTransport", lambda *a, **k: transport)
|
||||
result = validate_module._probe_winrm("10.0.0.5", config)
|
||||
assert not result.ok
|
||||
assert "command failed" in result.detail
|
||||
|
||||
|
||||
def test_probe_winrm_credential_missing(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
|
||||
class _Store:
|
||||
def get(self, target: str) -> Credential:
|
||||
raise KeyError(target)
|
||||
|
||||
monkeypatch.setattr(validate_module, "KeyringCredentialStore", _Store)
|
||||
result = validate_module._probe_winrm("10.0.0.5", config)
|
||||
assert not result.ok
|
||||
assert "credential lookup failed" in result.detail
|
||||
|
||||
|
||||
def test_probe_ssh_ok(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
transport = _FakeTransport()
|
||||
monkeypatch.setattr(validate_module, "SshTransport", lambda *a, **k: transport)
|
||||
result = validate_module._probe_ssh("10.0.0.6", config)
|
||||
assert result.ok
|
||||
assert transport.closed
|
||||
|
||||
|
||||
def test_probe_ssh_connect_fail(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
transport = _FakeTransport(raise_on_run=TransportConnectError("no route"))
|
||||
monkeypatch.setattr(validate_module, "SshTransport", lambda *a, **k: transport)
|
||||
result = validate_module._probe_ssh("10.0.0.6", config)
|
||||
assert not result.ok
|
||||
assert "no route" in result.detail
|
||||
|
||||
|
||||
def test_probe_ssh_command_fail(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
transport = _FakeTransport(raise_on_run=TransportCommandError(2, "", "nope"))
|
||||
monkeypatch.setattr(validate_module, "SshTransport", lambda *a, **k: transport)
|
||||
result = validate_module._probe_ssh("10.0.0.6", config)
|
||||
assert not result.ok
|
||||
assert "command failed" in result.detail
|
||||
|
||||
|
||||
def test_validate_guest_ssh_ok(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
transport = _FakeTransport()
|
||||
monkeypatch.setattr(validate_module, "load_config", lambda: config)
|
||||
monkeypatch.setattr(validate_module, "SshTransport", lambda *a, **k: transport)
|
||||
result = CliRunner().invoke(
|
||||
cli, ["validate", "guest", "--host", "10.0.0.6", "--ssh"]
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "OK" in result.output
|
||||
|
||||
|
||||
def test_validate_guest_winrm_fail_exits_nonzero(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
_install_keyring_store(monkeypatch)
|
||||
transport = _FakeTransport(raise_on_run=TransportConnectError("bad creds"))
|
||||
monkeypatch.setattr(validate_module, "load_config", lambda: config)
|
||||
monkeypatch.setattr(validate_module, "WinRmTransport", lambda *a, **k: transport)
|
||||
result = CliRunner().invoke(
|
||||
cli, ["validate", "guest", "--host", "10.0.0.5", "--winrm"]
|
||||
)
|
||||
assert result.exit_code == 1, result.output
|
||||
assert "FAIL" in result.output
|
||||
|
||||
|
||||
def test_validate_guest_requires_transport_hint(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
monkeypatch.setattr(validate_module, "load_config", lambda: config)
|
||||
result = CliRunner().invoke(cli, ["validate", "guest", "--host", "10.0.0.5"])
|
||||
assert result.exit_code != 0
|
||||
assert "transport" in result.output.lower()
|
||||
|
||||
|
||||
def test_validate_guest_infers_from_guest_os(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = _make_config(tmp_path)
|
||||
transport = _FakeTransport()
|
||||
monkeypatch.setattr(validate_module, "load_config", lambda: config)
|
||||
monkeypatch.setattr(validate_module, "SshTransport", lambda *a, **k: transport)
|
||||
result = CliRunner().invoke(
|
||||
cli, ["validate", "guest", "--host", "10.0.0.6", "--guest-os", "linux"]
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
Reference in New Issue
Block a user