From 5dd28b2fc6a754ac714d60731234c0d7faa6efa7 Mon Sep 17 00:00:00 2001 From: OG T Date: Wed, 1 Apr 2026 21:11:45 +0800 Subject: [PATCH] test(telegram): add ADR-050 info action tests (detail/reanalyze/history/keyboard) Co-Authored-By: Claude Sonnet 4.6 --- apps/api/tests/test_telegram_adr050.py | 217 +++++++++++++++++++++++++ 1 file changed, 217 insertions(+) create mode 100644 apps/api/tests/test_telegram_adr050.py diff --git a/apps/api/tests/test_telegram_adr050.py b/apps/api/tests/test_telegram_adr050.py new file mode 100644 index 00000000..95222d7b --- /dev/null +++ b/apps/api/tests/test_telegram_adr050.py @@ -0,0 +1,217 @@ +""" +ADR-050 Telegram 互動式 Incident 管理 2.0 測試 +============================================== +驗證 6鍵 Inline Keyboard + detail/reanalyze/history 訊息格式 + +測試策略 (遵循 feedback_no_mock_testing.md): +- 直接測試訊息格式化邏輯與 keyboard 結構 +- 透過 source code 驗證實作正確性（不依賴 Runtime） +- 不發送實際 Telegram 訊息，不呼叫外部 API + +版本: v1.0 +建立: 2026-04-01 (台北時區) +建立者: Claude Code (ADR-050 P2 Tests) +""" + + +# ============================================================================= +# Test: _build_inline_keyboard 6鍵結構 +# ============================================================================= + +class TestInlineKeyboardStructure: + """驗證 ADR-050 6鍵 Inline Keyboard 結構""" + + def _read_gateway(self) -> str: + with open("src/services/telegram_gateway.py", encoding="utf-8") as f: + return f.read() + + def test_keyboard_row1_approve_button_exists(self): + """第一行有 ✅ 批准 按鈕""" + assert "✅ 批准" in self._read_gateway() + + def test_keyboard_row1_reject_button_exists(self): + """第一行有 ❌ 拒絕 按鈕""" + assert "❌ 拒絕" in self._read_gateway() + + def test_keyboard_row1_silence_button_exists(self): + """第一行有 🔕 靜默 按鈕""" + assert "🔕 靜默" in self._read_gateway() + + def test_keyboard_row2_detail_uses_incident_id_format(self): + """📋 詳情 按鈕使用 detail:{incident_id} 格式""" + source = self._read_gateway() + assert '📋 詳情' in source + assert 'f"detail:{incident_id}"' in source + + def test_keyboard_row2_reanalyze_uses_incident_id_format(self): + """🔄 重診 按鈕使用 reanalyze:{incident_id} 格式""" + source = self._read_gateway() + assert '🔄 重診' in source + assert 'f"reanalyze:{incident_id}"' in source + + def test_keyboard_row2_history_uses_incident_id_format(self): + """📊 歷史 按鈕使用 history:{incident_id} 格式""" + source = self._read_gateway() + assert '📊 歷史' in source + assert 'f"history:{incident_id}"' in source + + def test_reanalyze_no_longer_placeholder(self): + """reanalyze 不再是 placeholder（功能開發中已移除）""" + assert "功能開發中" not in self._read_gateway(), ( + "reanalyze 仍是 placeholder！應已實作真實邏輯。" + ) + + +# ============================================================================= +# Test: detail 訊息格式 +# ============================================================================= + +class TestDetailMessageFormat: + """驗證 _send_incident_detail 訊息格式與資料來源""" + + def _read_gateway(self) -> str: + with open("src/services/telegram_gateway.py", encoding="utf-8") as f: + return f.read() + + def test_detail_method_exists(self): + """_send_incident_detail 方法存在""" + assert "_send_incident_detail" in self._read_gateway() + + def test_detail_uses_real_db(self): + """detail 從真實 DB 取得資料（IncidentRepository）""" + source = self._read_gateway() + assert "get_incident_repository" in source + assert "repo.get_by_id(incident_id)" in source + + def test_detail_handles_not_found(self): + """detail 處理找不到事件的情況""" + assert "找不到事件" in self._read_gateway() + + def test_detail_shows_incident_id(self): + """detail 訊息顯示 INC-ID""" + assert "incident.incident_id" in self._read_gateway() + + def test_detail_shows_severity(self): + """detail 訊息顯示嚴重度""" + assert "incident.severity" in self._read_gateway() + + +# ============================================================================= +# Test: history 訊息格式 +# ============================================================================= + +class TestHistoryMessageFormat: + """驗證 _send_incident_history 訊息格式與資料來源""" + + def _read_gateway(self) -> str: + with open("src/services/telegram_gateway.py", encoding="utf-8") as f: + return f.read() + + def test_history_method_exists(self): + """_send_incident_history 方法存在""" + assert "_send_incident_history" in self._read_gateway() + + def test_history_uses_real_frequency_stats(self): + """history 使用真實頻率統計資料""" + source = self._read_gateway() + assert "count_1h" in source + assert "count_24h" in source + assert "count_7d" in source + assert "count_30d" in source + + def test_history_handles_no_stats(self): + """history 處理無頻率統計資料的情況""" + assert "無頻率統計資料" in self._read_gateway() + + +# ============================================================================= +# Test: reanalyze 訊息格式 +# ============================================================================= + +class TestReanalyzeMessageFormat: + """驗證 _send_reanalyze_result 訊息格式""" + + def _read_gateway(self) -> str: + with open("src/services/telegram_gateway.py", encoding="utf-8") as f: + return f.read() + + def test_reanalyze_result_method_exists(self): + """_send_reanalyze_result 方法存在""" + assert "_send_reanalyze_result" in self._read_gateway() + + def test_reanalyze_calls_trigger_reanalysis(self): + """reanalyze handler 呼叫 trigger_reanalysis""" + assert "trigger_reanalysis" in self._read_gateway() + + def test_reanalyze_uses_lazy_import(self): + """reanalyze 使用 lazy import 避免循環依賴""" + assert "from src.services.incident_service import get_incident_service" in self._read_gateway() + + def test_reanalyze_scheduled_message(self): + """reanalyze 有「重診已排程」訊息""" + assert "重診已排程" in self._read_gateway() + + def test_reanalyze_in_progress_message(self): + """reanalyze 有「重診進行中」訊息""" + assert "重診進行中" in self._read_gateway() + + def test_reanalyze_failure_message(self): + """reanalyze 有「重診失敗」訊息""" + assert "重診失敗" in self._read_gateway() + + def test_reanalyze_uses_html_escape(self): + """reanalyze 訊息使用 html.escape 防 XSS""" + source = self._read_gateway() + # _send_reanalyze_result 之後應有 html.escape 呼叫 + reanalyze_section = source.split("_send_reanalyze_result")[1] if "_send_reanalyze_result" in source else "" + assert "html.escape" in reanalyze_section + + +# ============================================================================= +# Test: trigger_reanalysis 邏輯結構 +# ============================================================================= + +class TestTriggerReanalysisStructure: + """驗證 trigger_reanalysis() 方法結構（不連接真實 Redis）""" + + def _read_service(self) -> str: + with open("src/services/incident_service.py", encoding="utf-8") as f: + return f.read() + + def test_method_exists(self): + """trigger_reanalysis 方法存在於 IncidentService""" + assert "async def trigger_reanalysis" in self._read_service() + + def test_uses_redis_setnx(self): + """使用 Redis SETNX (nx=True) 去重""" + assert "nx=True" in self._read_service() + + def test_dedup_key_format(self): + """去重 key 格式為 reanalyze_dedup:{incident_id}""" + assert '"reanalyze_dedup:"' in self._read_service() or "reanalyze_dedup:" in self._read_service() + + def test_ttl_is_600_seconds(self): + """去重 TTL 為 600 秒（10 分鐘）""" + assert "600" in self._read_service() + + def test_returns_triggered_key(self): + """回傳值包含 triggered key""" + assert '"triggered"' in self._read_service() + + def test_returns_message_key(self): + """回傳值包含 message key""" + assert '"message"' in self._read_service() + + def test_returns_already_analyzing_key(self): + """回傳值包含 already_analyzing key""" + assert '"already_analyzing"' in self._read_service() + + def test_deletes_key_on_not_found(self): + """找不到 incident 時刪除去重 key（允許重試）""" + assert "redis_client.delete(dedup_key)" in self._read_service() + + def test_fallback_to_episodic_memory(self): + """Working Memory 找不到時 fallback 到 Episodic Memory (DB)""" + source = self._read_service() + assert "get_from_working_memory" in source + assert "get_from_episodic_memory" in source