89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
from src.services.incident_timeline_service import (
|
|
STAGE_DEFS,
|
|
_approval_execution_truth,
|
|
_approval_status_to_timeline_status,
|
|
_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
|
|
|
|
|
|
def test_approval_execution_success_without_repair_is_timeline_info() -> None:
|
|
approval = type("Approval", (), {
|
|
"action": "OBSERVE",
|
|
"extra_metadata": {"execution_kind": "no_action", "repair_executed": False},
|
|
})()
|
|
|
|
truth = _approval_execution_truth(approval)
|
|
|
|
assert truth == {"execution_kind": "no_action", "repair_executed": False}
|
|
assert _approval_status_to_timeline_status(
|
|
"execution_success",
|
|
repair_executed=truth["repair_executed"],
|
|
) == "info"
|
|
|
|
|
|
def test_approval_execution_success_with_repair_stays_success() -> None:
|
|
approval = type("Approval", (), {
|
|
"action": "kubectl rollout restart deployment/awoooi-api",
|
|
"extra_metadata": {"execution_kind": "kubectl", "repair_executed": True},
|
|
})()
|
|
|
|
truth = _approval_execution_truth(approval)
|
|
|
|
assert truth == {"execution_kind": "kubectl", "repair_executed": True}
|
|
assert _approval_status_to_timeline_status(
|
|
"execution_success",
|
|
repair_executed=truth["repair_executed"],
|
|
) == "success"
|