Use Cases
- Testing webhook delivery reliability and retry logic
- Validating webhook payload formats and signatures
- Load testing webhook endpoints under high volume
- Testing webhook security (signatures, authentication)
- Simulating webhook failures and recovery
Basic Webhook Testing
Copy
Ask AI
from locust import HttpUser, task, between
import json
import time
import hashlib
import hmac
import uuid
class WebhookUser(HttpUser):
wait_time = between(1, 3)
def on_start(self):
"""Initialize webhook testing"""
self.webhook_secret = "webhook-secret-key-123"
self.delivery_attempts = {}
@task(3)
def test_webhook_delivery(self):
"""Test basic webhook payload delivery"""
webhook_id = str(uuid.uuid4())
payload = {
"event": "user.created",
"data": {
"user_id": f"user_{int(time.time())}",
"email": f"test{int(time.time())}@example.com",
"created_at": int(time.time())
},
"webhook_id": webhook_id,
"timestamp": int(time.time())
}
headers = {
'Content-Type': 'application/json',
'User-Agent': 'MyApp-Webhooks/1.0',
'X-Webhook-ID': webhook_id,
'X-Webhook-Event': 'user.created'
}
# Add webhook signature
signature = self._generate_signature(json.dumps(payload))
headers['X-Webhook-Signature'] = signature
response = self.client.post('/webhooks/user-events',
json=payload,
headers=headers,
name="webhook_delivery")
if response.status_code == 200:
self.delivery_attempts[webhook_id] = 'success'
elif response.status_code in [202, 204]:
self.delivery_attempts[webhook_id] = 'accepted'
else:
self.delivery_attempts[webhook_id] = 'failed'
def _generate_signature(self, payload):
"""Generate HMAC signature for webhook security"""
return hmac.new(
self.webhook_secret.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
@task(2)
def test_webhook_retry_logic(self):
"""Test webhook retry mechanism"""
webhook_id = str(uuid.uuid4())
payload = {
"event": "payment.failed",
"data": {
"payment_id": f"pay_{int(time.time())}",
"amount": 99.99,
"currency": "USD",
"failure_reason": "insufficient_funds"
},
"webhook_id": webhook_id,
"attempt": 1,
"max_attempts": 3
}
headers = {
'Content-Type': 'application/json',
'X-Webhook-ID': webhook_id,
'X-Webhook-Retry': 'true',
'X-Webhook-Signature': self._generate_signature(json.dumps(payload))
}
# First attempt - simulate failure by expecting specific response
response = self.client.post('/webhooks/payment-events',
json=payload,
headers=headers,
name="webhook_retry_attempt")
# Test retry with incremented attempt count
if response.status_code >= 400:
time.sleep(1) # Simulate retry delay
payload["attempt"] = 2
headers['X-Webhook-Signature'] = self._generate_signature(json.dumps(payload))
retry_response = self.client.post('/webhooks/payment-events',
json=payload,
headers=headers,
name="webhook_retry_attempt")
@task(2)
def test_webhook_validation(self):
"""Test webhook payload validation"""
test_cases = [
# Valid payload
{
"event": "order.completed",
"data": {
"order_id": f"order_{int(time.time())}",
"total": 149.99,
"items": [{"id": "item1", "quantity": 2}]
}
},
# Invalid payload - missing required fields
{
"event": "order.completed",
"data": {
"total": 149.99
# Missing order_id and items
}
},
# Invalid payload - wrong data types
{
"event": "order.completed",
"data": {
"order_id": 12345, # Should be string
"total": "invalid", # Should be number
"items": "not_array" # Should be array
}
}
]
for i, payload in enumerate(test_cases):
webhook_id = f"validation_{i}_{int(time.time())}"
payload["webhook_id"] = webhook_id
headers = {
'Content-Type': 'application/json',
'X-Webhook-ID': webhook_id,
'X-Webhook-Signature': self._generate_signature(json.dumps(payload))
}
response = self.client.post('/webhooks/order-events',
json=payload,
headers=headers,
name="webhook_validation")
# First payload should succeed, others should fail validation
expected_success = (i == 0)
if expected_success and response.status_code == 200:
continue
elif not expected_success and response.status_code == 400:
continue
else:
print(f"Validation test {i} unexpected result: {response.status_code}")
@task(1)
def test_webhook_security(self):
"""Test webhook security features"""
payload = {
"event": "security.test",
"data": {"test": True},
"webhook_id": str(uuid.uuid4())
}
# Test without signature (should fail)
response = self.client.post('/webhooks/security-test',
json=payload,
headers={'Content-Type': 'application/json'},
name="webhook_no_signature")
# Test with invalid signature (should fail)
headers = {
'Content-Type': 'application/json',
'X-Webhook-Signature': 'invalid-signature'
}
response = self.client.post('/webhooks/security-test',
json=payload,
headers=headers,
name="webhook_invalid_signature")
# Test with valid signature (should succeed)
headers['X-Webhook-Signature'] = self._generate_signature(json.dumps(payload))
response = self.client.post('/webhooks/security-test',
json=payload,
headers=headers,
name="webhook_valid_signature")
Advanced Webhook Testing
Copy
Ask AI
from locust import HttpUser, task, between
import json
import time
import random
class AdvancedWebhookUser(HttpUser):
wait_time = between(1, 4)
def on_start(self):
"""Initialize advanced webhook testing"""
self.webhook_types = [
'user.created', 'user.updated', 'user.deleted',
'order.created', 'order.paid', 'order.shipped',
'payment.succeeded', 'payment.failed', 'payment.refunded'
]
self.batch_id = str(uuid.uuid4())
@task(2)
def test_webhook_batch_delivery(self):
"""Test batch webhook delivery"""
batch_size = random.randint(5, 15)
webhooks = []
for i in range(batch_size):
webhook = {
"webhook_id": f"batch_{self.batch_id}_{i}",
"event": random.choice(self.webhook_types),
"data": {
"id": f"entity_{i}_{int(time.time())}",
"batch_id": self.batch_id,
"sequence": i
},
"timestamp": int(time.time())
}
webhooks.append(webhook)
payload = {
"batch_id": self.batch_id,
"webhooks": webhooks,
"total_count": len(webhooks)
}
headers = {
'Content-Type': 'application/json',
'X-Batch-ID': self.batch_id,
'X-Webhook-Count': str(len(webhooks))
}
response = self.client.post('/webhooks/batch',
json=payload,
headers=headers,
name="webhook_batch_delivery")
@task(1)
def test_webhook_idempotency(self):
"""Test webhook idempotency handling"""
webhook_id = str(uuid.uuid4())
payload = {
"event": "test.idempotency",
"data": {"test_id": webhook_id},
"webhook_id": webhook_id,
"idempotency_key": f"idem_{webhook_id}"
}
headers = {
'Content-Type': 'application/json',
'X-Webhook-ID': webhook_id,
'X-Idempotency-Key': payload["idempotency_key"]
}
# Send same webhook twice
for attempt in range(2):
response = self.client.post('/webhooks/idempotency-test',
json=payload,
headers=headers,
name=f"webhook_idempotency_attempt_{attempt + 1}")
# Both attempts should succeed (idempotent)
if response.status_code not in [200, 202]:
print(f"Idempotency test failed on attempt {attempt + 1}")
@task(1)
def test_webhook_timeout_handling(self):
"""Test webhook timeout scenarios"""
payload = {
"event": "test.timeout",
"data": {
"processing_time": random.choice([1, 5, 10, 30]), # seconds
"should_timeout": random.choice([True, False])
},
"webhook_id": str(uuid.uuid4())
}
headers = {
'Content-Type': 'application/json',
'X-Timeout-Test': 'true'
}
# Set shorter timeout for testing
response = self.client.post('/webhooks/timeout-test',
json=payload,
headers=headers,
timeout=15, # 15 second timeout
name="webhook_timeout_test")
def on_stop(self):
"""Cleanup webhook test data"""
cleanup_payload = {
"action": "cleanup",
"batch_id": self.batch_id,
"timestamp": int(time.time())
}
try:
self.client.post('/webhooks/cleanup',
json=cleanup_payload,
headers={'Content-Type': 'application/json'},
name="webhook_cleanup")
except:
pass # Ignore cleanup errors
Webhook Testing Patterns
- Delivery Verification: Confirm webhooks reach endpoints successfully
- Retry Logic: Test automatic retry mechanisms and backoff strategies
- Payload Validation: Verify webhook data format and required fields
- Security Testing: Validate signatures, authentication, and authorization
- Idempotency: Test duplicate webhook handling
- Batch Processing: Test bulk webhook delivery
- Timeout Handling: Test webhook processing time limits
Common Webhook Events
- User Events: Registration, profile updates, deletions
- Order Events: Creation, payment, fulfillment, cancellation
- Payment Events: Success, failure, refunds, disputes
- System Events: Maintenance, alerts, status changes
- Integration Events: Third-party service notifications
Copy
Ask AI
This guide provides comprehensive webhook testing patterns for reliable event-driven architectures with proper validation and error handling.