feat(api): Phase 13.1 #74 GitHub Webhook → OpenClaw 整合
- POST /api/v1/webhooks/github endpoint - 處理 pull_request 和 push 事件 - 驗證 X-Hub-Signature-256 - Telegram 通知整合 - GitHubWebhookService 封裝 Redis 操作 (leWOOOgo 合規) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
516
apps/api/tests/test_github_webhook.py
Normal file
516
apps/api/tests/test_github_webhook.py
Normal file
@@ -0,0 +1,516 @@
|
||||
"""
|
||||
Phase 13.1: GitHub Webhook 整合測試
|
||||
===================================
|
||||
測試 GitHub Webhook → OpenClaw AI 代碼審查整合
|
||||
|
||||
測試策略 (遵循 feedback_no_mock_testing.md):
|
||||
- 使用 ASGITransport 撞擊真實端點
|
||||
- 不使用 Mock，直接測試 HTTP 層
|
||||
- 驗證 HMAC 簽章邏輯
|
||||
|
||||
🔴 IMPORTANT: 禁止 Mock 測試！
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from httpx import ASGITransport
|
||||
|
||||
# 設定測試環境變數 (在 import main 之前)
|
||||
os.environ.setdefault("MOCK_MODE", "true")
|
||||
os.environ.setdefault("ENVIRONMENT", "dev")
|
||||
os.environ.setdefault("GITHUB_WEBHOOK_SECRET", "test-secret-key-12345")
|
||||
os.environ.setdefault("GITHUB_ALLOWED_REPOS", "test-owner/test-repo,wooo-ai/awoooi")
|
||||
|
||||
from src.main import app # noqa: E402
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Test Fixtures
|
||||
# =============================================================================
|
||||
|
||||
@pytest.fixture
|
||||
def webhook_secret():
|
||||
"""測試用 Webhook Secret"""
|
||||
return "test-secret-key-12345"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_pr_payload():
|
||||
"""範例 PR Payload"""
|
||||
return {
|
||||
"action": "opened",
|
||||
"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,
|
||||
"avatar_url": "https://github.com/images/error/octocat_happy.gif",
|
||||
},
|
||||
"pull_request": {
|
||||
"id": 1,
|
||||
"number": 42,
|
||||
"title": "Add new feature",
|
||||
"body": "This PR adds a new feature",
|
||||
"state": "open",
|
||||
"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-branch", "sha": "abc123"},
|
||||
"base": {"ref": "main", "sha": "def456"},
|
||||
"additions": 50,
|
||||
"deletions": 10,
|
||||
"changed_files": 3,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_push_payload():
|
||||
"""範例 Push Payload"""
|
||||
return {
|
||||
"ref": "refs/heads/main",
|
||||
"before": "0000000000000000000000000000000000000000",
|
||||
"after": "abc1234567890abcdef1234567890abcdef12345",
|
||||
"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,
|
||||
},
|
||||
"pusher": {
|
||||
"name": "test-user",
|
||||
"email": "test@example.com",
|
||||
},
|
||||
"commits": [
|
||||
{
|
||||
"id": "abc1234567890abcdef1234567890abcdef12345",
|
||||
"message": "feat: add new feature",
|
||||
"timestamp": "2026-03-26T10:00:00+08:00",
|
||||
"url": "https://github.com/test-owner/test-repo/commit/abc123",
|
||||
"author": {"name": "Test User", "email": "test@example.com"},
|
||||
"added": ["new_file.py"],
|
||||
"removed": [],
|
||||
"modified": ["existing_file.py"],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ping_payload():
|
||||
"""Ping 事件 Payload"""
|
||||
return {
|
||||
"zen": "Responsive is better than fast.",
|
||||
"hook_id": 12345,
|
||||
"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,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def generate_signature(secret: str, body: bytes) -> str:
|
||||
"""生成 GitHub Webhook 簽章"""
|
||||
signature = hmac.new(
|
||||
secret.encode(),
|
||||
body,
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
return f"sha256={signature}"
|
||||
|
||||
|
||||
def prepare_request(secret: str, payload: dict) -> tuple[bytes, str]:
|
||||
"""準備請求 body 和簽章"""
|
||||
body = json.dumps(payload, separators=(',', ':')).encode()
|
||||
signature = generate_signature(secret, body)
|
||||
return body, signature
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# HMAC 簽章驗證測試
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_missing_signature(sample_pr_payload):
|
||||
"""測試: 缺少簽章應該被拒絕"""
|
||||
body = json.dumps(sample_pr_payload, separators=(',', ':')).encode()
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "pull_request",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
# 故意不提供 X-Hub-Signature-256
|
||||
},
|
||||
)
|
||||
|
||||
# 在 dev 環境下，缺少 secret 但配置了 secret，應該要求簽章
|
||||
# 由於我們設定了 GITHUB_WEBHOOK_SECRET，缺少簽章應該返回 401
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_invalid_signature(sample_pr_payload):
|
||||
"""測試: 無效簽章應該被拒絕"""
|
||||
body = json.dumps(sample_pr_payload, separators=(',', ':')).encode()
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "pull_request",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": "sha256=invalid_signature_here",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_valid_signature(sample_pr_payload, webhook_secret):
|
||||
"""測試: 有效簽章應該被接受"""
|
||||
body, signature = prepare_request(webhook_secret, sample_pr_payload)
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "pull_request",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
|
||||
# 應該返回 202 Accepted
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "accepted"
|
||||
assert data["event_type"] == "pull_request"
|
||||
assert data["review_id"] is not None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 倉庫白名單測試
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_repo_not_in_whitelist(webhook_secret):
|
||||
"""測試: 不在白名單的倉庫應該被忽略"""
|
||||
payload = {
|
||||
"action": "opened",
|
||||
"repository": {
|
||||
"id": 999999,
|
||||
"name": "unauthorized-repo",
|
||||
"full_name": "unknown-owner/unauthorized-repo", # 不在白名單
|
||||
"private": False,
|
||||
"html_url": "https://github.com/unknown-owner/unauthorized-repo",
|
||||
},
|
||||
"sender": {
|
||||
"login": "hacker",
|
||||
"id": 999,
|
||||
},
|
||||
"pull_request": {
|
||||
"id": 1,
|
||||
"number": 1,
|
||||
"title": "Malicious PR",
|
||||
"state": "open",
|
||||
"html_url": "https://github.com/unknown-owner/unauthorized-repo/pull/1",
|
||||
"diff_url": "https://github.com/unknown-owner/unauthorized-repo/pull/1.diff",
|
||||
"user": {"login": "hacker", "id": 999},
|
||||
"head": {"ref": "evil", "sha": "bad123"},
|
||||
"base": {"ref": "main", "sha": "good456"},
|
||||
},
|
||||
}
|
||||
|
||||
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/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "pull_request",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "ignored"
|
||||
assert "not in whitelist" in data["message"]
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 事件類型測試
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_ping_event(ping_payload, webhook_secret):
|
||||
"""測試: Ping 事件應該回應 Pong"""
|
||||
body, signature = prepare_request(webhook_secret, ping_payload)
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "ping",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "accepted"
|
||||
assert "Pong" in data["message"]
|
||||
assert data["event_type"] == "ping"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_push_event(sample_push_payload, webhook_secret):
|
||||
"""測試: Push 到主分支應該觸發審查"""
|
||||
body, signature = prepare_request(webhook_secret, sample_push_payload)
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "push",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "accepted"
|
||||
assert data["event_type"] == "push"
|
||||
assert data["review_id"] is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_push_non_default_branch(webhook_secret):
|
||||
"""測試: Push 到非主分支應該被忽略"""
|
||||
payload = {
|
||||
"ref": "refs/heads/feature-branch", # 非主分支
|
||||
"before": "0000000000000000000000000000000000000000",
|
||||
"after": "abc123",
|
||||
"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},
|
||||
"commits": [
|
||||
{
|
||||
"id": "abc123",
|
||||
"message": "feature commit",
|
||||
"timestamp": "2026-03-26T10:00:00+08:00",
|
||||
"url": "https://github.com/test-owner/test-repo/commit/abc123",
|
||||
"author": {"name": "Test", "email": "test@example.com"},
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
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/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "push",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "ignored"
|
||||
assert "non-default branch" in data["message"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_unsupported_event(webhook_secret):
|
||||
"""測試: 不支援的事件類型應該被忽略"""
|
||||
payload = {
|
||||
"action": "added",
|
||||
"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},
|
||||
}
|
||||
|
||||
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/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "star", # 不支援的事件
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": 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_unsupported_action(webhook_secret):
|
||||
"""測試: PR 不支援的 action 應該被忽略"""
|
||||
payload = {
|
||||
"action": "closed", # 我們只處理 opened, synchronize, reopened
|
||||
"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 PR",
|
||||
"state": "closed",
|
||||
"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/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "pull_request",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": 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_wrong_signature_format(sample_pr_payload):
|
||||
"""測試: 錯誤的簽章格式應該被拒絕"""
|
||||
body = json.dumps(sample_pr_payload, separators=(',', ':')).encode()
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/github",
|
||||
content=body,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-GitHub-Event": "pull_request",
|
||||
"X-GitHub-Delivery": "test-delivery-id",
|
||||
"X-Hub-Signature-256": "md5=wrong_format", # 錯誤格式
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 執行測試
|
||||
# =============================================================================
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user