From d63fca596743762b1f021b4602eb0b498814f9ec Mon Sep 17 00:00:00 2001 From: Simone Date: Sun, 7 Jun 2026 20:15:10 +0200 Subject: [PATCH] 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 --- src/ci_orchestrator/__main__.py | 8 + src/ci_orchestrator/commands/bench.py | 704 +++++++++++++++++++++++ src/ci_orchestrator/commands/creds.py | 87 +++ src/ci_orchestrator/commands/smoke.py | 351 +++++++++++ src/ci_orchestrator/commands/validate.py | 285 +++++++++ tests/python/test_commands_bench.py | 487 ++++++++++++++++ tests/python/test_commands_creds.py | 184 ++++++ tests/python/test_commands_smoke.py | 286 +++++++++ tests/python/test_commands_validate.py | 542 +++++++++++++++++ 9 files changed, 2934 insertions(+) create mode 100644 src/ci_orchestrator/commands/bench.py create mode 100644 src/ci_orchestrator/commands/creds.py create mode 100644 src/ci_orchestrator/commands/smoke.py create mode 100644 src/ci_orchestrator/commands/validate.py create mode 100644 tests/python/test_commands_bench.py create mode 100644 tests/python/test_commands_creds.py create mode 100644 tests/python/test_commands_smoke.py create mode 100644 tests/python/test_commands_validate.py diff --git a/src/ci_orchestrator/__main__.py b/src/ci_orchestrator/__main__.py index f06cabe..3449c11 100644 --- a/src/ci_orchestrator/__main__.py +++ b/src/ci_orchestrator/__main__.py @@ -13,12 +13,16 @@ import click from ci_orchestrator import __version__ from ci_orchestrator.commands.artifacts import artifacts +from ci_orchestrator.commands.bench import bench from ci_orchestrator.commands.build import build +from ci_orchestrator.commands.creds import creds from ci_orchestrator.commands.job import job from ci_orchestrator.commands.monitor import monitor from ci_orchestrator.commands.report import report from ci_orchestrator.commands.retention import retention +from ci_orchestrator.commands.smoke import smoke from ci_orchestrator.commands.template import template +from ci_orchestrator.commands.validate import validate from ci_orchestrator.commands.vm import vm # Re-export so legacy A1 tests that patched ``cli_module.`` keep working. @@ -46,6 +50,10 @@ cli.add_command(report) cli.add_command(retention) cli.add_command(template) cli.add_command(job) +cli.add_command(bench) +cli.add_command(smoke) +cli.add_command(validate) +cli.add_command(creds) if __name__ == "__main__": # pragma: no cover diff --git a/src/ci_orchestrator/commands/bench.py b/src/ci_orchestrator/commands/bench.py new file mode 100644 index 0000000..a140ae7 --- /dev/null +++ b/src/ci_orchestrator/commands/bench.py @@ -0,0 +1,704 @@ +"""``bench`` sub-commands. + +Phase C ports the manual PowerShell benchmark/burn-in tooling to Python so +the Linux host no longer needs ``pwsh``: + +* ``bench run`` — concurrency burn-in (replaces ``Test-CapacityBurnIn.ps1`` + + ``Start-BurnInTest*.ps1``). +* ``bench measure`` — phase timing (replaces ``Measure-CIBenchmark.ps1``). + +Design rule: import ``backends.load_backend`` and the existing transports +only — never ``WorkstationVmrunBackend`` directly (keeps the Phase D ESXi +path open). No ``Start-Job``/``pwsh``/``$IsWindows``. +""" + +from __future__ import annotations + +import contextlib +import json +import shutil +import socket +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import asdict, dataclass, field +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING + +import click + +from ci_orchestrator.backends import load_backend +from ci_orchestrator.backends.errors import BackendError, BackendNotAvailable +from ci_orchestrator.backends.protocol import VmHandle +from ci_orchestrator.config import Config, load_config + +if TYPE_CHECKING: # pragma: no cover + from ci_orchestrator.backends.protocol import VmBackend + + +# Clone directories left behind by a job are named ``Clone__...`` +# (see commands/job.py:_make_clone_name and commands/vm.py:vm_new). +_CLONE_PREFIX = "Clone_" +# A vm-start.lock older than this (seconds) is considered stale leftover. +_STALE_LOCK_SECONDS = 5 * 60 +# Linux guests probe SSH/22; everything else probes WinRM/5986. +_SSH_PORT = 22 +_WINRM_PORT = 5986 + + +def _now() -> float: + return time.monotonic() + + +def _timestamp() -> str: + return datetime.now(tz=UTC).strftime("%Y%m%d-%H%M%S") + + +# ─────────────────────────────────────────────────────────────────── group + + +@click.group("bench") +def bench() -> None: + """Capacity burn-in and phase-timing benchmarks.""" + + +# ═══════════════════════════════════════════════════════════════ bench run + + +@dataclass +class JobResult: + """Outcome of a single burn-in job slot.""" + + job_id: str + slot: int + exit_code: int + status: str + elapsed_sec: float + error: str = "" + + +@dataclass +class RoundResult: + """Outcome of one burn-in round (all slots + cleanup assertions).""" + + round: int + passed: int + failed: int + status: str + elapsed_sec: float + orphans: list[str] = field(default_factory=list) + stale_lock: bool = False + jobs: list[JobResult] = field(default_factory=list) + + +def _run_one_job( + *, + job_id: str, + guest_os: str, + build_command: str, + clone_base_dir: Path, +) -> tuple[int, str]: + """Run a single job end to end, returning ``(exit_code, error_message)``. + + Invokes the in-process ``commands.job.job`` command (the same code path + production uses) rather than re-spawning ``python -m ci_orchestrator`` or + a PowerShell child — closes the ``Start-Job``/``$IsWindows`` shim bug. + + Isolated in its own module-level function so the burn-in harness can be + unit tested by monkeypatching it (no real VMs needed). + """ + from click.testing import CliRunner # local import: drives the job command + + from ci_orchestrator.commands.job import job + + args = [ + "--job-id", + job_id, + "--repo-url", + "", + "--branch", + "main", + "--guest-os", + guest_os, + "--clone-base-dir", + str(clone_base_dir), + "--skip-artifact", + ] + if build_command: + args += ["--build-command", build_command] + result = CliRunner().invoke(job, args, catch_exceptions=True) + if result.exit_code == 0: + return 0, "" + tail = result.output.strip().splitlines() + return result.exit_code, tail[-1] if tail else "" + + +def _find_orphans(clone_base: Path, job_ids: list[str]) -> list[Path]: + """Return clone dirs in ``clone_base`` belonging to the given jobs.""" + if not clone_base.is_dir(): + return [] + prefixes = tuple(f"{_CLONE_PREFIX}{jid}_" for jid in job_ids) + return [ + entry + for entry in clone_base.iterdir() + if entry.is_dir() and entry.name.startswith(prefixes) + ] + + +def _stale_lock(lock_file: Path, max_age_seconds: int = _STALE_LOCK_SECONDS) -> bool: + """True if ``lock_file`` exists and is older than ``max_age_seconds``.""" + if not lock_file.is_file(): + return False + try: + mtime = lock_file.stat().st_mtime + except OSError: + return False + return (time.time() - mtime) > max_age_seconds + + +def _run_round( + *, + round_no: int, + concurrency: int, + guest_os: str, + build_command: str, + clone_base_dir: Path, + lock_file: Path, +) -> RoundResult: + """Launch ``concurrency`` concurrent jobs and assert cleanup invariants.""" + click.echo(f"\n[bench run] Round {round_no}: launching {concurrency} job(s)...") + started = _now() + ts = _timestamp() + slots = { + slot: f"burnin-r{round_no}-j{slot}-{ts}" + for slot in range(1, concurrency + 1) + } + + jobs: list[JobResult] = [] + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = { + pool.submit( + _run_one_job, + job_id=job_id, + guest_os=guest_os, + build_command=build_command, + clone_base_dir=clone_base_dir, + ): (slot, job_id) + for slot, job_id in slots.items() + } + for fut in as_completed(futures): + slot, job_id = futures[fut] + try: + exit_code, error = fut.result() + except Exception as exc: # pragma: no cover - defensive + exit_code, error = 1, str(exc) + status = "PASS" if exit_code == 0 else "FAIL" + jobs.append( + JobResult( + job_id=job_id, + slot=slot, + exit_code=exit_code, + status=status, + elapsed_sec=round(_now() - started, 2), + error=error, + ) + ) + + jobs.sort(key=lambda j: j.slot) + passed = sum(1 for j in jobs if j.status == "PASS") + failed = len(jobs) - passed + + # ── Assertion: no orphaned clone directories ───────────────────────── + orphans = _find_orphans(clone_base_dir, list(slots.values())) + if orphans: + failed += 1 + + # ── Assertion: no stale vm-start.lock ──────────────────────────────── + stale = _stale_lock(lock_file) + if stale: + failed += 1 + + status = "PASS" if failed == 0 else "FAIL" + result = RoundResult( + round=round_no, + passed=passed, + failed=failed, + status=status, + elapsed_sec=round(_now() - started, 2), + orphans=[str(p) for p in orphans], + stale_lock=stale, + jobs=jobs, + ) + + for j in jobs: + line = f" slot {j.slot} {j.job_id:<32} {j.status:<5} exit={j.exit_code}" + if j.error: + line += f" ({j.error})" + click.echo(line) + click.echo( + f" clone cleanup : {'OK' if not orphans else f'FAIL ({len(orphans)} orphan)'}" + ) + for p in result.orphans: + click.echo(f" orphan: {p}") + click.echo(f" lock cleanup : {'OK' if not stale else 'FAIL (stale lock)'}") + click.echo( + f" [Round {round_no}] {status} " + f"({passed} PASS, {failed} FAIL, {result.elapsed_sec}s)" + ) + return result + + +def _write_run_report(report: dict[str, object], json_out: Path) -> None: + json_out.parent.mkdir(parents=True, exist_ok=True) + json_out.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + + +@bench.command("run") +@click.option( + "--guest-os", + type=click.Choice(["windows", "linux"], case_sensitive=False), + default="linux", + show_default=True, + help="Guest OS family of the build VMs under test.", +) +@click.option( + "--concurrency", + type=click.IntRange(1, 32), + default=4, + show_default=True, + help="Number of concurrent jobs per round.", +) +@click.option( + "--rounds", + type=click.IntRange(1, 1000), + default=10, + show_default=True, + help="Number of sequential rounds to run.", +) +@click.option( + "--build-command", + "build_command", + default="", + help="Build command each job runs (default: a no-op sleep/ping).", +) +@click.option( + "--clone-base-dir", + "clone_base_dir", + default=None, + help="Override clone base dir (defaults to config.paths.build_vms).", +) +@click.option( + "--json-out", + "json_out", + default=None, + help="Path for the JSON report (default: CI_ARTIFACTS/bench/.json).", +) +def bench_run( + guest_os: str, + concurrency: int, + rounds: int, + build_command: str, + clone_base_dir: str | None, + json_out: str | None, +) -> None: + """Run a concurrency burn-in and assert cleanup invariants per round. + + Replaces ``Test-CapacityBurnIn.ps1`` + ``Start-BurnInTest*.ps1``. For + each of ``--rounds`` rounds it launches ``--concurrency`` concurrent + jobs (in-process, via :func:`concurrent.futures`), then asserts: every + job exited 0, no orphaned clone dir remains in the clone base dir, and + no ``vm-start.lock`` older than 5 minutes is left behind. + """ + guest_os = guest_os.lower() + config = load_config() + base = Path(clone_base_dir) if clone_base_dir else config.paths.build_vms + lock_file = base / "vm-start.lock" + + started_at = datetime.now(tz=UTC) + out_path = ( + Path(json_out) + if json_out + else config.paths.artifacts / "bench" / f"{_timestamp()}.json" + ) + + click.echo("=" * 60) + click.echo("[bench run] capacity burn-in") + click.echo(f" guest-os : {guest_os}") + click.echo(f" concurrency : {concurrency}") + click.echo(f" rounds : {rounds}") + click.echo(f" clone base : {base}") + click.echo(f" build cmd : {build_command or '(default no-op)'}") + click.echo("=" * 60) + + round_results: list[RoundResult] = [] + for r in range(1, rounds + 1): + round_results.append( + _run_round( + round_no=r, + concurrency=concurrency, + guest_os=guest_os, + build_command=build_command, + clone_base_dir=base, + lock_file=lock_file, + ) + ) + + total_pass = sum(rr.passed for rr in round_results) + total_fail = sum(rr.failed for rr in round_results) + rounds_failed = sum(1 for rr in round_results if rr.status == "FAIL") + overall = "PASS" if total_fail == 0 else "FAIL" + + # ── Summary table ──────────────────────────────────────────────────── + click.echo("\n" + "=" * 60) + click.echo("[bench run] SUMMARY") + click.echo(f" {'Round':<7}{'Pass':<6}{'Fail':<6}{'Elapsed':<10}Status") + click.echo(f" {'-' * 5:<7}{'-' * 4:<6}{'-' * 4:<6}{'-' * 8:<10}{'-' * 6}") + for rr in round_results: + click.echo( + f" {rr.round:<7}{rr.passed:<6}{rr.failed:<6}" + f"{str(rr.elapsed_sec) + 's':<10}{rr.status}" + ) + click.echo("") + click.echo(f" Total jobs : {rounds * concurrency}") + click.echo(f" Passed : {total_pass}") + click.echo(f" Failed : {total_fail}") + click.echo(f" Rounds OK : {rounds - rounds_failed} / {rounds}") + click.echo(f"\n OVERALL: {overall}") + click.echo("=" * 60) + + report: dict[str, object] = { + "ts": started_at.isoformat(), + "guestOS": guest_os, + "concurrency": concurrency, + "rounds": rounds, + "buildCommand": build_command, + "totalJobs": rounds * concurrency, + "totalPass": total_pass, + "totalFail": total_fail, + "roundsFailed": rounds_failed, + "overall": overall, + "results": [asdict(rr) for rr in round_results], + } + try: + _write_run_report(report, out_path) + click.echo(f"[bench run] report written: {out_path}") + except OSError as exc: + click.echo(f"[bench run] could not write report: {exc}", err=True) + + if total_fail > 0: + raise SystemExit(1) + + +# ═══════════════════════════════════════════════════════════ bench measure + + +@dataclass +class PhaseTiming: + """Per-iteration phase timings, mirroring ``Measure-CIBenchmark.ps1``. + + Field names of the serialised record (see :meth:`to_record`) are kept + identical to the PowerShell version for trend-log continuity. + """ + + iteration: int + run_id: str + template: str + snapshot: str + guest_os: str + clone_sec: float | None = None + start_sec: float | None = None + ip_sec: float | None = None + ready_sec: float | None = None + destroy_sec: float | None = None + delta_kb: int | None = None + vm_ip: str | None = None + error: str | None = None + + @property + def total_boot_sec(self) -> float | None: + if self.ip_sec is None or self.ready_sec is None: + return None + return round( + (self.clone_sec or 0.0) + + (self.start_sec or 0.0) + + self.ip_sec + + self.ready_sec, + 2, + ) + + def to_record(self, ts: str) -> dict[str, object]: + """Build a JSON record with the legacy ``Measure-CIBenchmark`` keys.""" + return { + "ts": ts, + "runId": self.run_id, + "iteration": self.iteration, + "template": self.template, + "snapshot": self.snapshot, + "guestOS": self.guest_os, + "cloneSec": self.clone_sec, + "startSec": self.start_sec, + "ipSec": self.ip_sec, + "readySec": self.ready_sec, + "destroySec": self.destroy_sec, + "totalBootSec": self.total_boot_sec, + "deltaKB": self.delta_kb, + "vmIP": self.vm_ip, + "error": self.error, + } + + +def _dir_size_kb(path: Path) -> int | None: + """Total size of files under ``path`` in KiB, or None on error.""" + try: + total = sum(p.stat().st_size for p in path.rglob("*") if p.is_file()) + except OSError: + return None + return round(total / 1024) + + +def _tcp_open(host: str, port: int, timeout: float = 3.0) -> bool: + """Return True if a TCP connection to ``host:port`` succeeds.""" + try: + with socket.create_connection((host, port), timeout=timeout): + return True + except OSError: + return False + + +def _wait_port(host: str, port: int, deadline: float, poll: float = 3.0) -> bool: + while _now() < deadline: + if _tcp_open(host, port): + return True + time.sleep(poll) + return False + + +def _measure_iteration( + *, + backend: VmBackend, + iteration: int, + template: Path, + snapshot: str, + guest_os: str, + clone_base_dir: Path, + transport_port: int, + timeout_seconds: float, +) -> PhaseTiming: + """Clone → start → IP → transport-ready → destroy, timing each phase.""" + run_id = f"bench-{_timestamp()}-{iteration}" + clone_name = f"{_CLONE_PREFIX}{run_id}" + clone_dir = clone_base_dir / clone_name + clone_vmx = clone_dir / f"{clone_name}.vmx" + timing = PhaseTiming( + iteration=iteration, + run_id=run_id, + template=template.name, + snapshot=snapshot, + guest_os=guest_os, + ) + handle: VmHandle | None = None + try: + # ── clone ──────────────────────────────────────────────────────── + click.echo(f"[bench measure] iteration {iteration}: cloning...") + t0 = _now() + handle = backend.clone_linked( + template=str(template), + snapshot=snapshot, + name=clone_name, + destination=str(clone_vmx), + ) + timing.clone_sec = round(_now() - t0, 2) + timing.delta_kb = _dir_size_kb(clone_dir) + + # ── start ──────────────────────────────────────────────────────── + click.echo("[bench measure] starting...") + t0 = _now() + backend.start(handle, headless=True) + timing.start_sec = round(_now() - t0, 2) + + # ── IP acquire ──────────────────────────────────────────────────── + click.echo("[bench measure] waiting for guest IP...") + t0 = _now() + deadline = t0 + timeout_seconds + vm_ip: str | None = None + while _now() < deadline: + vm_ip = backend.get_ip(handle, timeout=min(5.0, deadline - _now())) + if vm_ip: + break + if not vm_ip: + raise TimeoutError(f"no guest IP after {timeout_seconds}s") + timing.ip_sec = round(_now() - t0, 2) + timing.vm_ip = vm_ip + + # ── transport ready ─────────────────────────────────────────────── + click.echo(f"[bench measure] waiting for port {transport_port}...") + t0 = _now() + if not _wait_port(vm_ip, transport_port, _now() + timeout_seconds): + raise TimeoutError( + f"port {transport_port} on {vm_ip} not open after {timeout_seconds}s" + ) + timing.ready_sec = round(_now() - t0, 2) + except (BackendError, TimeoutError, OSError) as exc: + timing.error = str(exc) + click.echo(f"[bench measure] iteration {iteration} failed: {exc}", err=True) + finally: + # ── destroy ─────────────────────────────────────────────────────── + t0 = _now() + if handle is not None: + with contextlib.suppress(BackendError): + backend.stop(handle, hard=True) + with contextlib.suppress(BackendError): + backend.delete(handle) + if clone_dir.exists(): + shutil.rmtree(clone_dir, ignore_errors=True) + timing.destroy_sec = round(_now() - t0, 2) + return timing + + +def _resolve_template(config: Config, guest_os: str) -> tuple[Path, str]: + """Pick a default template VMX + snapshot for ``guest_os``.""" + if guest_os == "linux": + return ( + config.paths.templates / "LinuxBuild2404" / "LinuxBuild2404.vmx", + "BaseClean-Linux", + ) + return ( + config.paths.templates / "WinBuild2025" / "WinBuild2025.vmx", + "BaseClean", + ) + + +@bench.command("measure") +@click.option( + "--guest-os", + type=click.Choice(["windows", "linux"], case_sensitive=False), + default="linux", + show_default=True, + help="Guest OS family of the VM to benchmark.", +) +@click.option( + "--iterations", + type=click.IntRange(1, 100), + default=1, + show_default=True, + help="Number of ephemeral VMs to time.", +) +@click.option( + "--template-path", + "template_path", + default=None, + help="Template VMX path (defaults to the per-guest-os template).", +) +@click.option( + "--snapshot-name", + "snapshot_name", + default=None, + help="Snapshot to clone from (defaults to BaseClean / BaseClean-Linux).", +) +@click.option( + "--clone-base-dir", + "clone_base_dir", + default=None, + help="Override clone base dir (defaults to config.paths.build_vms).", +) +@click.option( + "--timeout-seconds", + "timeout_seconds", + type=click.FloatRange(30.0, 1200.0), + default=300.0, + show_default=True, + help="Per-iteration timeout for IP and transport readiness.", +) +@click.option( + "--json-out", + "json_out", + default=None, + help="Path to append JSONL records (default: CI_ARTIFACTS/benchmark.jsonl).", +) +def bench_measure( + guest_os: str, + iterations: int, + template_path: str | None, + snapshot_name: str | None, + clone_base_dir: str | None, + timeout_seconds: float, + json_out: str | None, +) -> None: + """Measure clone/start/IP/transport/destroy phase timings. + + Replaces ``Measure-CIBenchmark.ps1``. Records are appended to + ``benchmark.jsonl`` with the same field names so historical trend data + stays comparable. + """ + guest_os = guest_os.lower() + config = load_config() + + default_tpl, default_snap = _resolve_template(config, guest_os) + template = Path(template_path) if template_path else default_tpl + snapshot = snapshot_name or default_snap + base = Path(clone_base_dir) if clone_base_dir else config.paths.build_vms + transport_port = _SSH_PORT if guest_os == "linux" else _WINRM_PORT + out_path = Path(json_out) if json_out else config.paths.artifacts / "benchmark.jsonl" + + if not template.is_file(): + raise click.ClickException(f"Template VMX not found: {template}") + + try: + backend = load_backend(config) + except BackendNotAvailable as exc: + raise click.ClickException(f"backend unavailable: {exc}") from exc + + base.mkdir(parents=True, exist_ok=True) + out_path.parent.mkdir(parents=True, exist_ok=True) + + click.echo("=" * 60) + click.echo("[bench measure] phase timing") + click.echo(f" template : {template.name}") + click.echo(f" snapshot : {snapshot}") + click.echo(f" guest-os : {guest_os} (port {transport_port})") + click.echo(f" iterations : {iterations}") + click.echo("=" * 60) + + timings: list[PhaseTiming] = [] + for i in range(1, iterations + 1): + timing = _measure_iteration( + backend=backend, + iteration=i, + template=template, + snapshot=snapshot, + guest_os=guest_os, + clone_base_dir=base, + transport_port=transport_port, + timeout_seconds=timeout_seconds, + ) + timings.append(timing) + record = timing.to_record(datetime.now(tz=UTC).isoformat()) + try: + with out_path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record) + "\n") + except OSError as exc: + click.echo(f"[bench measure] could not append jsonl: {exc}", err=True) + + # ── Summary table ──────────────────────────────────────────────────── + click.echo("\n[bench measure] Results") + header = f" {'Iter':<5}{'Clone':<8}{'Start':<8}{'IP':<8}{'Ready':<8}{'Destroy':<9}{'Boot':<8}IP" + click.echo(header) + for t in timings: + click.echo( + f" {t.iteration:<5}" + f"{t.clone_sec!s:<8}{t.start_sec!s:<8}{t.ip_sec!s:<8}" + f"{t.ready_sec!s:<8}{t.destroy_sec!s:<9}" + f"{t.total_boot_sec!s:<8}{t.vm_ip or '-'}" + ) + if t.error: + click.echo(f" error: {t.error}") + + successful = [t for t in timings if t.error is None and t.total_boot_sec is not None] + if len(successful) > 1: + boots = [t.total_boot_sec for t in successful if t.total_boot_sec is not None] + avg = round(sum(boots) / len(boots), 2) + click.echo(f"\n Average boot-to-ready ({len(successful)} ok): {avg}s") + click.echo(f"\n[bench measure] records appended to: {out_path}") + + +__all__ = ["bench"] diff --git a/src/ci_orchestrator/commands/creds.py b/src/ci_orchestrator/commands/creds.py new file mode 100644 index 0000000..c0ea44d --- /dev/null +++ b/src/ci_orchestrator/commands/creds.py @@ -0,0 +1,87 @@ +"""``creds`` sub-commands. + +Phase C replaces ``Set-CIGuestCredential.ps1`` on the Linux host with a +keyring-native writer (``keyrings.alt.file.PlaintextKeyring``), matching the +read path in :mod:`ci_orchestrator.credentials`. + +* ``creds set`` — store a guest username/password under a target. Passwords + are never passed on the command line. + +Write scheme (must mirror :meth:`KeyringCredentialStore.get`): + +* ``keyring.set_password("{target}:meta", "username", )`` — so the + reader can recover the username for file backends where + ``get_credential(target, None)`` is a no-op. +* ``keyring.set_password(target, , )`` — the password, + keyed by the username under the target service. +""" + +from __future__ import annotations + +import sys + +import click + + +def _store_credential(target: str, user: str, password: str) -> None: + """Write the two keyring entries the credential reader expects. + + Raises :class:`RuntimeError` if ``keyring`` is not installed. + """ + try: + import keyring + except ImportError as exc: # pragma: no cover - dep declared in pyproject + raise RuntimeError( + "keyring is not installed; run `pip install keyring`." + ) from exc + + # 1. Username under the "{target}:meta" service, key "username". + keyring.set_password(f"{target}:meta", "username", user) + # 2. Password under the target service, keyed by the username. + keyring.set_password(target, user, password) + + +@click.group("creds") +def creds() -> None: + """Manage guest credentials in the keyring.""" + + +@creds.command("set") +@click.option( + "--target", + default="BuildVMGuest", + show_default=True, + help="Credential target name (matches config.guest_cred_target).", +) +@click.option("--user", "user", required=True, help="Guest username to store.") +@click.option( + "--password-stdin", + "password_stdin", + is_flag=True, + default=False, + help="Read the password from stdin instead of prompting.", +) +def creds_set(target: str, user: str, password_stdin: bool) -> None: + """Store guest credentials so ci_orchestrator can read them back.""" + if password_stdin: + # Read the whole of stdin and strip a single trailing newline only, + # so passwords containing internal whitespace survive intact. + password = sys.stdin.readline().rstrip("\n").rstrip("\r") + else: + password = click.prompt( + "Guest password", + hide_input=True, + confirmation_prompt=True, + ) + if not password: + raise click.UsageError("password must not be empty.") + + try: + _store_credential(target, user, password) + except RuntimeError as exc: + raise click.ClickException(str(exc)) from exc + + click.echo(f"Stored credential for target {target!r} (user={user}).") + + +__all__ = ["creds"] diff --git a/src/ci_orchestrator/commands/smoke.py b/src/ci_orchestrator/commands/smoke.py new file mode 100644 index 0000000..9b544a0 --- /dev/null +++ b/src/ci_orchestrator/commands/smoke.py @@ -0,0 +1,351 @@ +"""``smoke`` sub-commands. + +Phase C4 ports the manual smoke / E2E build tooling to Python so the Linux +host no longer needs ``pwsh``: + +* ``smoke run`` — end-to-end no-op (or real build) job, replacing + ``Test-Smoke.ps1`` and ``Test-*Build-Linux.ps1``. + +The command runs **one** full end-to-end job through the in-process +``job`` orchestrator (reusing ``commands/job.py`` — no re-spawn of +``python -m ci_orchestrator``) and then asserts the three smoke +invariants of ``Test-Smoke.ps1``: + +1. the pipeline exited 0, +2. the per-job artifact directory was created, and +3. a ``job``/``success`` event is present in the ``invoke-ci.jsonl`` + event log (the same JSONL format ``report job`` consumes). + +Because the in-process ``job`` command does not itself emit the JSONL +event log (that was the job of the retired ``Invoke-CIJob.ps1``), this +command writes the ``job``/``start`` and ``job``/``success`` events +around the orchestration, exactly as the PowerShell wrapper did. + +Design rule: import ``backends.load_backend`` and the existing transports +only — never ``WorkstationVmrunBackend`` directly (keeps the Phase D ESXi +path open). +""" + +from __future__ import annotations + +import json +import time +from datetime import UTC, datetime +from pathlib import Path + +import click + +from ci_orchestrator.commands.job import job as job_command +from ci_orchestrator.config import Config, load_config + +# ─────────────────────────────────────────────────────── OS presets + +# Per-guest defaults mirroring ``Test-Smoke.ps1``: a trivial build that +# writes a marker file into the standard output directory, plus the +# matching template / snapshot / artifact-source for the OS family. +_LINUX_TEMPLATE = "/var/lib/ci/templates/LinuxBuild2404/LinuxBuild2404.vmx" +_WINDOWS_TEMPLATE = r"F:\CI\Templates\WinBuild2025\WinBuild2025.vmx" + +_NOOP_BUILD_LINUX = "mkdir -p /opt/ci/output && echo smoke-ok > /opt/ci/output/smoke.txt" +_NOOP_BUILD_WINDOWS = ( + "New-Item -ItemType Directory -Path C:\\CI\\output -Force | Out-Null; " + 'Set-Content C:\\CI\\output\\smoke.txt "smoke-ok"' +) + +# Real Linux build E2E presets (ports of ``Test-*Build-Linux.ps1``). +# Exposed via ``--preset`` so the no-op default is never overridden by a +# hard-coded ``F:\`` Windows path. Each preset supplies a build command +# and the in-guest artifact source directory. +_PRESETS: dict[str, tuple[str, str]] = { + "ns7zip": ( + "python3 build_plugin.py --host linux --7zip-version 26.01 " + "--configs x64-unicode --verbose", + "plugins", + ), + "nsinnounp": ( + "python3 build_plugin.py --final --dist-dir dist", + "dist", + ), +} + + +def _default_template(guest_os: str) -> str: + return _LINUX_TEMPLATE if guest_os == "linux" else _WINDOWS_TEMPLATE + + +def _default_snapshot(guest_os: str) -> str: + return "BaseClean-Linux" if guest_os == "linux" else "BaseClean" + + +def _noop_build_command(guest_os: str) -> tuple[str, str]: + """Return ``(build_command, guest_artifact_source)`` for the marker job.""" + if guest_os == "linux": + return _NOOP_BUILD_LINUX, "/opt/ci/output" + return _NOOP_BUILD_WINDOWS, "C:\\CI\\output" + + +def _utcnow_iso() -> str: + return datetime.now(tz=UTC).isoformat() + + +def _write_event( + jsonl_path: Path, + *, + job_id: str, + phase: str, + status: str, + elapsed_sec: float | None = None, + error: str | None = None, +) -> None: + """Append one event to ``invoke-ci.jsonl`` in ``report job`` format.""" + data: dict[str, object] = {} + if elapsed_sec is not None: + data["elapsedSec"] = int(elapsed_sec) + if error is not None: + data["error"] = error + record: dict[str, object] = { + "jobId": job_id, + "phase": phase, + "status": status, + "ts": _utcnow_iso(), + "data": data, + } + jsonl_path.parent.mkdir(parents=True, exist_ok=True) + with jsonl_path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record) + "\n") + + +def _has_job_success(jsonl_path: Path) -> bool: + """True iff the JSONL log contains a ``job``/``success`` event.""" + try: + text = jsonl_path.read_text(encoding="utf-8") + except OSError: + return False + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(obj, dict): + continue + if obj.get("phase") == "job" and obj.get("status") == "success": + return True + return False + + +# ────────────────────────────────────────────────────────── command + + +@click.group("smoke") +def smoke() -> None: + """End-to-end smoke and build validation.""" + + +@smoke.command("run") +@click.option( + "--guest-os", + type=click.Choice(["windows", "linux"], case_sensitive=False), + default="linux", + show_default=True, + help="Guest OS family to smoke-test.", +) +@click.option( + "--build-command", + "build_command", + default="", + help="Build command to run (default: a marker no-op job).", +) +@click.option( + "--preset", + type=click.Choice(sorted(_PRESETS), case_sensitive=False), + default=None, + help="Real Linux build E2E preset (ns7zip / nsinnounp). Linux only.", +) +@click.option( + "--repo-url", + "repo_url", + default="https://example.invalid/smoke-test.git", + show_default=True, + help="Repository to build (default: a placeholder; the no-op marker " + "job never depends on the clone).", +) +@click.option("--branch", default="main", show_default=True) +@click.option( + "--job-id", + "job_id", + default=None, + help="Job id (default: smoke-).", +) +@click.option( + "--template-path", + "template_path", + default=None, + help="Template VMX path (default: per-guest-os standard template).", +) +@click.option( + "--snapshot-name", + "snapshot_name", + default=None, + help="Snapshot to clone from (default: per-guest-os BaseClean[-Linux]).", +) +@click.option( + "--ssh-key-path", + "ssh_key_path", + default=None, + help="SSH private key for the guest (Linux).", +) +@click.option( + "--log-dir", + "log_dir", + default=None, + help="Base directory for invoke-ci.jsonl logs (default: CI_ARTIFACTS).", +) +def smoke_run( + guest_os: str, + build_command: str, + preset: str | None, + repo_url: str, + branch: str, + job_id: str | None, + template_path: str | None, + snapshot_name: str | None, + ssh_key_path: str | None, + log_dir: str | None, +) -> None: + """Run one end-to-end job and assert success + artifact + event.""" + guest_os = guest_os.lower() + + if preset is not None: + preset = preset.lower() + if guest_os != "linux": + raise click.UsageError("--preset is only supported with --guest-os linux.") + if build_command: + raise click.UsageError("--preset and --build-command are mutually exclusive.") + preset_cmd, artifact_source = _PRESETS[preset] + effective_build_command = preset_cmd + elif build_command: + effective_build_command = build_command + # A custom command writes to the standard output dir for the OS. + _, artifact_source = _noop_build_command(guest_os) + else: + effective_build_command, artifact_source = _noop_build_command(guest_os) + + config: Config = load_config() + resolved_job_id = job_id or f"smoke-{datetime.now(tz=UTC).strftime('%Y%m%d-%H%M%S')}" + resolved_template = template_path or _default_template(guest_os) + resolved_snapshot = snapshot_name or _default_snapshot(guest_os) + + artifact_base: Path = config.paths.artifacts + # Logs live alongside (not inside) the artifacts tree so that writing the + # event log never accidentally creates the per-job artifact directory and + # masks a build that produced no artifacts (mirrors F:\CI\Logs vs + # F:\CI\Artifacts in Test-Smoke.ps1). + log_base = Path(log_dir) if log_dir else config.paths.root / "logs" + job_artifact_dir = artifact_base / resolved_job_id + jsonl_path = log_base / resolved_job_id / "invoke-ci.jsonl" + + click.echo("=" * 60) + click.echo(f"[smoke] Guest OS : {guest_os}") + click.echo(f"[smoke] Template : {resolved_template}") + click.echo(f"[smoke] Snapshot : {resolved_snapshot}") + click.echo(f"[smoke] Job id : {resolved_job_id}") + if preset is not None: + click.echo(f"[smoke] Preset : {preset}") + click.echo(f"[smoke] Artifacts : {job_artifact_dir}") + click.echo("=" * 60) + + _write_event( + jsonl_path, job_id=resolved_job_id, phase="job", status="start" + ) + + start = time.monotonic() + failed = False + try: + job_command.callback( # type: ignore[misc] + job_id=resolved_job_id, + repo_url=repo_url, + branch=branch, + commit="", + configuration="Release", + build_command=effective_build_command, + guest_artifact_source=artifact_source, + submodules=True, + use_git_clone=True, + use_shared_cache=False, + skip_artifact=False, + use_xvfb=False, + gitea_credential_target="GiteaPAT", + template_path=resolved_template, + snapshot_name=resolved_snapshot, + clone_base_dir=None, + artifact_base_dir=str(artifact_base), + guest_os=guest_os, + guest_credential_target=None, + ssh_key_path=ssh_key_path, + ssh_user="ci_build", + ssh_known_hosts_file=None, + extra_env_json="", + guest_cpu=0, + guest_memory_mb=0, + ready_timeout=600.0, + poll_interval=5.0, + vmrun_path=None, + ) + except click.ClickException as exc: + failed = True + _write_event( + jsonl_path, + job_id=resolved_job_id, + phase="job", + status="failure", + elapsed_sec=time.monotonic() - start, + error=exc.format_message(), + ) + click.echo(f"[smoke] FAIL: pipeline raised: {exc.format_message()}", err=True) + + if not failed: + _write_event( + jsonl_path, + job_id=resolved_job_id, + phase="job", + status="success", + elapsed_sec=time.monotonic() - start, + ) + + # ── Assert the three smoke invariants ─────────────────────────────── + click.echo("\n[smoke] Verifying results...") + checks_ok = True + + if failed: + checks_ok = False + else: + click.echo("[smoke] OK: pipeline exit 0") + + if job_artifact_dir.is_dir(): + click.echo(f"[smoke] OK: artifact dir exists: {job_artifact_dir}") + else: + click.echo( + f"[smoke] FAIL: artifact dir not found: {job_artifact_dir}", err=True + ) + checks_ok = False + + if _has_job_success(jsonl_path): + click.echo("[smoke] OK: JSONL contains job/success event") + else: + click.echo( + f"[smoke] FAIL: JSONL missing job/success event in: {jsonl_path}", + err=True, + ) + checks_ok = False + + click.echo("") + if checks_ok: + click.echo(f"[smoke] PASSED — {guest_os} smoke test completed successfully.") + return + raise click.ClickException("smoke test FAILED — one or more checks did not pass.") + + +__all__ = ["smoke"] diff --git a/src/ci_orchestrator/commands/validate.py b/src/ci_orchestrator/commands/validate.py new file mode 100644 index 0000000..193295d --- /dev/null +++ b/src/ci_orchestrator/commands/validate.py @@ -0,0 +1,285 @@ +"""``validate`` sub-commands. + +Phase C adds Linux-native host/guest validation, replacing host-side use of +``Validate-HostState.ps1`` and the diagnostic part of ``Test-CIGuestWinRM.ps1``: + +* ``validate host`` — vmrun present, template snapshots present, keyring + readable, ``/var/lib/ci`` permissions, systemd units active. +* ``validate guest`` — connect to a guest over WinRM/SSH and run a trivial + command, reporting the explicit failure cause. + +The host checks are a Linux-native equivalent of ``Validate-HostState.ps1`` +(not a 1:1 port): no Windows drive paths, no Windows Credential Manager, no +NTFS ACLs. The guest check is the diagnostic core of ``Test-CIGuestWinRM.ps1``: +unlike the ``is_ready`` probe (which swallows errors and returns a bool), +this reports the explicit failure cause. +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +from dataclasses import dataclass + +import click + +from ci_orchestrator.backends import load_backend +from ci_orchestrator.backends.errors import BackendError, BackendNotAvailable +from ci_orchestrator.backends.protocol import VmHandle +from ci_orchestrator.config import Config, load_config +from ci_orchestrator.credentials import KeyringCredentialStore +from ci_orchestrator.transport.errors import TransportCommandError, TransportConnectError +from ci_orchestrator.transport.ssh import SshTransport +from ci_orchestrator.transport.winrm import WinRmTransport + +# Each template directory and the snapshot a clone is taken from. Mirrors the +# "VM templates and snapshots" table in CLAUDE.md. +_TEMPLATE_SNAPSHOTS: tuple[tuple[str, str], ...] = ( + ("WinBuild2025", "BaseClean"), + ("WinBuild2022", "BaseClean"), + ("LinuxBuild2404", "BaseClean-Linux"), +) + +# systemd units expected to be active on a configured Linux host. +_EXPECTED_UNITS: tuple[str, ...] = ("act-runner.service",) + + +@dataclass(frozen=True) +class CheckResult: + """Outcome of a single host check.""" + + name: str + ok: bool + detail: str + + +@click.group("validate") +def validate() -> None: + """Host and guest readiness checks.""" + + +# ──────────────────────────────────────────────────────── validate host + +def _check_vmrun(config: Config) -> CheckResult: + """vmrun is present and executable (via the backend factory).""" + try: + backend = load_backend(config) + except BackendNotAvailable as exc: + return CheckResult("vmrun", False, str(exc)) + except BackendError as exc: # pragma: no cover - defensive + return CheckResult("vmrun", False, str(exc)) + vmrun_path = getattr(backend, "vmrun_path", None) + if vmrun_path: + return CheckResult("vmrun", True, f"found at {vmrun_path}") + return CheckResult("vmrun", True, "backend loaded") + + +def _check_snapshots(config: Config) -> CheckResult: + """Each known template VMX exists and exposes its expected snapshot.""" + try: + backend = load_backend(config) + except BackendError as exc: + return CheckResult("snapshots", False, f"backend unavailable: {exc}") + + templates_dir = config.paths.templates + problems: list[str] = [] + found: list[str] = [] + for template_name, snapshot in _TEMPLATE_SNAPSHOTS: + vmx = templates_dir / template_name / f"{template_name}.vmx" + if not vmx.is_file(): + # A host may run only a subset of templates; skip absent ones + # rather than failing on them. + continue + try: + snapshots = backend.list_snapshots(VmHandle(identifier=str(vmx))) + except BackendError as exc: + problems.append(f"{template_name}: listSnapshots failed ({exc})") + continue + if snapshot in snapshots: + found.append(f"{template_name}->{snapshot}") + else: + problems.append( + f"{template_name}: snapshot {snapshot!r} absent " + f"(have: {', '.join(snapshots) or 'none'})" + ) + + if problems: + return CheckResult("snapshots", False, "; ".join(problems)) + if not found: + return CheckResult( + "snapshots", False, f"no known template VMX found under {templates_dir}" + ) + return CheckResult("snapshots", True, ", ".join(found)) + + +def _check_keyring(config: Config) -> CheckResult: + """The file-based keyring vault is readable for the guest cred target.""" + target = config.guest_cred_target + try: + cred = KeyringCredentialStore().get(target) + except KeyError: + return CheckResult( + "keyring", False, f"credential {target!r} not found in keyring" + ) + except RuntimeError as exc: + return CheckResult("keyring", False, str(exc)) + return CheckResult("keyring", True, f"{target!r} readable (user={cred.username})") + + +def _check_permissions(config: Config) -> CheckResult: + """``$CI_ROOT`` and key sub-dirs exist and are readable+writable.""" + root = config.paths.root + to_check = [root, config.paths.build_vms, config.paths.artifacts] + problems: list[str] = [] + for path in to_check: + if not path.exists(): + problems.append(f"{path}: missing") + continue + if not os.access(path, os.R_OK | os.W_OK): + problems.append(f"{path}: not read/write for current user") + if problems: + return CheckResult("permissions", False, "; ".join(problems)) + return CheckResult("permissions", True, f"{root} read/write") + + +def _systemctl_is_active(unit: str) -> str | None: + """Return ``systemctl is-active`` state, or ``None`` if unavailable.""" + systemctl = shutil.which("systemctl") + if systemctl is None: + return None + try: + result = subprocess.run( + [systemctl, "is-active", unit], + capture_output=True, + text=True, + check=False, + timeout=10.0, + ) + except (subprocess.TimeoutExpired, OSError): + return None + return result.stdout.strip().lower() or None + + +def _check_units() -> CheckResult: + """Expected systemd units report ``active``.""" + systemctl = shutil.which("systemctl") + if systemctl is None: + return CheckResult( + "systemd", False, "systemctl not found (not a systemd host?)" + ) + inactive: list[str] = [] + active: list[str] = [] + for unit in _EXPECTED_UNITS: + state = _systemctl_is_active(unit) + if state == "active": + active.append(unit) + else: + inactive.append(f"{unit}={state or 'unknown'}") + if inactive: + return CheckResult("systemd", False, "; ".join(inactive)) + return CheckResult("systemd", True, ", ".join(active)) + + +def run_host_checks(config: Config) -> list[CheckResult]: + """Run every host check and return the per-check results.""" + return [ + _check_vmrun(config), + _check_snapshots(config), + _check_keyring(config), + _check_permissions(config), + _check_units(), + ] + + +@validate.command("host") +def validate_host() -> None: + """Check host prerequisites; exit non-zero on any failure.""" + config = load_config() + results = run_host_checks(config) + failed = [r for r in results if not r.ok] + for result in results: + marker = "OK " if result.ok else "FAIL" + click.echo(f"[{marker}] {result.name}: {result.detail}") + if failed: + click.echo(f"\n{len(failed)} check(s) failed.") + raise SystemExit(1) + click.echo("\nAll host checks passed.") + + +# ──────────────────────────────────────────────────────── validate guest + +def _resolve_use_winrm(use_winrm: bool | None, guest_os: str | None) -> bool: + """Decide the transport: explicit flag wins, else infer from guest OS.""" + if use_winrm is not None: + return use_winrm + if guest_os is not None: + return guest_os.lower() != "linux" + raise click.UsageError( + "specify the transport: pass --winrm/--ssh or --guest-os." + ) + + +def _probe_winrm(host: str, config: Config) -> CheckResult: + """Connect over WinRM and run a trivial command, reporting the cause.""" + try: + cred = KeyringCredentialStore().get(config.guest_cred_target) + except (KeyError, RuntimeError) as exc: + return CheckResult("winrm", False, f"credential lookup failed: {exc}") + transport = WinRmTransport(host, cred.username, cred.password) + try: + result = transport.run("$true | Out-Null", check=True) + except TransportConnectError as exc: + return CheckResult("winrm", False, f"connect/auth failed: {exc}") + except TransportCommandError as exc: + return CheckResult("winrm", False, f"command failed: {exc}") + finally: + transport.close() + return CheckResult("winrm", True, f"connected; exit={result.returncode}") + + +def _probe_ssh(host: str, config: Config) -> CheckResult: + """Connect over SSH and run a trivial command, reporting the cause.""" + key_path = str(config.ssh_key_path) if config.ssh_key_path else None + transport = SshTransport(host, key_path=key_path) + try: + result = transport.run("true", check=True, timeout=10.0) + except TransportConnectError as exc: + return CheckResult("ssh", False, f"connect/auth failed: {exc}") + except TransportCommandError as exc: + return CheckResult("ssh", False, f"command failed: {exc}") + finally: + transport.close() + return CheckResult("ssh", True, f"connected; exit={result.returncode}") + + +@validate.command("guest") +@click.option("--host", "host", required=True, help="Guest IP or hostname.") +@click.option( + "--winrm/--ssh", + "use_winrm", + default=None, + help="Force transport (default: inferred from --guest-os).", +) +@click.option( + "--guest-os", + type=click.Choice(["windows", "linux"], case_sensitive=False), + default=None, + help="Guest OS family (used to pick transport when --winrm/--ssh omitted).", +) +def validate_guest( + host: str, + use_winrm: bool | None, + guest_os: str | None, +) -> None: + """Probe a guest's transport and report the explicit failure cause.""" + config = load_config() + winrm = _resolve_use_winrm(use_winrm, guest_os) + result = _probe_winrm(host, config) if winrm else _probe_ssh(host, config) + marker = "OK " if result.ok else "FAIL" + click.echo(f"[{marker}] {result.name} {host}: {result.detail}") + if not result.ok: + raise SystemExit(1) + + +__all__ = ["validate"] diff --git a/tests/python/test_commands_bench.py b/tests/python/test_commands_bench.py new file mode 100644 index 0000000..b85c87a --- /dev/null +++ b/tests/python/test_commands_bench.py @@ -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 diff --git a/tests/python/test_commands_creds.py b/tests/python/test_commands_creds.py new file mode 100644 index 0000000..dda562f --- /dev/null +++ b/tests/python/test_commands_creds.py @@ -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" diff --git a/tests/python/test_commands_smoke.py b/tests/python/test_commands_smoke.py new file mode 100644 index 0000000..b16ca62 --- /dev/null +++ b/tests/python/test_commands_smoke.py @@ -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() diff --git a/tests/python/test_commands_validate.py b/tests/python/test_commands_validate.py new file mode 100644 index 0000000..7e8ea52 --- /dev/null +++ b/tests/python/test_commands_validate.py @@ -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