Test webhook endpoints with validation, retry logic, and security checks
LoadForge can record your browser, graphically build tests, scan your site with a wizard and more. Sign up now to run your first test.
This guide demonstrates how to test webhook endpoints with LoadForge, including payload validation, retry mechanisms, and security verification.
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")
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
This guide provides comprehensive webhook testing patterns for reliable event-driven architectures with proper validation and error handling.