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 == []


async def test_signed_unsupported_event_is_ignored_without_parsing() -> None:
    repository = FakeRepository()

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

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


@pytest.mark.parametrize(
    ("command", "requester"),
    [("ordinary discussion", "alice"), ("/agent plan", "AgentCI")],
)
async def test_non_command_and_bot_comment_are_ignored(command: str, requester: str) -> None:
    repository = FakeRepository()

    response = await post_webhook(repository, encoded(payload(command, requester=requester)))

    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",
    [
        b"{",
        encoded([]),
        encoded({"action": "created"}),
        encoded({"action": "created", "comment": {}}),
    ],
)
async def test_malformed_json_and_payload_contract_return_bad_request(body: bytes) -> None:
    repository = FakeRepository()

    response = await post_webhook(repository, body)

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


async def test_command_requires_delivery_header() -> None:
    repository = FakeRepository()

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

    assert response.status_code == 400
    assert response.json() == {"detail": "Missing X-Gitea-Delivery"}
    assert repository.accepted == []


async def test_non_created_comment_is_ignored_without_nested_payload() -> None:
    repository = FakeRepository()

    response = await post_webhook(repository, encoded({"action": "edited"}))

    assert response.status_code == 204
    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")