wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(t0): Task A 按鈕一致性測試 + Task C Gitea→Telegram 通知收尾

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-04-26 20:17:17 +08:00
parent 7cd53c0228
commit d3a4fb4d15
3 changed files with 568 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
apps/api/tests/test_gitea_webhook.py
View File

@@ -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"
# =============================================================================
# 簽章格式測試
# =============================================================================

197
apps/api/tests/test_telegram_button_consistency.py Normal file
View File

@@ -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()