Testing Webhooks

Test webhook implementations locally and in production using various testing strategies.

Dashboard Testing

Send Test Event

Quick test from dashboard:

Settings → Webhooks
Select your webhook
Click “Send Test” button
Choose event type from dropdown
Review payload
Click “Send”
Check delivery status

The test event:

Uses real event schema
Includes valid signature
Logs like any normal event
Can be tested multiple times

View Delivery Status

After sending test event:

Settings → Webhooks → [Your Webhook]
Scroll to “Delivery Status” tab
See if delivery succeeded/failed
View response code and body
Check logs for errors

Local Development with ngrok

1. Start Local Server

Python Flask:

from flask import Flask, request
import json

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    print(f"Received: {request.json}")
    return {'status': 'success'}, 200

if __name__ == '__main__':
    app.run(port=5000)

Node.js Express:

const express = require('express');
const app = express();
app.use(express.json());

app.post('/webhook', (req, res) => {
  console.log('Received:', req.body);
  res.json({ status: 'success' });
});

app.listen(5000, () => console.log('Listening on 5000'));

Start server:

# Python
python app.py

# Node.js
node app.js

2. Expose with ngrok

# Download from https://ngrok.com/download
./ngrok http 5000

Output:

ngrok by @inconshreverous (Ctrl+C to quit)

Session Status    online
Account           yourname@email.com
Version           3.3.5
Region            us (United States)
Forwarding        https://abc-123-def.ngrok.io -> http://localhost:5000
Connections       ttl opn rt1 rt5 rt10
                  0   0   0   0   0

3. Configure Webhook

In TruthVouch dashboard:

Settings → Webhooks
Edit or create webhook
Set URL to: https://abc-123-def.ngrok.io/webhook
Save

4. Test

Send test event from dashboard. Watch local server:

127.0.0.1 - - [15/Jan/2024 10:30:00] "POST /webhook HTTP/1.1" 200 -
Received: {'event_id': 'evt-abc...', ...}

Webhook.site Testing

1. Get Unique URL

Visit https://webhook.site

Generates unique URL automatically
Example: https://webhook.site/abc-123-def

2. Configure Webhook

In TruthVouch dashboard:

Settings → Webhooks
Create webhook
Set URL to your webhook.site URL
Save

3. Send Test Event

Click “Send Test” in TruthVouch. Watch webhook.site:

See real-time delivery
View headers (including signature)
View full request/response
Export as cURL command

4. Verify Signature

On webhook.site, view headers:

X-TruthVouch-Signature: t=1705314600,v1=abcdef...

Copy header and verify locally:

from signature import verify_signature

header = "t=1705314600,v1=abcdef..."
body = '{"event_id":"evt-abc",...}'
secret = "whsec_..."

if verify_signature(header, body, secret):
    print("✓ Signature valid!")
else:
    print("✗ Signature invalid")

Automated Testing

Unit Tests (Signature Verification)

Python pytest:

import pytest
import hmac
import hashlib
from your_app import verify_signature

def test_valid_signature():
    """Test signature verification passes for valid signature."""
    secret = "whsec_test123"
    timestamp = 1705314600
    body = '{"event_id":"evt-test"}'

    signed_content = f"{timestamp}.{body}"
    signature = hmac.new(
        secret.encode(),
        signed_content.encode(),
        hashlib.sha256
    ).hexdigest()

    header = f"t={timestamp},v1={signature}"
    assert verify_signature(header, body, secret) == True

def test_invalid_signature():
    """Test signature verification fails for invalid signature."""
    header = "t=1705314600,v1=invalid"
    body = '{"event_id":"evt-test"}'
    secret = "whsec_test123"

    assert verify_signature(header, body, secret) == False

def test_expired_timestamp():
    """Test signature verification fails for old timestamp."""
    import time

    secret = "whsec_test123"
    old_timestamp = int(time.time()) - 400  # 400 seconds old
    body = '{"event_id":"evt-test"}'

    signed_content = f"{old_timestamp}.{body}"
    signature = hmac.new(
        secret.encode(),
        signed_content.encode(),
        hashlib.sha256
    ).hexdigest()

    header = f"t={old_timestamp},v1={signature}"
    assert verify_signature(header, body, secret) == False  # Expired

Run tests:

pytest test_webhooks.py -v

Integration Tests

Test event processing:

import json
import requests

def test_webhook_processing():
    """Test complete webhook flow."""
    # 1. Create test event
    event = {
        "event_id": "evt-test-123",
        "event_type": "alert.detected",
        "timestamp": "2024-01-15T10:30:00Z",
        "alert_id": "alert-xyz"
    }

    # 2. Generate signature
    import time, hmac, hashlib
    timestamp = int(time.time())
    body = json.dumps(event)
    secret = "whsec_test123"

    signed_content = f"{timestamp}.{body}"
    signature = hmac.new(
        secret.encode(),
        signed_content.encode(),
        hashlib.sha256
    ).hexdigest()

    # 3. Send to webhook
    response = requests.post(
        'http://localhost:5000/webhook',
        json=event,
        headers={
            'X-TruthVouch-Signature': f't={timestamp},v1={signature}'
        }
    )

    # 4. Verify response
    assert response.status_code == 200
    assert response.json()['status'] == 'success'

Performance Testing

Load Testing with Apache Bench

# Generate test event with signature
TIMESTAMP=$(date +%s)
BODY='{"event_id":"evt-test"}'
SIGNATURE=$(echo -n "$TIMESTAMP.$BODY" | \
  openssl dgst -sha256 -mac HMAC -macopt "key:whsec_test" -hex | \
  cut -d' ' -f2)

# Send 100 requests, 10 concurrent
ab -n 100 -c 10 \
  -H "X-TruthVouch-Signature: t=$TIMESTAMP,v1=$SIGNATURE" \
  -H "Content-Type: application/json" \
  -p test.json \
  https://yourserver.com/webhook

Load Testing with wrk

# Create script to add signature header
cat > script.lua << 'EOF'
local timestamp = os.time()
local body = '{"event_id":"evt-test"}'
local secret = "whsec_test"

-- Note: wrk doesn't have built-in HMAC, use Go version instead
EOF

# Run test
wrk -t 4 -c 100 -d 30s \
  https://yourserver.com/webhook

Testing in Production

Canary Testing

Test webhook in production without affecting real events:

Create second webhook endpoint (/webhook-test)
In dashboard: Create second webhook pointing to test endpoint
Use dashboard “Send Test” on second webhook
Monitor test endpoint for issues
When confident, migrate real traffic

Blue/Green Testing

# Route based on header
@app.route('/webhook', methods=['POST'])
def webhook():
    version = request.headers.get('X-Webhook-Version', 'blue')

    if version == 'green':
        # New implementation
        return process_event_v2(request.json)
    else:
        # Current implementation
        return process_event_v1(request.json)

Shadow Testing

Send each event to both old and new implementations, compare results:

@app.route('/webhook', methods=['POST'])
def webhook():
    event = request.json

    # Process with current version
    result_v1 = process_event_v1(event)

    # Shadow test with new version
    try:
        result_v2 = process_event_v2(event)

        # Compare results
        if result_v1 != result_v2:
            logger.warning(f"Result mismatch: {event['event_id']}")
    except Exception as e:
        logger.error(f"Shadow test error: {e}")

    # Return v1 result
    return result_v1

Monitoring and Alerting

Set Up Monitoring

Prometheus metrics:

from prometheus_client import Counter, Histogram

webhook_requests = Counter(
    'webhook_requests_total',
    'Total webhook requests',
    ['status']
)

webhook_processing_time = Histogram(
    'webhook_processing_seconds',
    'Webhook processing time'
)

@app.route('/webhook', methods=['POST'])
def webhook():
    with webhook_processing_time.time():
        try:
            event = request.json
            process_event(event)
            webhook_requests.labels(status='success').inc()
            return {'status': 'success'}, 200
        except Exception as e:
            webhook_requests.labels(status='error').inc()
            logger.error(f"Error: {e}")
            return {'error': str(e)}, 500

Set Up Alerts

AlertManager rule:

groups:
  - name: webhooks
    rules:
      - alert: WebhookErrorRate
        expr: |
          rate(webhook_requests_total{status="error"}[5m]) /
          rate(webhook_requests_total[5m]) > 0.05
        for: 5m
        annotations:
          summary: "Webhook error rate > 5%"

      - alert: WebhookHighLatency
        expr: |
          histogram_quantile(0.95, rate(webhook_processing_seconds_bucket[5m])) > 5
        for: 10m
        annotations:
          summary: "Webhook p95 latency > 5s"

Troubleshooting

Webhook Not Being Delivered

Check webhook status:

curl https://api.truthvouch.io/v1/webhooks/{webhook_id}/status \
  -H "Authorization: Bearer token"

Common causes:

URL unreachable
Server returns 4xx (permanent failure)
Firewall blocks TruthVouch IP

Signature Verification Fails

Verify locally:

# Get test event from webhook.site or logs
event_body = '{"event_id":"evt-test",...}'
header = 't=1705314600,v1=abcdef...'

# Test verification
from your_app import verify_signature
result = verify_signature(header, event_body, "whsec_...")
print(result)  # Should be True

Event Processing is Slow

Profile your code:

import cProfile
import pstats

profiler = cProfile.Profile()

@app.before_request
def before():
    profiler.enable()

@app.after_request
def after(response):
    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.print_stats()
    return response

See Webhook Events, Signatures, and Retries for more details.