Test webhooks in CI
If you build a service that sends webhooks, you need to prove in tests that the right payload, headers, and signature actually went out. The usual options are to run a mock HTTP server inside the test or to skip the assertion. catchbin gives you a third: point the service under test at a real catchbin URL and query the API for what arrived.
This is the same idea as Mailpit or Mailhog for email — a purpose-built spy your tests can query.
The pattern
Section titled “The pattern”-
Record a start time just before the action under test.
-
Trigger the action that should fire a webhook (for example, create a subscription).
-
Long-poll the events API for events that arrived after your start time. The
waitparameter holds the request open until an event shows up or the wait elapses, so you do not need to sleep-and-retry. -
Assert on the count, the event type, and the body fields.
import os, time, requests
API = "https://api.catchbin.io/v1"AUTH = {"Authorization": f"Bearer {os.environ['CATCHBIN_API_KEY']}"}endpoint_id = os.environ["CATCHBIN_ENDPOINT_ID"]
start = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
# Action under test — your service should POST a webhook to its catchbin URL.my_service.create_subscription(customer_id="cus_123")
# Long-poll: return as soon as an event arrives after `start`, or after 5s.resp = requests.get( f"{API}/endpoints/{endpoint_id}/events", params={"from": start, "wait": "5s", "limit": 10}, headers=AUTH,).json()
events = resp["events"]assert len(events) == 1assert events[0]["eventType"] == "subscription.created"Isolating parallel tests with ephemeral endpoints
Section titled “Isolating parallel tests with ephemeral endpoints”If many tests share one endpoint, they see each other’s events. Create a fresh endpoint per test and delete it afterwards:
# setupcreated = requests.post(f"{API}/endpoints", json={"slug": f"test-{uuid4().hex[:8]}"}, headers=AUTH).json()endpoint_id = created["id"]webhook_url = created["url"] # https://hooks.catchbin.io/wk_.../test-xxxxmy_service.set_webhook_url(webhook_url)
# ... run the test, assert via the events API ...
# teardownrequests.delete(f"{API}/endpoints/{endpoint_id}", headers=AUTH)What you can assert on
Section titled “What you can assert on”The events API returns each captured request with its headers, body, event type, response code, and signature-verification result. That is enough to assert that your service sent the right event, to the right URL, with a body that matches — and, because catchbin verifies signatures, that your outgoing signature is correct.