agentci/tests/test_webhook.py

import hashlib
import hmac
import json
from types import SimpleNamespace

import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient, Response

from agentci.api.routes.webhook import router
from agentci.engine.model import IncomingCommand
from agentci.integrations.gitea.webhooks import (
    incoming_command_from_payload,
    valid_signature,
)


class FakeRepository:
    def __init__(self, *, duplicate: bool = False) -> None:
        self.accepted: list[IncomingCommand] = []
        self.duplicate = duplicate

    async def accept(self, event: IncomingCommand) -> SimpleNamespace:
        self.accepted.append(event)
        job = SimpleNamespace(id="job", receive_sequence=1)
        return SimpleNamespace(job=job, duplicate=self.duplicate)


def payload(body: str, *, is_pull: bool = False, requester: str = "alice") -> dict:
    value = {
        "action": "created",
        "comment": {"id": 8, "body": body, "user": {"login": requester}},
        "repository": {"name": "repo", "owner": {"login": "org"}},
        "issue": {"number": 4},
        "is_pull": is_pull,
    }
    if is_pull:
        value["pull_request"] = {"number": 4}
    return value


def encoded(value: object) -> bytes:
    return json.dumps(value).encode()


def sign(body: bytes, secret: bytes = b"secret") -> str:
    return hmac.new(secret, body, hashlib.sha256).hexdigest()


async def post_webhook(
    repository: FakeRepository,
    body: bytes,
    *,
    event: str = "issue_comment",
    delivery: str | None = "delivery",
    signature: str | None = None,
    bot_username: str = "agentci",
) -> Response:
    app = FastAPI()
    app.include_router(router)
    app.state.runtime = SimpleNamespace(
        repository=repository,
        settings=SimpleNamespace(webhook_secret=b"secret", bot_username=bot_username),
    )
    headers = {
        "X-Gitea-Event-Type": event,
        "X-Gitea-Signature": sign(body) if signature is None else signature,
    }
    if delivery is not None:
        headers["X-Gitea-Delivery"] = delivery
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        return await client.post("/webhooks/gitea", content=body, headers=headers)


async def test_signed_supported_command_is_accepted_once_with_raw_body() -> None:
    repository = FakeRepository()
    body = encoded(payload("/agent iterate\n\nkeep raw body", is_pull=True))

    response = await post_webhook(repository, body)

    assert response.status_code == 202
    assert len(repository.accepted) == 1
    event = repository.accepted[0]
    assert event.delivery_id == "delivery"
    assert event.target_key == "org/repo:pr:4"
    assert event.requester == "alice"
    assert event.body == "/agent iterate\n\nkeep raw body"


@pytest.mark.parametrize("event", ["issue_comment", "push"])
async def test_invalid_signature_precedes_parsing_and_event_filtering(event: str) -> None:
    repository = FakeRepository()

    response = await post_webhook(repository, b"not-json", event=event, signature="invalid")

    assert response.status_code == 401
    assert response.json() == {"detail": "Invalid webhook signature"}
    assert repository.accepted == []


@pytest.mark.parametrize(
    ("body", "event"),
    [
        pytest.param(
            b"not-json",
            "push",
            id="unsupported-event-before-parsing",
        ),
        pytest.param(
            encoded(payload("ordinary discussion")),
            "issue_comment",
            id="non-command",
        ),
        pytest.param(
            encoded(payload("/agent plan", requester="AgentCI")),
            "issue_comment",
            id="bot-author",
        ),
        pytest.param(
            encoded({"action": "edited"}),
            "issue_comment",
            id="non-created-before-payload",
        ),
    ],
)
async def test_authenticated_irrelevant_webhooks_are_ignored(
    body: bytes,
    event: str,
) -> None:
    repository = FakeRepository()

    response = await post_webhook(repository, body, event=event)

    assert response.status_code == 204
    assert repository.accepted == []


async def test_duplicate_delivery_returns_ok() -> None:
    repository = FakeRepository(duplicate=True)

    response = await post_webhook(repository, encoded(payload("/agent plan")))

    assert response.status_code == 200
    assert len(repository.accepted) == 1


@pytest.mark.parametrize(
    ("body", "delivery", "detail"),
    [
        pytest.param(b"{", "delivery", "Invalid webhook payload", id="invalid-json"),
        pytest.param(encoded([]), "delivery", "Invalid webhook payload", id="non-object"),
        pytest.param(
            encoded({"action": "created"}),
            "delivery",
            "Invalid webhook payload",
            id="missing-comment",
        ),
        pytest.param(
            encoded({"action": "created", "comment": {}}),
            "delivery",
            "Invalid webhook payload",
            id="incomplete-comment",
        ),
        pytest.param(
            encoded(payload("/agent plan")),
            None,
            "Missing X-Gitea-Delivery",
            id="missing-delivery",
        ),
    ],
)
async def test_authenticated_bad_requests_are_rejected_without_persistence(
    body: bytes,
    delivery: str | None,
    detail: str,
) -> None:
    repository = FakeRepository()

    response = await post_webhook(repository, body, delivery=delivery)

    assert response.status_code == 400
    assert response.json() == {"detail": detail}
    assert repository.accepted == []


def test_payload_parser_supports_owner_username_and_pull_request() -> None:
    value = payload("/agent plan", is_pull=True)
    value["repository"]["owner"] = {"username": "fallback-owner"}

    event = incoming_command_from_payload("delivery", value)

    assert event is not None
    assert event.repo_owner == "fallback-owner"
    assert event.issue_number == 4
    assert event.pr_number == 4


def test_signature_validation_requires_exact_nonempty_digest() -> None:
    signature = sign(b"{}")

    assert valid_signature(b"secret", b"{}", signature)
    assert not valid_signature(b"secret", b"{}", "")
    assert not valid_signature(b"secret", b"{}", "bad")