awoooi/apps/api/tests/test_gitea_webhook.py

"""
ADR-059: Gitea Webhook 整合測試
===================================
測試 Gitea Webhook → OpenClaw AI 代碼審查整合

測試策略 (遵循 feedback_no_mock_testing.md):
- 使用 ASGITransport 撞擊真實端點
- 不使用 Mock，直接測試 HTTP 層
- 驗證 HMAC 簽章邏輯 (X-Gitea-Signature)

🔴 IMPORTANT: 禁止 Mock 測試！
"""

import hashlib
import hmac
import json

import httpx
import pytest
from fastapi import FastAPI
from httpx import ASGITransport

# 2026-04-05 Claude Code: 改用最小化 app，只掛載 gitea_webhook router
# 原 `from src.main import app` 會 import 整個應用，觸發 sqlalchemy.ext.asyncio
# C extension (asyncpg.protocol.protocol) 在 CI runner 上 segfault (exit 139)
# gitea_webhook router 的 import chain 不走 DB，可獨立測試
from src.api.v1.gitea_webhook import router as gitea_webhook_router

app = FastAPI()
app.include_router(gitea_webhook_router, prefix="/api/v1")

# 環境變數設定已移至 conftest.py (解決 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:
    """生成 Gitea Webhook 簽章 (X-Gitea-Signature)

    Gitea 送出純 hex（無 sha256= 前綴），與 GitHub 不同。
    2026-04-05 ogt: 修正為純 hex 格式
    """
    return hmac.new(
        secret.encode(),
        body,
        hashlib.sha256,
    ).hexdigest()


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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "pull_request",
                "X-Gitea-Delivery": "test-delivery-id",
                # 故意不提供 X-Gitea-Signature
            },
        )

        # 在 dev 環境下，缺少 secret 但配置了 secret，應該要求簽章
        # 由於我們設定了 GITEA_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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "pull_request",
                "X-Gitea-Delivery": "test-delivery-id",
                "X-Gitea-Signature": "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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "pull_request",
                "X-Gitea-Delivery": "test-delivery-id",
                "X-Gitea-Signature": 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/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 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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "ping",
                "X-Gitea-Delivery": "test-delivery-id",
                "X-Gitea-Signature": 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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "push",
                "X-Gitea-Delivery": "test-delivery-id",
                "X-Gitea-Signature": 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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "push",
                "X-Gitea-Delivery": "test-delivery-id",
                "X-Gitea-Signature": 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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "star",  # 不支援的事件
                "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_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/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_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/gitea",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Gitea-Event": "pull_request",
                "X-Gitea-Delivery": "test-delivery-id",
                "X-Gitea-Signature": "md5=wrong_format",  # 錯誤格式
            },
        )

        assert response.status_code == 401


# =============================================================================
# 執行測試
# =============================================================================

if __name__ == "__main__":
    pytest.main([__file__, "-v"])