feat(a2): port leaf PS scripts to ci_orchestrator CLI
Implements Phase A2 of plans/implementation-plan-A-B.md: - commands/wait.py -> wait-ready (replaces Wait-VMReady.ps1) - commands/vm.py -> vm remove + vm cleanup (replaces Remove-BuildVM.ps1, Cleanup-OrphanedBuildVMs.ps1); cleanup accepts injected VmBackend (Phase C ESXi hook preserved) - commands/monitor.py -> monitor disk + monitor runner (replaces Watch-DiskSpace.ps1, Watch-RunnerHealth.ps1) - commands/report.py -> report job (replaces Get-CIJobSummary.ps1) Each PS script reduced to a 3-line shim that delegates to the Python CLI and preserves \0. Tests: 69 pytest cases across new test_commands_*.py modules; ruff clean, mypy --strict clean, coverage 74.5% (>=70 gate).
This commit is contained in:
@@ -0,0 +1,108 @@
|
||||
"""Tests for ``ci_orchestrator report job``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from ci_orchestrator.__main__ import cli
|
||||
|
||||
|
||||
def _write_jsonl(path: Path, events: list[dict[str, object]]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("\n".join(json.dumps(e) for e in events) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def _success_events(job_id: str = "run-1") -> list[dict[str, object]]:
|
||||
return [
|
||||
{"jobId": job_id, "phase": "job", "status": "start", "ts": "2026-05-14T10:00:00Z"},
|
||||
{"jobId": job_id, "phase": "phase4.wait-ready", "status": "success", "ts": "2026-05-14T10:01:00Z"},
|
||||
{
|
||||
"jobId": job_id,
|
||||
"phase": "job",
|
||||
"status": "success",
|
||||
"ts": "2026-05-14T10:05:00Z",
|
||||
"data": {"elapsedSec": 305},
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def _failure_events(job_id: str = "run-2") -> list[dict[str, object]]:
|
||||
return [
|
||||
{"jobId": job_id, "phase": "job", "status": "start", "ts": "2026-05-14T11:00:00Z"},
|
||||
{
|
||||
"jobId": job_id,
|
||||
"phase": "job",
|
||||
"status": "failure",
|
||||
"ts": "2026-05-14T11:00:30Z",
|
||||
"data": {"elapsedSec": 30, "error": "Clone failed: source not found"},
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def test_report_job_summary_table(tmp_path: Path) -> None:
|
||||
_write_jsonl(tmp_path / "run-1" / "invoke-ci.jsonl", _success_events("run-1"))
|
||||
_write_jsonl(tmp_path / "run-2" / "invoke-ci.jsonl", _failure_events("run-2"))
|
||||
result = CliRunner().invoke(cli, ["report", "job", "--log-dir", str(tmp_path)])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "run-1" in result.output
|
||||
assert "run-2" in result.output
|
||||
assert "success" in result.output
|
||||
assert "FAILED" in result.output
|
||||
|
||||
|
||||
def test_report_job_failed_filter(tmp_path: Path) -> None:
|
||||
_write_jsonl(tmp_path / "run-1" / "invoke-ci.jsonl", _success_events("run-1"))
|
||||
_write_jsonl(tmp_path / "run-2" / "invoke-ci.jsonl", _failure_events("run-2"))
|
||||
result = CliRunner().invoke(
|
||||
cli, ["report", "job", "--log-dir", str(tmp_path), "--failed"]
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "run-1" not in result.output
|
||||
assert "run-2" in result.output
|
||||
|
||||
|
||||
def test_report_job_detail_view(tmp_path: Path) -> None:
|
||||
_write_jsonl(tmp_path / "run-2" / "invoke-ci.jsonl", _failure_events("run-2"))
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
["report", "job", "--log-dir", str(tmp_path), "--job-id", "run-2"],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Job: run-2" in result.output
|
||||
assert "Clone failed" in result.output
|
||||
|
||||
|
||||
def test_report_job_missing_log_dir(tmp_path: Path) -> None:
|
||||
missing = tmp_path / "nope"
|
||||
result = CliRunner().invoke(cli, ["report", "job", "--log-dir", str(missing)])
|
||||
assert result.exit_code == 1
|
||||
assert "not found" in result.output
|
||||
|
||||
|
||||
def test_report_job_no_logs(tmp_path: Path) -> None:
|
||||
result = CliRunner().invoke(cli, ["report", "job", "--log-dir", str(tmp_path)])
|
||||
assert result.exit_code == 0
|
||||
assert "No invoke-ci.jsonl" in result.output
|
||||
|
||||
|
||||
def test_report_job_json_output(tmp_path: Path) -> None:
|
||||
_write_jsonl(tmp_path / "run-1" / "invoke-ci.jsonl", _success_events("run-1"))
|
||||
result = CliRunner().invoke(
|
||||
cli, ["report", "job", "--log-dir", str(tmp_path), "--json"]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
payload = json.loads(result.output.strip())
|
||||
assert isinstance(payload, list)
|
||||
assert payload[0]["jobId"] == "run-1"
|
||||
assert payload[0]["status"] == "success"
|
||||
|
||||
|
||||
def test_report_job_unknown_id(tmp_path: Path) -> None:
|
||||
result = CliRunner().invoke(
|
||||
cli, ["report", "job", "--log-dir", str(tmp_path), "--job-id", "ghost"]
|
||||
)
|
||||
assert result.exit_code == 1
|
||||
assert "No JSONL log" in result.output
|
||||
Reference in New Issue
Block a user