test(telegram): add ADR-050 info action tests (detail/reanalyze/history/keyboard)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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