104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
from types import SimpleNamespace
|
||
|
||
from src.services.platform_operator_service import (
|
||
_collect_run_incident_ids,
|
||
_outbound_timeline_title,
|
||
_remediation_timeline_summary,
|
||
)
|
||
|
||
|
||
def test_outbound_timeline_title_labels_runbook_review() -> None:
|
||
title = _outbound_timeline_title(
|
||
"telegram",
|
||
"approval_request",
|
||
"📄 <b>RUNBOOK REVIEW｜待審核</b>\nIncident：INC-1",
|
||
)
|
||
|
||
assert title == "TELEGRAM：Runbook 待人工審核"
|
||
|
||
|
||
def test_outbound_timeline_title_labels_governance_alert() -> None:
|
||
title = _outbound_timeline_title(
|
||
"telegram",
|
||
"final",
|
||
"⚠️ *AI 治理警報｜知識庫劣化*",
|
||
)
|
||
|
||
assert title == "TELEGRAM：AI 治理警報"
|
||
|
||
|
||
def test_outbound_timeline_title_labels_cicd_status() -> None:
|
||
title = _outbound_timeline_title(
|
||
"telegram",
|
||
"final",
|
||
"✅ <b>[AWOOOI CI/CD]</b> | code-review\n📦 Code Review 完成・LOW",
|
||
)
|
||
|
||
assert title == "TELEGRAM：CI/CD 狀態通知"
|
||
|
||
|
||
def test_outbound_timeline_title_labels_auto_repair_handoff() -> None:
|
||
title = _outbound_timeline_title(
|
||
"telegram",
|
||
"error",
|
||
"🤖❌ HANDOFF REQUIRED｜AI 自動修復失敗，已轉人工",
|
||
)
|
||
|
||
assert title == "TELEGRAM：AI 自動修復失敗，已轉人工"
|
||
|
||
|
||
def test_outbound_timeline_title_falls_back_to_human_label() -> None:
|
||
title = _outbound_timeline_title("telegram", "interim", "正在調用 MCP 工具")
|
||
|
||
assert title == "TELEGRAM：漸進式狀態回饋"
|
||
|
||
|
||
def test_collect_run_incident_ids_reads_source_refs_and_legacy_text() -> None:
|
||
run = SimpleNamespace(
|
||
trigger_ref="not-an-incident",
|
||
error_detail=None,
|
||
)
|
||
inbound_events = [
|
||
SimpleNamespace(
|
||
source_envelope={
|
||
"source_refs": {
|
||
"incident_ids": ["INC-20260514-F85F21", "INC-20260514-F85F21"],
|
||
}
|
||
},
|
||
content_preview="Alertmanager inbound converged",
|
||
content_redacted=None,
|
||
)
|
||
]
|
||
outbound_messages = [
|
||
SimpleNamespace(
|
||
content_preview="詳情：INC-20260513-79ED5E",
|
||
send_error=None,
|
||
)
|
||
]
|
||
|
||
incident_ids = _collect_run_incident_ids(
|
||
run=run,
|
||
inbound_events=inbound_events,
|
||
outbound_messages=outbound_messages,
|
||
)
|
||
|
||
assert incident_ids == ["INC-20260514-F85F21", "INC-20260513-79ED5E"]
|
||
|
||
|
||
def test_remediation_timeline_summary_surfaces_route_and_write_flags() -> None:
|
||
summary = _remediation_timeline_summary({
|
||
"incident_id": "INC-20260514-F85F21",
|
||
"mode": "replay",
|
||
"verification_result_preview": "degraded",
|
||
"agent_id": "auto_repair_executor",
|
||
"tool_name": "ssh_diagnose",
|
||
"required_scope": "read",
|
||
"writes_incident_state": False,
|
||
"writes_auto_repair_result": False,
|
||
})
|
||
|
||
assert "incident=INC-20260514-F85F21" in summary
|
||
assert "route=auto_repair_executor/ssh_diagnose/read" in summary
|
||
assert "writes_incident=False" in summary
|
||
assert "writes_auto_repair=False" in summary
|