stack/apps/coordinator/tests/test_webhook.py

"""Tests for webhook endpoint handlers."""

import hmac
import json

import pytest
from fastapi.testclient import TestClient


class TestWebhookEndpoint:
    """Test suite for /webhook/gitea endpoint."""

    def _create_signature(self, payload: dict[str, object], secret: str) -> str:
        """Create HMAC SHA256 signature for payload."""
        # Use separators to match FastAPI's JSON encoding (no spaces)
        payload_bytes = json.dumps(payload, separators=(',', ':')).encode("utf-8")
        return hmac.new(secret.encode("utf-8"), payload_bytes, "sha256").hexdigest()

    def test_webhook_missing_signature(
        self, client: TestClient, sample_assigned_payload: dict[str, object]
    ) -> None:
        """Test that webhook without signature returns 401."""
        response = client.post("/webhook/gitea", json=sample_assigned_payload)
        assert response.status_code == 401
        assert "Invalid or missing signature" in response.json()["detail"]

    def test_webhook_invalid_signature(
        self, client: TestClient, sample_assigned_payload: dict[str, object]
    ) -> None:
        """Test that webhook with invalid signature returns 401."""
        headers = {"X-Gitea-Signature": "invalid_signature"}
        response = client.post(
            "/webhook/gitea", json=sample_assigned_payload, headers=headers
        )
        assert response.status_code == 401
        assert "Invalid or missing signature" in response.json()["detail"]

    def test_webhook_assigned_event(
        self,
        client: TestClient,
        sample_assigned_payload: dict[str, object],
        webhook_secret: str,
    ) -> None:
        """Test that assigned event is processed successfully."""
        signature = self._create_signature(sample_assigned_payload, webhook_secret)
        headers = {"X-Gitea-Signature": signature}

        response = client.post(
            "/webhook/gitea", json=sample_assigned_payload, headers=headers
        )

        assert response.status_code == 200
        assert response.json()["status"] == "success"
        assert response.json()["action"] == "assigned"
        assert response.json()["issue_number"] == 157

    def test_webhook_unassigned_event(
        self,
        client: TestClient,
        sample_unassigned_payload: dict[str, object],
        webhook_secret: str,
    ) -> None:
        """Test that unassigned event is processed successfully."""
        signature = self._create_signature(sample_unassigned_payload, webhook_secret)
        headers = {"X-Gitea-Signature": signature}

        response = client.post(
            "/webhook/gitea", json=sample_unassigned_payload, headers=headers
        )

        assert response.status_code == 200
        assert response.json()["status"] == "success"
        assert response.json()["action"] == "unassigned"
        assert response.json()["issue_number"] == 157

    def test_webhook_closed_event(
        self,
        client: TestClient,
        sample_closed_payload: dict[str, object],
        webhook_secret: str,
    ) -> None:
        """Test that closed event is processed successfully."""
        signature = self._create_signature(sample_closed_payload, webhook_secret)
        headers = {"X-Gitea-Signature": signature}

        response = client.post(
            "/webhook/gitea", json=sample_closed_payload, headers=headers
        )

        assert response.status_code == 200
        assert response.json()["status"] == "success"
        assert response.json()["action"] == "closed"
        assert response.json()["issue_number"] == 157

    def test_webhook_unsupported_action(
        self, client: TestClient, webhook_secret: str
    ) -> None:
        """Test that unsupported actions are handled gracefully."""
        payload = {
            "action": "opened",  # Not a supported action
            "number": 157,
            "issue": {"id": 157, "number": 157, "title": "Test"},
        }
        signature = self._create_signature(payload, webhook_secret)
        headers = {"X-Gitea-Signature": signature}

        response = client.post("/webhook/gitea", json=payload, headers=headers)

        assert response.status_code == 200
        assert response.json()["status"] == "ignored"
        assert response.json()["action"] == "opened"

    def test_webhook_malformed_payload(
        self, client: TestClient, webhook_secret: str
    ) -> None:
        """Test that malformed payload returns 422."""
        payload = {"invalid": "payload"}  # Missing required fields
        signature = self._create_signature(payload, webhook_secret)
        headers = {"X-Gitea-Signature": signature}

        response = client.post("/webhook/gitea", json=payload, headers=headers)

        assert response.status_code == 422

    def test_webhook_logs_events(
        self,
        client: TestClient,
        sample_assigned_payload: dict[str, object],
        webhook_secret: str,
        caplog: pytest.LogCaptureFixture,
    ) -> None:
        """Test that webhook events are logged."""
        signature = self._create_signature(sample_assigned_payload, webhook_secret)
        headers = {"X-Gitea-Signature": signature}

        with caplog.at_level("INFO"):
            response = client.post(
                "/webhook/gitea", json=sample_assigned_payload, headers=headers
            )

        assert response.status_code == 200
        # Check that event was logged
        assert any("Webhook event received" in record.message for record in caplog.records)
        assert any("action=assigned" in record.message for record in caplog.records)
        assert any("issue_number=157" in record.message for record in caplog.records)


class TestWebhookRateLimiting:
    """Test suite for webhook rate limiting."""

    def test_webhook_has_rate_limit_configured(self) -> None:
        """Test that webhook endpoint has rate limiting configured."""
        from src.webhook import handle_gitea_webhook

        # Verify the rate limit decorator is applied
        # slowapi adds __wrapped__ attribute to decorated functions
        assert hasattr(handle_gitea_webhook, "__wrapped__") or hasattr(
            handle_gitea_webhook, "__name__"
        )

        # Verify the endpoint is the webhook handler
        assert handle_gitea_webhook.__name__ == "handle_gitea_webhook"


class TestHealthEndpoint:
    """Test suite for /health endpoint."""

    def test_health_check_returns_200(self, client: TestClient) -> None:
        """Test that health check endpoint returns 200 OK."""
        response = client.get("/health")
        assert response.status_code == 200
        assert response.json()["status"] == "healthy"

    def test_health_check_includes_service_name(self, client: TestClient) -> None:
        """Test that health check includes service name."""
        response = client.get("/health")
        assert response.status_code == 200
        assert "service" in response.json()
        assert response.json()["service"] == "mosaic-coordinator"