9.2Testing EventBridge Rules

EventBridge is AWS's serverless event bus. Testing EventBridge means verifying that rules match the correct event patterns and route to the correct targets.

Pattern Matching Verification

# tests/integration/test_eventbridge.py
import boto3
import json
import pytest

def test_eventbridge_rule_matching():
    """Verify that EventBridge rules match expected event patterns."""
    client = boto3.client("events", endpoint_url="http://localhost:4566")  # LocalStack

    # Put a test event
    response = client.put_events(
        Entries=[{
            "Source": "myapp.orders",
            "DetailType": "OrderCreated",
            "Detail": json.dumps({
                "orderId": "ORD-TEST",
                "amount": 150.00,
                "region": "us-east-1"
            }),
            "EventBusName": "orders-bus"
        }]
    )

    assert response["FailedEntryCount"] == 0

    # Verify the rule exists and has correct pattern
    rule = client.describe_rule(
        Name="high-value-orders",
        EventBusName="orders-bus"
    )
    pattern = json.loads(rule["EventPattern"])
    assert pattern["source"] == ["myapp.orders"]
    assert pattern["detail"]["amount"] == [{"numeric": [">=", 100]}]

def test_event_pattern_does_not_match_low_value_orders():
    """Verify that low-value orders are NOT matched by the high-value rule."""
    # This is a negative test -- equally important
    pattern = {
        "source": ["myapp.orders"],
        "detail": {
            "amount": [{"numeric": [">=", 100]}]
        }
    }

    low_value_event = {
        "source": "myapp.orders",
        "detail": {
            "orderId": "ORD-SMALL",
            "amount": 25.00,
        }
    }

    # AWS provides a test-event-pattern API
    client = boto3.client("events", endpoint_url="http://localhost:4566")
    result = client.test_event_pattern(
        EventPattern=json.dumps(pattern),
        Event=json.dumps(low_value_event)
    )
    assert result["Result"] is False

End-to-End Event Flow Testing

def test_order_event_triggers_notification(localstack_clients):
    """Test the full event flow: OrderCreated -> EventBridge -> SNS -> Email."""
    events = localstack_clients["events"]
    sns = localstack_clients["sns"]
    sqs = localstack_clients["sqs"]

    # Create an SQS queue subscribed to the SNS topic
    # (SQS acts as a test observer for the notification)
    queue = sqs.create_queue(QueueName="test-notifications")
    queue_url = queue["QueueUrl"]
    queue_arn = sqs.get_queue_attributes(
        QueueUrl=queue_url,
        AttributeNames=["QueueArn"]
    )["Attributes"]["QueueArn"]

    # Subscribe the SQS queue to the notification topic
    sns.subscribe(
        TopicArn="arn:aws:sns:us-east-1:000000000000:order-notifications",
        Protocol="sqs",
        Endpoint=queue_arn,
    )

    # Publish the event
    events.put_events(Entries=[{
        "Source": "myapp.orders",
        "DetailType": "OrderCreated",
        "Detail": json.dumps({
            "orderId": "ORD-789",
            "amount": 250.00,
            "customerEmail": "test@example.com"
        }),
        "EventBusName": "orders-bus"
    }])

    # Wait and verify the notification arrived
    import time
    time.sleep(5)

    messages = sqs.receive_message(QueueUrl=queue_url, WaitTimeSeconds=10)
    assert "Messages" in messages
    notification = json.loads(messages["Messages"][0]["Body"])
    assert "ORD-789" in notification["Message"]