Simone
8dc7a4fb99
- Implement tests for disk alert JSON output to include a message field when below threshold. - Add tests for runner status when Gitea is online but no webhook URL is provided. - Introduce tests for job report details to ensure no error line appears when there are no failure events. - Add tests to skip empty JSONL files in job report listing.
247 lines
8.8 KiB
Python
247 lines
8.8 KiB
Python
"""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
|
|
|
|
|
|
# ── helpers ────────────────────────────────────────────────────────────────
|
|
|
|
import ci_orchestrator.commands.report as report_module # noqa: E402
|
|
|
|
|
|
def test_read_events_oserror_returns_empty(tmp_path: Path) -> None:
|
|
p = tmp_path / "d" # directory: read_text raises OSError
|
|
p.mkdir()
|
|
assert report_module._read_events(p) == []
|
|
|
|
|
|
def test_read_events_skips_blank_invalid_and_non_dict(tmp_path: Path) -> None:
|
|
p = tmp_path / "x.jsonl"
|
|
p.write_text(
|
|
"\n" # blank
|
|
"{not json\n" # invalid JSON
|
|
"[1, 2]\n" # not a dict
|
|
'{"phase": "job", "status": "start", "ts": "t"}\n'
|
|
'{"phase": "job", "status": "success", "data": {"elapsedSec": 1.5, "error": "e"}}\n',
|
|
encoding="utf-8",
|
|
)
|
|
events = report_module._read_events(p)
|
|
assert len(events) == 2
|
|
assert events[1].elapsed_sec == 1
|
|
assert events[1].error == "e"
|
|
|
|
|
|
def test_format_hms_negative_returns_question_mark() -> None:
|
|
assert report_module._format_hms(-1) == "?"
|
|
|
|
|
|
def test_format_hms_with_hours() -> None:
|
|
assert report_module._format_hms(3725) == "1h02m05s"
|
|
assert report_module._format_hms(45) == "0m45s"
|
|
|
|
|
|
def test_summarise_in_progress_when_no_terminal_event() -> None:
|
|
events = [
|
|
report_module._Event(
|
|
phase="job", status="start", ts="t", elapsed_sec=None, error=None,
|
|
raw={"jobId": "jx"},
|
|
)
|
|
]
|
|
row = report_module._summarise(events, "default")
|
|
assert row.status == "in-progress"
|
|
assert row.job_id == "jx"
|
|
|
|
|
|
def test_summarise_uses_default_id_when_jobid_missing() -> None:
|
|
events = [
|
|
report_module._Event(
|
|
phase="job", status="start", ts="t", elapsed_sec=None, error=None,
|
|
raw={},
|
|
)
|
|
]
|
|
row = report_module._summarise(events, "fallback")
|
|
assert row.job_id == "fallback"
|
|
|
|
|
|
def test_summarise_truncates_long_error() -> None:
|
|
long_err = "x" * 100
|
|
events = [
|
|
report_module._Event(
|
|
phase="job", status="failure", ts="t", elapsed_sec=10, error=long_err,
|
|
raw={"jobId": "j"},
|
|
)
|
|
]
|
|
row = report_module._summarise(events, "d")
|
|
assert row.status == "FAILED"
|
|
assert row.error.endswith("...")
|
|
assert len(row.error) == 60
|
|
|
|
|
|
def test_report_job_detail_json_emits_raw_events(tmp_path: Path) -> None:
|
|
_write_jsonl(tmp_path / "r1" / "invoke-ci.jsonl", _success_events("r1"))
|
|
result = CliRunner().invoke(
|
|
cli, ["report", "job", "--log-dir", str(tmp_path), "--job-id", "r1", "--json"]
|
|
)
|
|
assert result.exit_code == 0
|
|
payload = json.loads(result.output.strip())
|
|
assert isinstance(payload, list)
|
|
assert payload[0]["phase"] == "job"
|
|
|
|
|
|
def test_report_job_detail_empty_jsonl(tmp_path: Path) -> None:
|
|
p = tmp_path / "empty" / "invoke-ci.jsonl"
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text("")
|
|
result = CliRunner().invoke(
|
|
cli, ["report", "job", "--log-dir", str(tmp_path), "--job-id", "empty"]
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "empty or unreadable" in result.output
|
|
|
|
|
|
def test_report_job_failed_filter_no_results(tmp_path: Path) -> None:
|
|
_write_jsonl(tmp_path / "r1" / "invoke-ci.jsonl", _success_events("r1"))
|
|
result = CliRunner().invoke(
|
|
cli, ["report", "job", "--log-dir", str(tmp_path), "--failed"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "No matching jobs" in result.output
|
|
|
|
|
|
# ── NEW: coverage gap tests ───────────────────────────────────────────────────
|
|
|
|
|
|
def test_report_job_detail_no_failure_event(tmp_path: Path) -> None:
|
|
"""Lines 201->203: detail view where events exist but no failure → no 'Error:' line."""
|
|
_write_jsonl(tmp_path / "run-ok" / "invoke-ci.jsonl", _success_events("run-ok"))
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
["report", "job", "--log-dir", str(tmp_path), "--job-id", "run-ok"],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "Job: run-ok" in result.output
|
|
# No failure event → no "Error:" line should appear
|
|
assert "Error:" not in result.output
|
|
|
|
|
|
def test_report_job_list_skips_empty_jsonl(tmp_path: Path) -> None:
|
|
"""Line 218: list mode with an empty/unreadable JSONL file → that file is skipped."""
|
|
# Write one valid job
|
|
_write_jsonl(tmp_path / "run-good" / "invoke-ci.jsonl", _success_events("run-good"))
|
|
# Write an empty JSONL (no events) — should be silently skipped
|
|
empty_jsonl = tmp_path / "run-empty" / "invoke-ci.jsonl"
|
|
empty_jsonl.parent.mkdir(parents=True)
|
|
empty_jsonl.write_text("", encoding="utf-8")
|
|
|
|
result = CliRunner().invoke(cli, ["report", "job", "--log-dir", str(tmp_path)])
|
|
assert result.exit_code == 0, result.output
|
|
# Only the valid job appears
|
|
assert "run-good" in result.output
|
|
# The empty dir's name must not appear as a row (it was skipped)
|
|
# We can confirm by checking only one data row exists: "success" appears once
|
|
assert result.output.count("success") >= 1
|