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.engine.model import IncomingCommand
from agentci.webhook import _event_from_payload, router, 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 = _event_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")