57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
from src.services.incident_timeline_service import (
|
|
STAGE_DEFS,
|
|
_reconciliation_event,
|
|
format_ascii_timeline,
|
|
)
|
|
|
|
|
|
def _stages(status_by_stage: dict[str, str]) -> list[dict]:
|
|
return [
|
|
{"stage": stage, "status": status_by_stage.get(stage, "skipped")}
|
|
for stage, _label in STAGE_DEFS
|
|
]
|
|
|
|
|
|
def test_format_ascii_timeline_skips_unrecorded_stages() -> None:
|
|
stages = _stages({
|
|
"webhook": "completed",
|
|
"ai_router": "success",
|
|
"executor": "error",
|
|
"km": "pending",
|
|
})
|
|
|
|
assert format_ascii_timeline(stages) == (
|
|
"webhook:ok > ai_router:ok > executor:fail > km:wait"
|
|
)
|
|
|
|
|
|
def test_format_ascii_timeline_has_empty_fallback() -> None:
|
|
assert format_ascii_timeline(_stages({})) == "webhook:skip > ai:skip > executor:skip"
|
|
|
|
|
|
def test_reconciliation_event_marks_safe_stage_failed() -> None:
|
|
event = _reconciliation_event({
|
|
"applicable": True,
|
|
"consistency_status": "blocked",
|
|
"operator_next_state": "manual_required",
|
|
"mismatches": [
|
|
{"code": "approval_approved_without_execution_record"},
|
|
{"code": "evidence_all_sensors_failed"},
|
|
],
|
|
})
|
|
|
|
assert event is not None
|
|
assert event["stage"] == "safe"
|
|
assert event["status"] == "error"
|
|
assert event["title"] == "Lifecycle reconciliation: blocked"
|
|
assert "approval_approved_without_execution_record" in event["description"]
|
|
assert event["data"]["operator_next_state"] == "manual_required"
|
|
|
|
|
|
def test_reconciliation_event_omits_consistent_state() -> None:
|
|
assert _reconciliation_event({
|
|
"applicable": True,
|
|
"consistency_status": "consistent",
|
|
"mismatches": [],
|
|
}) is None
|