10.4Webhook Testing

Webhooks require a different approach: you spin up a temporary HTTP server, register it as a webhook target, trigger the event, and verify the webhook was called.

from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import json

class WebhookCapture:
    """Temporary HTTP server that captures webhook calls."""

    def __init__(self, port: int = 0):
        self.received = []
        self._server = None
        self._thread = None
        self.port = port

    def start(self):
        capture = self

        class Handler(BaseHTTPRequestHandler):
            def do_POST(self):
                content_length = int(self.headers.get('Content-Length', 0))
                body = self.rfile.read(content_length)
                capture.received.append({
                    "path": self.path,
                    "headers": dict(self.headers),
                    "body": json.loads(body) if body else None,
                    "timestamp": time.time(),
                })
                self.send_response(200)
                self.end_headers()

            def log_message(self, *args):
                pass  # Suppress request logging

        self._server = HTTPServer(("0.0.0.0", self.port), Handler)
        self.port = self._server.server_address[1]
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()
        return self

    def stop(self):
        if self._server:
            self._server.shutdown()

    def wait_for_call(self, timeout: int = 10) -> dict | None:
        deadline = time.time() + timeout
        while time.time() < deadline:
            if self.received:
                return self.received[-1]
            time.sleep(0.1)
        return None

    @property
    def url(self):
        return f"http://localhost:{self.port}"


class TestWebhookDelivery:
    """Test webhook delivery for order events."""

    @pytest.fixture
    def webhook_server(self):
        server = WebhookCapture().start()
        yield server
        server.stop()

    def test_order_webhook_delivered(self, api_client, webhook_server, admin_token):
        """Creating an order should trigger a webhook to registered URL."""
        # Register webhook
        api_client.post("/api/webhooks", json={
            "url": f"{webhook_server.url}/hooks/orders",
            "events": ["order.created"],
        }, headers={"Authorization": f"Bearer {admin_token}"})

        # Create an order (triggers the webhook)
        api_client.post("/api/orders", json={
            "items": [{"product_id": "prod-1", "quantity": 1}],
            "idempotency_key": "idem-1",
        }, headers={"Authorization": f"Bearer {admin_token}"})

        # Verify webhook was called
        call = webhook_server.wait_for_call(timeout=10)
        assert call is not None, "Webhook was not delivered within 10s"
        assert call["path"] == "/hooks/orders"
        assert call["body"]["event_type"] == "order.created"
        assert call["body"]["data"]["items"][0]["product_id"] == "prod-1"

    def test_webhook_retry_on_failure(self, api_client, admin_token):
        """Webhooks should retry when the target returns a 5xx error."""
        # Register webhook pointing to a non-existent server
        api_client.post("/api/webhooks", json={
            "url": "http://localhost:19999/will-fail",
            "events": ["order.created"],
        }, headers={"Authorization": f"Bearer {admin_token}"})

        # Create an order
        api_client.post("/api/orders", json={
            "items": [{"product_id": "prod-1", "quantity": 1}],
            "idempotency_key": "idem-retry",
        }, headers={"Authorization": f"Bearer {admin_token}"})

        # Check webhook delivery log shows retry attempts
        time.sleep(15)  # Wait for retry cycle
        log = api_client.get("/api/webhooks/delivery-log",
                           headers={"Authorization": f"Bearer {admin_token}"})
        deliveries = log.json()["deliveries"]
        failed = [d for d in deliveries if d["status"] == "failed"]
        assert len(failed) >= 2, "Expected at least 2 retry attempts"

    def test_webhook_signature_verification(self, webhook_server, api_client, admin_token):
        """Webhook payloads should include a signature header for verification."""
        api_client.post("/api/webhooks", json={
            "url": f"{webhook_server.url}/hooks/signed",
            "events": ["order.created"],
            "secret": "webhook-secret-123",
        }, headers={"Authorization": f"Bearer {admin_token}"})

        api_client.post("/api/orders", json={
            "items": [{"product_id": "prod-1", "quantity": 1}],
            "idempotency_key": "idem-signed",
        }, headers={"Authorization": f"Bearer {admin_token}"})

        call = webhook_server.wait_for_call(timeout=10)
        assert call is not None
        assert "X-Webhook-Signature" in call["headers"], (
            "Webhook signature header is missing"
        )