#!/usr/bin/env python3
"""Test Stripe webhook handler with simulated signed events.
This script creates properly signed Stripe webhook events and sends them
to the local AgentForms relay service to verify the webhook handler.
Usage:
# Option 1: Direct test (no Stripe CLI needed)
python3 scripts/test_stripe_webhook.py
# Option 2: Generate events for Stripe CLI
# Run Stripe CLI forward first:
# stripe listen --forward-to http://localhost:5060/billing/stripe/webhook
# Then trigger events from your Stripe dashboard or this script.
Requirements:
- STRIPE_WEBHOOK_SECRET env var (or set in data/.env)
- Relay service running on localhost:5060
"""
import hmac
import hashlib
import json
import os
import sys
import time
import urllib.request
import urllib.error
# Load .env for test config
try:
from dotenv import load_dotenv
load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", ".env"))
except ImportError:
pass
WEBHOOK_URL = os.environ.get("TEST_WEBHOOK_URL", "http://localhost:5060/billing/stripe/webhook")
WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_test_default_secret_for_local")
VERBOSE = "--verbose" in sys.argv or os.environ.get("VERBOSE", "").lower() == "true"
def log(msg):
print(msg)
def generate_signature(payload: str, secret: str) -> str:
"""Generate a Stripe-compatible webhook signature.
Stripe signature format: {timestamp}.{signature}
Algorithm: HMAC-SHA256 of "{timestamp}.{payload}"
"""
timestamp = str(int(time.time()))
signed_payload = f"{timestamp}.{payload}"
signature = hmac.new(
secret.encode("utf-8"),
signed_payload.encode("utf-8"),
hashlib.sha256,
).hexdigest()
return f"t={timestamp},v1={signature}"
def send_webhook(event_type: str, data: dict, secret: str | None = None) -> dict:
"""Send a simulated Stripe webhook event and return the response."""
secret = secret or WEBHOOK_SECRET
event = {
"id": f"evt_test_{int(time.time() * 1000)}",
"object": "event",
"type": event_type,
"api_version": "2024-12-18.acacia",
"created": int(time.time()),
"data": {
"object": data,
},
"pending_webhooks": [],
}
payload = json.dumps(event)
sig = generate_signature(payload, secret)
req = urllib.request.Request(
WEBHOOK_URL,
data=payload.encode("utf-8"),
headers={
"Content-Type": "application/json",
"Stripe-Signature": sig,
"User-Agent": "agentforms-webhook-test",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
body = resp.read().decode("utf-8")
return {"status": resp.status, "body": body}
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8")
return {"status": e.code, "body": body}
except Exception as e:
return {"status": 0, "error": str(e)}
def test_checkout_completed():
"""Test checkout.session.completed event (subscription upgrade)."""
log("\n=== Test: checkout.session.completed ===")
# Use a real-looking but test price ID
data = {
"id": "cs_test_checkout_001",
"object": "checkout.session",
"mode": "subscription",
"payment_status": "paid",
"status": "complete",
"customer_details": {
"email": "test-webhook@example.com",
"name": "Webhook Test User",
},
"subscription": "sub_test_001",
"line_items": {
"object": "list",
"data": [
{
"id": "li_test_001",
"object": "line_item",
"price": {
"id": os.environ.get("STRIPE_STARTER_PRICE_ID", "price_test_starter"),
"unit_amount": 900,
"currency": "usd",
"recurring": {"interval": "month"},
},
"quantity": 1,
}
],
},
"metadata": {},
}
result = send_webhook("checkout.session.completed", data)
log(f"Response: {result}")
assert result["status"] == 200, f"Expected 200, got {result['status']}: {result.get('body')}"
log("✓ checkout.session.completed handled successfully")
return result
def test_subscription_updated():
"""Test customer.subscription.updated event."""
log("\n=== Test: customer.subscription.updated ===")
data = {
"id": "sub_test_002",
"object": "subscription",
"status": "active",
"customer": "cus_test_webhook_001",
"current_period_end": int(time.time()) + 86400 * 30,
"items": {
"object": "list",
"data": [
{
"id": "si_test_002",
"object": "subscription_item",
"price": {
"id": os.environ.get("STRIPE_PRO_PRICE_ID", "price_test_pro"),
"unit_amount": 2900,
"currency": "usd",
"recurring": {"interval": "month"},
},
"quantity": 1,
}
],
},
}
result = send_webhook("customer.subscription.updated", data)
log(f"Response: {result}")
assert result["status"] == 200, f"Expected 200, got {result['status']}: {result.get('body')}"
log("✓ customer.subscription.updated handled successfully")
return result
def test_subscription_deleted():
"""Test customer.subscription.deleted event (downgrade to free)."""
log("\n=== Test: customer.subscription.deleted ===")
data = {
"id": "sub_test_003",
"object": "subscription",
"status": "canceled",
"customer": "cus_test_webhook_001",
"canceled_at": int(time.time()),
"current_period_end": int(time.time()),
}
result = send_webhook("customer.subscription.deleted", data)
log(f"Response: {result}")
assert result["status"] == 200, f"Expected 200, got {result['status']}: {result.get('body')}"
log("✓ customer.subscription.deleted handled successfully")
return result
def test_invoice_payment_succeeded():
"""Test invoice.payment_succeeded event."""
log("\n=== Test: invoice.payment_succeeded ===")
data = {
"id": "in_test_001",
"object": "invoice",
"status": "paid",
"payment_intent": "pi_test_webhook_001",
"customer": "cus_test_webhook_001",
"lines": {
"object": "list",
"data": [
{
"id": "il_test_001",
"object": "line_item",
"amount": 900,
"currency": "usd",
}
],
},
}
result = send_webhook("invoice.payment_succeeded", data)
log(f"Response: {result}")
assert result["status"] == 200, f"Expected 200, got {result['status']}: {result.get('body')}"
log("✓ invoice.payment_succeeded handled successfully")
return result
def test_duplicate_event():
"""Test idempotency — sending the same event twice."""
log("\n=== Test: Duplicate event (idempotency) ===")
event_id = f"evt_duplicate_{int(time.time())}"
data = {
"id": "sub_test_dup",
"object": "subscription",
"status": "active",
"customer": "cus_test_webhook_dup",
"items": {
"object": "list",
"data": [
{
"price": {
"id": "price_test_dup",
},
"quantity": 1,
}
],
},
}
secret = WEBHOOK_SECRET
event = {
"id": event_id,
"object": "event",
"type": "customer.subscription.updated",
"api_version": "2024-12-18.acacia",
"created": int(time.time()),
"data": {"object": data},
"pending_webhooks": [],
}
payload = json.dumps(event)
sig = generate_signature(payload, secret)
# First request
req = urllib.request.Request(
WEBHOOK_URL,
data=payload.encode("utf-8"),
headers={
"Content-Type": "application/json",
"Stripe-Signature": sig,
},
method="POST",
)
with urllib.request.urlopen(req, timeout=10) as resp:
body1 = json.loads(resp.read().decode("utf-8"))
# Second request (same event)
with urllib.request.urlopen(req, timeout=10) as resp:
body2 = json.loads(resp.read().decode("utf-8"))
log(f"First response: {body1}")
log(f"Second response: {body2}")
assert body2.get("dedup") is True, f"Expected dedup=True on second request, got {body2}"
log("✓ Duplicate event correctly detected and skipped")
return body2
def test_invalid_signature():
"""Test that invalid signatures are rejected."""
log("\n=== Test: Invalid signature rejection ===")
req = urllib.request.Request(
WEBHOOK_URL,
data=json.dumps({"type": "test"}).encode("utf-8"),
headers={
"Content-Type": "application/json",
"Stripe-Signature": "t=123,v1=invalid_signature",
},
method="POST",
)
try:
urllib.request.urlopen(req, timeout=10)
log("✗ Invalid signature was NOT rejected!")
assert False, "Expected 400 for invalid signature"
except urllib.error.HTTPError as e:
log(f"Response: {e.code} (expected 400)")
assert e.code == 400, f"Expected 400, got {e.code}"
log("✓ Invalid signature correctly rejected with 400")
except Exception as e:
log(f"Error: {e}")
assert False, f"Unexpected error: {e}"
def test_unhandled_event_type():
"""Test that unknown event types are logged and accepted."""
log("\n=== Test: Unhandled event type ===")
data = {
"id": "cus_test_unknown",
"object": "customer",
"email": "unknown@example.com",
}
result = send_webhook("customer.created", data)
log(f"Response: {result}")
assert result["status"] == 200, f"Expected 200, got {result['status']}: {result.get('body')}"
log("✓ Unhandled event type accepted (should be logged)")
return result
def main():
log(f"Stripe Webhook Handler Test Suite")
log(f"=" * 50)
log(f"Target: {WEBHOOK_URL}")
log(f"Secret: {WEBHOOK_SECRET[:8]}...")
log(f"")
log("Note: This script sends REAL webhook events to your running relay.")
log("Make sure the relay service is running before executing.")
log("")
passed = 0
failed = 0
tests = [
("checkout.session.completed", test_checkout_completed),
("customer.subscription.updated", test_subscription_updated),
("customer.subscription.deleted", test_subscription_deleted),
("invoice.payment_succeeded", test_invoice_payment_succeeded),
("duplicate event (idempotency)", test_duplicate_event),
("invalid signature rejection", test_invalid_signature),
("unhandled event type", test_unhandled_event_type),
]
for name, test_fn in tests:
try:
test_fn()
passed += 1
except AssertionError as e:
log(f"✗ FAILED: {name} — {e}")
failed += 1
except Exception as e:
log(f"✗ ERROR: {name} — {e}")
failed += 1
log(f"\n{'=' * 50}")
log(f"Results: {passed} passed, {failed} failed")
if failed > 0:
log("\nSome tests failed. Check the relay logs for details:")
log(" docker logs relay # if running in Docker")
sys.exit(1)
else:
log("\nAll tests passed! The webhook handler is working correctly.")
sys.exit(0)
if __name__ == "__main__":
main()