fix(telegram): thread incident follow-up messages
All checks were successful
Code Review / ai-code-review (push) Successful in 10s
CD Pipeline / tests (push) Successful in 1m4s
CD Pipeline / build-and-deploy (push) Successful in 3m30s
CD Pipeline / post-deploy-checks (push) Successful in 1m19s

This commit is contained in:
Your Name
2026-05-07 01:11:02 +08:00
parent 1a72f771de
commit 1f4a16e625
2 changed files with 143 additions and 2 deletions

View File

@@ -251,10 +251,79 @@ def test_outbound_message_type_inference():
assert (
telegram_gateway_module._infer_outbound_message_type(
"✅ 執行成功",
{"reply_to_message_id": 123},
{"reply_parameters": {"message_id": 123}},
)
== "final"
)
assert (
telegram_gateway_module._infer_outbound_message_type(
"📄 <b>RUNBOOK REVIEW待審核</b>",
{"reply_parameters": {"message_id": 123}},
)
== "approval_request"
)
def test_extract_incident_id_from_text():
"""Telegram 出站文字可擷取 Incident ID供後續訊息接回原告警卡片。"""
assert (
telegram_gateway_module._extract_incident_id_from_text(
"Incident: INC-20260506-E54736\nEntry ID: abc"
)
== "INC-20260506-E54736"
)
assert telegram_gateway_module._extract_incident_id_from_text("沒有事件編號") is None
@pytest.mark.asyncio
async def test_attach_incident_thread_reply(monkeypatch):
"""非主卡、含 Incident ID 的後續訊息,應自動 reply 原告警 message_id。"""
class FakeRedis:
async def get(self, key):
assert key == "tg_msg:INC-20260506-E54736"
return "9876"
gateway = TelegramGateway()
payload = {
"chat_id": gateway.alert_chat_id,
"text": "📄 RUNBOOK REVIEW待審核\nIncident: INC-20260506-E54736",
}
monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: FakeRedis())
await gateway._attach_incident_thread_reply("sendMessage", payload)
assert payload["reply_parameters"] == {
"message_id": 9876,
"allow_sending_without_reply": True,
}
@pytest.mark.asyncio
async def test_attach_incident_thread_skips_root_action_card(monkeypatch):
"""ACTION REQUIRED 主告警卡不應自動 reply 舊訊息。"""
class FakeRedis:
async def get(self, key): # pragma: no cover - 不應被呼叫
raise AssertionError(key)
gateway = TelegramGateway()
payload = {
"chat_id": gateway.alert_chat_id,
"text": (
" ACTION REQUIRED | 低風險\n"
"📋 INC-20260506-E54736\n"
"🤖 AI 自動化鏈路"
),
}
monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: FakeRedis())
await gateway._attach_incident_thread_reply("sendMessage", payload)
assert "reply_parameters" not in payload
def test_legacy_outbound_run_id_is_stable():