Simone
d63fca5967
Add four native command groups so the Linux host no longer needs pwsh for
manual ops. __main__ registers all four at once, so they ship together.
- bench run (C2): N concurrent jobs in-process via concurrent.futures,
per-round orphan-clone + stale-lock assertions. Replaces
Test-CapacityBurnIn.ps1 + Start-BurnInTest*.ps1. No
Start-Job/pwsh/$IsWindows.
- bench measure (C3): clone/start/IP/transport/destroy timings appended to
benchmark.jsonl (legacy field names kept). Replaces
Measure-CIBenchmark.ps1.
- smoke run (C4): one E2E job; asserts exit 0 + artifact dir + job/success
event. ns7zip/nsinnounp presets. Replaces Test-Smoke.ps1
+ Test-*Build-Linux.ps1.
- validate host (C5): vmrun/snapshot/keyring/permissions/systemd checks.
- validate guest(C6): transport probe with explicit failure cause.
- creds set (C6): keyring writer matching credentials.py read scheme;
password via stdin/prompt only. Replaces
Set-CIGuestCredential.ps1.
All groups import backends.load_backend only (Phase D ESXi path stays open).
Suite green, coverage 95.10% (gate 90%); new modules at 100%.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
287 lines
9.8 KiB
Python
287 lines
9.8 KiB
Python
"""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()
|