From d3a4fb4d15f9487a3b281d94ab97bb5cfafbc109 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 26 Apr 2026 20:17:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(t0):=20Task=20A=20=E6=8C=89=E9=88=95?= =?UTF-8?q?=E4=B8=80=E8=87=B4=E6=80=A7=E6=B8=AC=E8=A9=A6=20+=20Task=20C=20?= =?UTF-8?q?Gitea=E2=86=92Telegram=20=E9=80=9A=E7=9F=A5=E6=94=B6=E5=B0=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task A — Telegram 按鈕鬼魂鐵律測試（補測 production telegram_gateway.py） - test_telegram_button_consistency.py 新增 14 測試 - send_info_notification 兩鍵 [📋 詳情][📊 歷史] - _send_approval_card_to_group reply_markup - callback_data 對齊 INFO_ACTIONS 白名單 - parse_callback_data + handler 完整性 Task C — Gitea CI/CD → Telegram 告警轉發 - GiteaPullRequest.merged 欄位（HasMerged bool json:"merged"） - _send_gitea_notification helper：Redis SET NX EX 600s 去重 - handle_pull_request: closed+merged → PR Merged Telegram 卡片 - handle_workflow_run: status=failure → 部署/構建失敗卡片 - 不加按鈕（feedback_no_ghost_buttons.md 合規） - test_gitea_webhook.py +247 行新測試 驗收: K8s GITEA_WEBHOOK_SECRET 64 bytes ✅ Gitea hook #4 events: pull_request + push + workflow_run ✅ 端點 HMAC 401 驗簽 ✅ Co-Authored-By: Claude Opus 4.7 (1M context) --- apps/api/src/api/v1/gitea_webhook.py | 125 ++++++++- apps/api/tests/test_gitea_webhook.py | 247 ++++++++++++++++++ .../tests/test_telegram_button_consistency.py | 197 ++++++++++++++ 3 files changed, 568 insertions(+), 1 deletion(-) create mode 100644 apps/api/tests/test_telegram_button_consistency.py diff --git a/apps/api/src/api/v1/gitea_webhook.py b/apps/api/src/api/v1/gitea_webhook.py index 9356e840..ff90e4be 100644 --- a/apps/api/src/api/v1/gitea_webhook.py +++ b/apps/api/src/api/v1/gitea_webhook.py @@ -52,6 +52,11 @@ router = APIRouter(prefix="/webhooks/gitea", tags=["Gitea Webhook"]) # OpenClaw 配置 (使用 settings 中的 OPENCLAW_URL) OPENCLAW_URL = settings.OPENCLAW_URL +# Telegram 通知去重 TTL — 10 分鐘，與 Sentry/SLO Watchdog 對齊 +# 2026-04-25 ogt + Claude Sonnet 4.6 (Task C: Gitea CI/CD 告警轉發 Telegram) +GITEA_TG_DEDUP_TTL = 600 # 秒 +GITEA_TG_DEDUP_KEY_PREFIX = "gitea:tg:dedup:" + # ============================================================================= # Pydantic Models # ============================================================================= @@ -87,6 +92,9 @@ class GiteaPullRequest(BaseModel): additions: int = 0 deletions: int = 0 changed_files: int = 0 + # Gitea: HasMerged bool json:"merged" — True 代表 PR 已合併 (action=closed + merged=true) + # 2026-04-25 ogt + Claude Sonnet 4.6 (Task C: Gitea CI/CD 告警轉發 Telegram) + merged: bool = False class GiteaCommit(BaseModel): @@ -364,6 +372,61 @@ async def handle_gitea_webhook( ) from e +# ============================================================================= +# Telegram 通知 Helper (帶 Redis 去重) +# 2026-04-25 ogt + Claude Sonnet 4.6 (Task C: Gitea CI/CD 告警轉發 Telegram) +# 設計原則: +# - 純通知，不加按鈕（遵循 feedback_no_ghost_buttons.md） +# - Redis SET NX EX 600s 去重（同一 repo+event+id 10 分鐘內不重複） +# - 不改動 incident 通知鏈路，獨立背景任務 +# - Telegram token/chat_id 從 settings (K8s Secret 注入) 讀取，不寫死 +# ============================================================================= + +async def _send_gitea_notification( + dedup_key: str, + message: str, +) -> None: + """ + 發送 Gitea 事件 Telegram 通知（帶去重） + + Args: + dedup_key: Redis 去重 key（格式: {event}:{repo}:{id}，不含 prefix） + message: HTML 格式 Telegram 訊息 + """ + try: + # 去重檢查：同一 key 在 TTL 內不重複發送 + from src.core.redis_client import get_redis # type: ignore[import] + redis = await get_redis() + full_key = GITEA_TG_DEDUP_KEY_PREFIX + dedup_key + acquired = await redis.set( + full_key, + "1", + ex=GITEA_TG_DEDUP_TTL, + nx=True, # NX: 只在 key 不存在時設定（原子操作） + ) + if not acquired: + logger.debug( + "gitea_tg_dedup_skip", + dedup_key=dedup_key, + ttl=GITEA_TG_DEDUP_TTL, + ) + return + + if not settings.OPENCLAW_TG_BOT_TOKEN: + logger.debug("gitea_tg_skipped", reason="Bot token not configured") + return + + from src.services.telegram_gateway import get_telegram_gateway # type: ignore[import] + gateway = get_telegram_gateway() + await gateway.initialize() + await gateway.send_notification(message) + + logger.info("gitea_tg_notification_sent", dedup_key=dedup_key) + + except Exception as e: + logger.warning("gitea_tg_notification_failed", dedup_key=dedup_key, error=str(e)) + + # ============================================================================= # Event Handlers (HTTP 層: 解析、驗證、回應 — 業務邏輯在 Service 層) # ============================================================================= @@ -380,6 +443,7 @@ async def handle_pull_request( - opened: 新建 PR - synchronize: 推送新 commit 到 PR - reopened: 重新開啟 PR + - closed + merged=True: PR 合併完成 → Telegram 通知 (Task C 2026-04-25) """ pr = payload.pull_request if not pr: @@ -389,6 +453,40 @@ async def handle_pull_request( event_type="pull_request", ) + # PR 合併完成通知 (action=closed + merged=True) + # 2026-04-25 ogt + Claude Sonnet 4.6 (Task C: Gitea CI/CD 告警轉發 Telegram) + if payload.action == "closed" and pr.merged: + repo = payload.repository.full_name + author = payload.sender.login + pr_url = pr.html_url + base_branch = pr.base.get("ref", "main") if isinstance(pr.base, dict) else "main" + + # 格式遵循 feedback_telegram_alert_format.md + message = ( + f"PR Merged | {repo}\n" + "──────────────────────\n" + f"├─ PR: #{pr.number} {pr.title[:60]}\n" + f"├─ 作者: @{author}\n" + f"├─ 目標分支: {base_branch}\n" + f"└─ 變更: +{pr.additions} -{pr.deletions} ({pr.changed_files} 檔)" + ) + + dedup_key = f"pr_merged:{repo}:{pr.number}" + background_tasks.add_task(_send_gitea_notification, dedup_key, message) + + logger.info( + "gitea_pr_merged_notification_scheduled", + repo=repo, + pr_number=pr.number, + author=author, + ) + + return GiteaWebhookResponse( + status="accepted", + message=f"PR #{pr.number} merge notification scheduled", + event_type="pull_request", + ) + # 只處理需要審查的 action supported_actions = {"opened", "synchronize", "reopened"} if payload.action not in supported_actions: @@ -498,7 +596,11 @@ async def handle_workflow_run( 處理 Gitea Actions workflow_run 事件 — ADR-074 M3 只處理 status=failure（或 conclusion=failure）的管線失敗。 - 建立 TYPE-1 Incident（純通知，不自動修復）。 + 雙路並行： + 1. 建立 TYPE-1 Incident（既有路徑，保持不變） + 2. 直接發 Telegram 通知（Task C 2026-04-25 新增） + - workflow name 含 deploy → "部署失敗" + - 否則 → "構建失敗" """ wf = payload.workflow_run if not wf: @@ -531,6 +633,7 @@ async def handle_workflow_run( run_url=run_url, ) + # 既有路徑：建立 TYPE-1 Incident (保持不變) async def _create_ci_incident() -> None: try: svc = get_incident_service() @@ -562,6 +665,26 @@ async def handle_workflow_run( background_tasks.add_task(_create_ci_incident) + # 新增路徑：直接 Telegram 通知 (Task C 2026-04-25 ogt + Claude Sonnet 4.6) + # workflow name 含 deploy 關鍵字 → 部署失敗；否則 → 構建失敗 + # 格式遵循 feedback_telegram_alert_format.md：狀態 + 資源 + 連結 + is_deploy = "deploy" in wf.name.lower() + event_label = "Deployment Failed" if is_deploy else "Build Failed" + run_link = f" | 查看日誌" if run_url else "" + + tg_message = ( + f"{event_label} | {repo}\n" + "──────────────────────\n" + f"├─ Workflow: {wf.name}\n" + f"├─ 分支: {branch}\n" + f"├─ Commit: {sha_short}\n" + f"└─ 狀態: failure{run_link}" + ) + + # 去重 key：同一 repo + workflow + branch + sha 的失敗，10 分鐘內不重複 + dedup_key = f"workflow_failure:{repo}:{wf.name}:{branch}:{sha_short}" + background_tasks.add_task(_send_gitea_notification, dedup_key, tg_message) + return GiteaWebhookResponse( status="accepted", message=f"CI pipeline failure for '{wf.name}' on '{branch}' queued as TYPE-1 incident", diff --git a/apps/api/tests/test_gitea_webhook.py b/apps/api/tests/test_gitea_webhook.py index 4e2f2859..0af97061 100644 --- a/apps/api/tests/test_gitea_webhook.py +++ b/apps/api/tests/test_gitea_webhook.py @@ -487,6 +487,253 @@ async def test_webhook_pr_unsupported_action(webhook_secret): assert "not supported" in data["message"] +# ============================================================================= +# Task C: Gitea CI/CD 告警轉發 Telegram 測試 (2026-04-25) +# ============================================================================= + +@pytest.mark.asyncio +async def test_webhook_pr_closed_not_merged_ignored(webhook_secret): + """測試: PR closed 但未合併 (merged=False) 應該被忽略（不發通知）""" + payload = { + "action": "closed", + "repository": { + "id": 123456, + "name": "test-repo", + "full_name": "test-owner/test-repo", + "private": False, + "html_url": "https://github.com/test-owner/test-repo", + }, + "sender": {"login": "test-user", "id": 1}, + "pull_request": { + "id": 1, + "number": 42, + "title": "Closed without merge", + "state": "closed", + "merged": False, # 明確標記未合併 + "html_url": "https://github.com/test-owner/test-repo/pull/42", + "diff_url": "https://github.com/test-owner/test-repo/pull/42.diff", + "user": {"login": "test-user", "id": 1}, + "head": {"ref": "feature", "sha": "abc"}, + "base": {"ref": "main", "sha": "def"}, + }, + } + + body, signature = prepare_request(webhook_secret, payload) + + async with httpx.AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + ) as client: + response = await client.post( + "/api/v1/webhooks/gitea", + content=body, + headers={ + "Content-Type": "application/json", + "X-Gitea-Event": "pull_request", + "X-Gitea-Delivery": "test-delivery-id", + "X-Gitea-Signature": signature, + }, + ) + + assert response.status_code == 202 + data = response.json() + assert data["status"] == "ignored" + assert "not supported" in data["message"] + + +@pytest.mark.asyncio +async def test_webhook_pr_merged_accepted(webhook_secret): + """測試: PR merged (action=closed + merged=True) 應該回 accepted 並排入通知""" + payload = { + "action": "closed", + "repository": { + "id": 123456, + "name": "test-repo", + "full_name": "test-owner/test-repo", + "private": False, + "html_url": "https://github.com/test-owner/test-repo", + }, + "sender": {"login": "test-user", "id": 1}, + "pull_request": { + "id": 1, + "number": 99, + "title": "feat: add CI/CD Telegram notifications", + "state": "closed", + "merged": True, # 合併完成 + "html_url": "https://github.com/test-owner/test-repo/pull/99", + "diff_url": "https://github.com/test-owner/test-repo/pull/99.diff", + "user": {"login": "test-user", "id": 1}, + "head": {"ref": "feat/ci-telegram", "sha": "abc123"}, + "base": {"ref": "main", "sha": "def456"}, + "additions": 150, + "deletions": 20, + "changed_files": 5, + }, + } + + body, signature = prepare_request(webhook_secret, payload) + + async with httpx.AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + ) as client: + response = await client.post( + "/api/v1/webhooks/gitea", + content=body, + headers={ + "Content-Type": "application/json", + "X-Gitea-Event": "pull_request", + "X-Gitea-Delivery": "test-delivery-id", + "X-Gitea-Signature": signature, + }, + ) + + assert response.status_code == 202 + data = response.json() + assert data["status"] == "accepted" + assert "merge notification" in data["message"] + assert data["event_type"] == "pull_request" + + +@pytest.mark.asyncio +async def test_webhook_workflow_run_failure_accepted(webhook_secret): + """測試: workflow_run failure 應該回 accepted 並排入 incident + Telegram 通知""" + payload = { + "repository": { + "id": 123456, + "name": "test-repo", + "full_name": "test-owner/test-repo", + "private": False, + "html_url": "https://github.com/test-owner/test-repo", + }, + "sender": {"login": "ci-bot", "id": 2}, + "workflow_run": { + "id": 9001, + "name": "CI Build", + "status": "failure", + "conclusion": "failure", + "head_sha": "deadbeef12345678", + "head_branch": "main", + "html_url": "https://github.com/test-owner/test-repo/actions/runs/9001", + }, + } + + body, signature = prepare_request(webhook_secret, payload) + + async with httpx.AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + ) as client: + response = await client.post( + "/api/v1/webhooks/gitea", + content=body, + headers={ + "Content-Type": "application/json", + "X-Gitea-Event": "workflow_run", + "X-Gitea-Delivery": "test-delivery-wf", + "X-Gitea-Signature": signature, + }, + ) + + assert response.status_code == 202 + data = response.json() + assert data["status"] == "accepted" + assert data["event_type"] == "workflow_run" + assert "CI Build" in data["message"] + + +@pytest.mark.asyncio +async def test_webhook_workflow_run_success_ignored(webhook_secret): + """測試: workflow_run success 應該被忽略（只關心失敗）""" + payload = { + "repository": { + "id": 123456, + "name": "test-repo", + "full_name": "test-owner/test-repo", + "private": False, + "html_url": "https://github.com/test-owner/test-repo", + }, + "sender": {"login": "ci-bot", "id": 2}, + "workflow_run": { + "id": 9002, + "name": "CI Build", + "status": "success", + "conclusion": "success", + "head_sha": "aabbccdd12345678", + "head_branch": "main", + "html_url": "https://github.com/test-owner/test-repo/actions/runs/9002", + }, + } + + body, signature = prepare_request(webhook_secret, payload) + + async with httpx.AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + ) as client: + response = await client.post( + "/api/v1/webhooks/gitea", + content=body, + headers={ + "Content-Type": "application/json", + "X-Gitea-Event": "workflow_run", + "X-Gitea-Delivery": "test-delivery-wf-ok", + "X-Gitea-Signature": signature, + }, + ) + + assert response.status_code == 202 + data = response.json() + assert data["status"] == "ignored" + assert "not a failure" in data["message"] + + +@pytest.mark.asyncio +async def test_webhook_deploy_failure_detected(webhook_secret): + """測試: workflow name 含 deploy → 事件標記為部署失敗（不同於構建失敗）""" + payload = { + "repository": { + "id": 123456, + "name": "test-repo", + "full_name": "test-owner/test-repo", + "private": False, + "html_url": "https://github.com/test-owner/test-repo", + }, + "sender": {"login": "ci-bot", "id": 2}, + "workflow_run": { + "id": 9003, + "name": "Deploy to Production", # 含 deploy + "status": "failure", + "conclusion": "failure", + "head_sha": "feed1234abcd5678", + "head_branch": "main", + "html_url": "https://github.com/test-owner/test-repo/actions/runs/9003", + }, + } + + body, signature = prepare_request(webhook_secret, payload) + + async with httpx.AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + ) as client: + response = await client.post( + "/api/v1/webhooks/gitea", + content=body, + headers={ + "Content-Type": "application/json", + "X-Gitea-Event": "workflow_run", + "X-Gitea-Delivery": "test-delivery-deploy", + "X-Gitea-Signature": signature, + }, + ) + + assert response.status_code == 202 + data = response.json() + assert data["status"] == "accepted" + assert data["event_type"] == "workflow_run" + + # ============================================================================= # 簽章格式測試 # ============================================================================= diff --git a/apps/api/tests/test_telegram_button_consistency.py b/apps/api/tests/test_telegram_button_consistency.py new file mode 100644 index 00000000..0112addb --- /dev/null +++ b/apps/api/tests/test_telegram_button_consistency.py @@ -0,0 +1,197 @@ +""" +Telegram 按鈕一致性測試 +====================== +驗證 send_info_notification 與 _send_approval_card_to_group 的 reply_markup 鏈路 + +測試策略 (遵循 feedback_no_mock_testing.md): +- 直接讀取 source code 驗證實作結構（不發送實際 Telegram 訊息） +- 鬼魂按鈕鐵律（feedback_no_ghost_buttons.md）: + 每個按鈕必須同時具備 callback_data 格式/handler/MCP能力 + +建立: 2026-04-25 (台北時區) +建立者: ogt + Claude Sonnet 4.6 +""" + +import re + + +def _read_gateway() -> str: + with open("src/services/telegram_gateway.py", encoding="utf-8") as f: + return f.read() + + +def _read_security_interceptor() -> str: + with open("src/services/security_interceptor.py", encoding="utf-8") as f: + return f.read() + + +# ============================================================================= +# Test: send_info_notification 按鈕結構 +# ============================================================================= + +class TestSendInfoNotificationKeyboard: + """驗證 TYPE-1 純資訊通知現在附帶 read-only 按鈕""" + + def test_send_info_notification_has_reply_markup(self): + """send_info_notification payload 包含 reply_markup""" + source = _read_gateway() + # 取出 send_info_notification 函式體（從定義到下一個 async def） + match = re.search( + r"async def send_info_notification.*?(?=\n async def |\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 send_info_notification 函式" + fn_body = match.group(0) + assert '"reply_markup"' in fn_body, ( + "send_info_notification 必須在 payload 中包含 reply_markup" + ) + + def test_send_info_notification_detail_button_correct_format(self): + """detail 按鈕使用 2-part info 格式 detail:{incident_id}""" + source = _read_gateway() + match = re.search( + r"async def send_info_notification.*?(?=\n async def |\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 send_info_notification 函式" + fn_body = match.group(0) + assert 'f"detail:{incident_id}"' in fn_body, ( + "detail 按鈕 callback_data 必須使用 detail:{incident_id} 格式（ADR-050 2-part info）" + ) + + def test_send_info_notification_history_button_correct_format(self): + """history 按鈕使用 2-part info 格式 history:{incident_id}""" + source = _read_gateway() + match = re.search( + r"async def send_info_notification.*?(?=\n async def |\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 send_info_notification 函式" + fn_body = match.group(0) + assert 'f"history:{incident_id}"' in fn_body, ( + "history 按鈕 callback_data 必須使用 history:{incident_id} 格式（ADR-050 2-part info）" + ) + + def test_send_info_notification_no_nonce_in_buttons(self): + """send_info_notification 按鈕不含 nonce（純查類，無副作用）""" + source = _read_gateway() + match = re.search( + r"async def send_info_notification.*?(?=\n async def |\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 send_info_notification 函式" + fn_body = match.group(0) + assert "generate_callback_nonce" not in fn_body, ( + "send_info_notification 的按鈕不應包含 nonce（TYPE-1 為純資訊，不應有寫類操作）" + ) + + +# ============================================================================= +# Test: _send_approval_card_to_group 按鈕結構 +# ============================================================================= + +class TestSendApprovalCardToGroupKeyboard: + """驗證群組卡片現在附帶 read-only 按鈕""" + + def test_group_card_calls_send_to_group_with_reply_markup(self): + """_send_approval_card_to_group 呼叫 send_to_group 時傳遞 reply_markup""" + source = _read_gateway() + match = re.search( + r"async def _send_approval_card_to_group.*?(?=\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 _send_approval_card_to_group 函式" + fn_body = match.group(0) + assert "reply_markup" in fn_body, ( + "_send_approval_card_to_group 必須傳 reply_markup 給 send_to_group" + ) + + def test_group_card_detail_button_correct_format(self): + """群組卡片 detail 按鈕使用 2-part info 格式""" + source = _read_gateway() + match = re.search( + r"async def _send_approval_card_to_group.*?(?=\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 _send_approval_card_to_group 函式" + fn_body = match.group(0) + assert 'f"detail:{incident_id}"' in fn_body, ( + "群組卡片 detail 按鈕必須使用 2-part info 格式" + ) + + def test_group_card_no_nonce_buttons(self): + """群組卡片絕不包含 nonce（ADR-075 斷點 C：防洩漏）""" + source = _read_gateway() + match = re.search( + r"async def _send_approval_card_to_group.*?(?=\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 _send_approval_card_to_group 函式" + fn_body = match.group(0) + assert "generate_callback_nonce" not in fn_body, ( + "_send_approval_card_to_group 群組訊息絕不含 nonce（ADR-075 斷點 C 安全設計）" + ) + + def test_group_card_keyboard_conditional_on_incident_id(self): + """群組按鍵僅在有 incident_id 時才建立（防 empty callback_data）""" + source = _read_gateway() + match = re.search( + r"async def _send_approval_card_to_group.*?(?=\n # ====)", + source, + re.DOTALL, + ) + assert match, "找不到 _send_approval_card_to_group 函式" + fn_body = match.group(0) + # 確認有 if incident_id 守衛 + assert "if incident_id" in fn_body, ( + "群組按鍵必須有 if incident_id 守衛，避免 callback_data 空字串" + ) + + +# ============================================================================= +# Test: callback handler 完整性（鬼魂按鈕鐵律） +# ============================================================================= + +class TestCallbackHandlerCompleteness: + """驗證 detail/history 按鈕有對應的 handler（鬼魂按鈕鐵律三條件之二）""" + + def test_detail_action_in_info_actions_whitelist(self): + """detail action 在 INFO_ACTIONS 白名單中""" + source = _read_security_interceptor() + assert '"detail"' in source, "detail 必須在 INFO_ACTIONS 白名單" + # 確認是在 INFO_ACTIONS 集合定義中 + assert "INFO_ACTIONS" in source + + def test_history_action_in_info_actions_whitelist(self): + """history action 在 INFO_ACTIONS 白名單中""" + source = _read_security_interceptor() + assert '"history"' in source, "history 必須在 INFO_ACTIONS 白名單" + + def test_detail_handler_exists_in_handle_callback(self): + """handle_callback 有 detail handler 分支""" + source = _read_gateway() + assert 'action == "detail"' in source, ( + "handle_callback 必須有 detail handler 分支" + ) + + def test_history_handler_exists_in_handle_callback(self): + """handle_callback 有 history handler 分支""" + source = _read_gateway() + assert 'action == "history"' in source, ( + "handle_callback 必須有 history handler 分支" + ) + + def test_send_incident_detail_method_exists(self): + """_send_incident_detail 方法存在（handler 的實際執行體）""" + assert "_send_incident_detail" in _read_gateway() + + def test_send_incident_history_method_exists(self): + """_send_incident_history 方法存在（handler 的實際執行體）""" + assert "_send_incident_history" in _read_gateway()