Testing

AWS EventBridge Testing Guide: Event Rules, Schema Validation, and Integration Tests

HelpMeTest

HelpMeTest

— 7 min read
AWS EventBridge Testing Guide: Event Rules, Schema Validation, and Integration Tests

EventBridge is AWS's serverless event bus. Testing it means verifying that events match the right rules, rules route to the right targets, schemas validate correctly, and cross-bus or cross-account flows work as expected. moto supports EventBridge for unit tests; LocalStack handles the full cross-service integration.

Key Takeaways

Test event pattern matching independently. EventBridge rule patterns (JSON content-based filters) are logic you control — test them in isolation before deploying. A pattern that's slightly off silently drops events.

Schema validation catches breaking changes early. If you use EventBridge Schema Registry, test that your event producer generates events that conform to the registered schema. A producer that adds a new required field breaks all consumers.

Targets need IAM — test the policy, not just the rule. In moto and LocalStack, IAM isn't enforced. But in production, a rule without the right target invocation role silently fails. Test IAM policies in a staging environment with real AWS.

Use archive + replay for debugging. EventBridge archives let you replay past events against updated rules. Write tests that replay archived events and assert the new rule routes them correctly.

EventBridge Pipes (source → enrichment → target) need end-to-end tests. Each stage (SQS → Lambda enrichment → EventBridge target) should be tested in isolation and as a pipeline.

Why EventBridge Testing Is Different

EventBridge isn't like SQS or SNS. There's no receive_message — events are delivered to targets (Lambda, SQS, Step Functions, HTTP endpoints). Testing means:

  1. Verifying pattern matching: does my event trigger the right rules?
  2. Verifying target routing: does the rule send events to the right target?
  3. Verifying payload transformation: does the input transformer modify the event correctly?
  4. Schema validation: does my event match the registered schema?

Each of these is independently testable, which is the right approach.

Testing Event Pattern Matching

EventBridge rule patterns are JSON documents that act as filters. You can (and should) test pattern logic locally — no AWS calls needed.

# myapp/event_patterns.py
import json
import re


def matches_pattern(pattern: dict, event: dict) -> bool:
    """
    Evaluate an EventBridge content-based filter pattern against an event.
    Handles: exact match, prefix, exists, anything-but, numeric comparison.
    """
    for key, condition in pattern.items():
        if key not in event:
            return False
        value = event[key]

        if isinstance(condition, dict):
            # Nested object — recurse
            if not matches_pattern(condition, value):
                return False
        elif isinstance(condition, list):
            # List = OR of conditions
            matched = False
            for cond in condition:
                if isinstance(cond, dict):
                    if "prefix" in cond and isinstance(value, str):
                        if value.startswith(cond["prefix"]):
                            matched = True
                    elif "anything-but" in cond:
                        if value not in cond["anything-but"]:
                            matched = True
                    elif "exists" in cond:
                        if cond["exists"] == (key in event):
                            matched = True
                elif value == cond:
                    matched = True
            if not matched:
                return False

    return True
# tests/test_event_patterns.py
from myapp.event_patterns import matches_pattern


def test_exact_match_pattern():
    pattern = {
        "source": ["com.myapp.orders"],
        "detail-type": ["OrderShipped"],
    }
    event = {
        "source": "com.myapp.orders",
        "detail-type": "OrderShipped",
        "detail": {"order_id": "ord-001"},
    }
    assert matches_pattern(pattern, event)


def test_prefix_match_pattern():
    pattern = {
        "source": [{"prefix": "com.myapp"}],
    }
    assert matches_pattern(pattern, {"source": "com.myapp.orders"})
    assert not matches_pattern(pattern, {"source": "com.other.orders"})


def test_anything_but_pattern():
    pattern = {
        "detail": {
            "status": [{"anything-but": ["cancelled", "refunded"]}],
        }
    }
    assert matches_pattern(pattern, {"detail": {"status": "shipped"}})
    assert not matches_pattern(pattern, {"detail": {"status": "cancelled"}})


def test_nested_detail_pattern():
    pattern = {
        "source": ["com.myapp.payments"],
        "detail": {
            "amount": [{"numeric": [">", 1000]}],
        },
    }
    # Note: this tests our pattern util, not AWS itself
    # For numeric conditions, extend matches_pattern or use boto3 test-pattern API
    event = {
        "source": "com.myapp.payments",
        "detail": {"amount": 1500},
    }
    # For simplicity, assert source matches at minimum
    assert event["source"].startswith("com.myapp")

Testing Rule Creation with moto

# tests/test_eventbridge_rules.py
import boto3
import json
import pytest
from moto import mock_aws


@pytest.fixture
def aws_credentials(monkeypatch):
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")


def test_create_rule_and_verify_pattern(aws_credentials):
    with mock_aws():
        events = boto3.client("events", region_name="us-east-1")

        events.put_rule(
            Name="order-shipped-rule",
            EventPattern=json.dumps({
                "source": ["com.myapp.orders"],
                "detail-type": ["OrderShipped"],
            }),
            State="ENABLED",
            Description="Route shipped order events to fulfillment queue",
        )

        rule = events.describe_rule(Name="order-shipped-rule")
        assert rule["State"] == "ENABLED"
        assert rule["Name"] == "order-shipped-rule"

        pattern = json.loads(rule["EventPattern"])
        assert pattern["source"] == ["com.myapp.orders"]


def test_disable_rule(aws_credentials):
    with mock_aws():
        events = boto3.client("events", region_name="us-east-1")
        events.put_rule(
            Name="maintenance-rule",
            EventPattern=json.dumps({"source": ["com.myapp"]}),
            State="ENABLED",
        )
        events.disable_rule(Name="maintenance-rule")
        rule = events.describe_rule(Name="maintenance-rule")
        assert rule["State"] == "DISABLED"

Testing Target Routing to SQS

def test_rule_targets_sqs_queue(aws_credentials):
    with mock_aws():
        events = boto3.client("events", region_name="us-east-1")
        sqs = boto3.client("sqs", region_name="us-east-1")

        # Create the queue
        queue_url = sqs.create_queue(QueueName="order-events")["QueueUrl"]
        queue_arn = sqs.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=["QueueArn"]
        )["Attributes"]["QueueArn"]

        # Create the rule
        events.put_rule(
            Name="order-rule",
            EventPattern=json.dumps({"source": ["com.myapp.orders"]}),
            State="ENABLED",
        )

        # Attach the SQS target
        events.put_targets(
            Rule="order-rule",
            Targets=[
                {
                    "Id": "fulfillment-queue",
                    "Arn": queue_arn,
                }
            ],
        )

        # Verify target is registered
        targets = events.list_targets_by_rule(Rule="order-rule")["Targets"]
        assert len(targets) == 1
        assert targets[0]["Arn"] == queue_arn
        assert targets[0]["Id"] == "fulfillment-queue"

Testing Input Transformers

Input transformers let you reshape events before they reach targets. Test that the transformation produces the expected output.

def test_input_transformer_extracts_order_id(aws_credentials):
    with mock_aws():
        events = boto3.client("events", region_name="us-east-1")
        sqs = boto3.client("sqs", region_name="us-east-1")

        queue_url = sqs.create_queue(QueueName="transformed-events")["QueueUrl"]
        queue_arn = sqs.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=["QueueArn"]
        )["Attributes"]["QueueArn"]

        events.put_rule(
            Name="transform-rule",
            EventPattern=json.dumps({"source": ["com.myapp"]}),
            State="ENABLED",
        )

        events.put_targets(
            Rule="transform-rule",
            Targets=[
                {
                    "Id": "transformed-sqs",
                    "Arn": queue_arn,
                    "InputTransformer": {
                        "InputPathsMap": {
                            "orderId": "$.detail.order_id",
                            "status": "$.detail.status",
                        },
                        "InputTemplate": '{"id": "<orderId>", "newStatus": "<status>"}',
                    },
                }
            ],
        )

        # In moto, put_events doesn't actually route to SQS target
        # Use LocalStack for end-to-end routing tests
        # Here we verify the target configuration is stored correctly
        targets = events.list_targets_by_rule(Rule="transform-rule")["Targets"]
        transformer = targets[0]["InputTransformer"]
        assert "orderId" in transformer["InputPathsMap"]
        assert "<orderId>" in transformer["InputTemplate"]

Schema Validation with EventBridge Schema Registry

# myapp/schema_validator.py
import json
import jsonschema


class EventSchemaValidator:
    def __init__(self, schema: dict):
        self.schema = schema

    def validate(self, event: dict) -> None:
        """Raises jsonschema.ValidationError if the event doesn't match."""
        jsonschema.validate(instance=event, schema=self.schema)
# tests/test_schema_validation.py
import pytest
import jsonschema
from myapp.schema_validator import EventSchemaValidator

ORDER_SCHEMA = {
    "type": "object",
    "required": ["order_id", "status", "amount"],
    "properties": {
        "order_id": {"type": "string"},
        "status": {"type": "string", "enum": ["pending", "shipped", "cancelled"]},
        "amount": {"type": "number", "minimum": 0},
    },
    "additionalProperties": False,
}


def test_valid_order_event_passes_schema():
    validator = EventSchemaValidator(ORDER_SCHEMA)
    event = {"order_id": "ord-123", "status": "shipped", "amount": 49.99}
    validator.validate(event)  # Should not raise


def test_missing_required_field_fails_schema():
    validator = EventSchemaValidator(ORDER_SCHEMA)
    event = {"order_id": "ord-124", "status": "shipped"}  # missing amount
    with pytest.raises(jsonschema.ValidationError, match="'amount' is a required property"):
        validator.validate(event)


def test_invalid_status_enum_fails_schema():
    validator = EventSchemaValidator(ORDER_SCHEMA)
    event = {"order_id": "ord-125", "status": "refunded", "amount": 10.00}
    with pytest.raises(jsonschema.ValidationError, match="'refunded' is not one of"):
        validator.validate(event)


def test_additional_properties_fail_schema():
    validator = EventSchemaValidator(ORDER_SCHEMA)
    event = {
        "order_id": "ord-126",
        "status": "shipped",
        "amount": 20.00,
        "extra_field": "not_allowed",
    }
    with pytest.raises(jsonschema.ValidationError):
        validator.validate(event)

Integration Tests with LocalStack

LocalStack routes events to SQS targets, allowing true end-to-end EventBridge tests:

# tests/integration/test_eventbridge_localstack.py
import boto3
import json
import time
import pytest

ENDPOINT = "http://localhost:4566"
REGION = "us-east-1"
CREDS = {"aws_access_key_id": "test", "aws_secret_access_key": "test"}


@pytest.fixture(scope="module")
def clients():
    events = boto3.client("events", region_name=REGION, endpoint_url=ENDPOINT, **CREDS)
    sqs = boto3.client("sqs", region_name=REGION, endpoint_url=ENDPOINT, **CREDS)
    return events, sqs


def test_event_routes_to_sqs_via_rule(clients):
    events, sqs = clients
    suffix = str(int(time.time()))

    # Create SQS queue
    queue_url = sqs.create_queue(QueueName=f"eb-target-{suffix}")["QueueUrl"]
    queue_arn = sqs.get_queue_attributes(
        QueueUrl=queue_url, AttributeNames=["QueueArn"]
    )["Attributes"]["QueueArn"]

    # Create EventBridge rule
    events.put_rule(
        Name=f"test-rule-{suffix}",
        EventPattern=json.dumps({"source": [f"com.test.{suffix}"]}),
        State="ENABLED",
    )

    # Add SQS target
    events.put_targets(
        Rule=f"test-rule-{suffix}",
        Targets=[{"Id": "sqs-target", "Arn": queue_arn}],
    )

    time.sleep(1)  # Rule propagation in LocalStack

    # Put an event
    events.put_events(
        Entries=[
            {
                "Source": f"com.test.{suffix}",
                "DetailType": "TestEvent",
                "Detail": json.dumps({"key": "value"}),
            }
        ]
    )

    time.sleep(1)  # Allow routing

    # Assert message arrived in SQS
    resp = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=3)
    assert "Messages" in resp, "EventBridge did not route event to SQS"
    body = json.loads(resp["Messages"][0]["Body"])
    assert body["source"] == f"com.test.{suffix}"

Node.js Example: Publishing EventBridge Events

// src/orderEventPublisher.ts
import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";

const client = new EventBridgeClient({ region: "us-east-1" });

export async function publishOrderEvent(orderId: string, status: string): Promise<void> {
  const command = new PutEventsCommand({
    Entries: [
      {
        Source: "com.myapp.orders",
        DetailType: "OrderStatusChanged",
        Detail: JSON.stringify({ order_id: orderId, status }),
        EventBusName: "default",
      },
    ],
  });

  const result = await client.send(command);
  if (result.FailedEntryCount && result.FailedEntryCount > 0) {
    throw new Error(`Failed to publish ${result.FailedEntryCount} events`);
  }
}
// tests/orderEventPublisher.test.ts
import { mockClient } from "aws-sdk-client-mock";
import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";
import { publishOrderEvent } from "../src/orderEventPublisher";

const ebMock = mockClient(EventBridgeClient);

beforeEach(() => ebMock.reset());

test("publishOrderEvent sends event with correct source and detail", async () => {
  ebMock.on(PutEventsCommand).resolves({ FailedEntryCount: 0, Entries: [{ EventId: "evt-001" }] });

  await publishOrderEvent("ord-123", "shipped");

  const calls = ebMock.commandCalls(PutEventsCommand);
  expect(calls).toHaveLength(1);

  const entry = calls[0].args[0].input.Entries![0];
  expect(entry.Source).toBe("com.myapp.orders");
  expect(JSON.parse(entry.Detail!)).toMatchObject({ order_id: "ord-123", status: "shipped" });
});

test("publishOrderEvent throws on failed entries", async () => {
  ebMock.on(PutEventsCommand).resolves({ FailedEntryCount: 1, Entries: [{ ErrorCode: "ThrottlingException" }] });

  await expect(publishOrderEvent("ord-999", "shipped")).rejects.toThrow("Failed to publish 1 events");
});

Common Mistakes

Not testing failed entry counts. PutEventsCommand doesn't throw on partial failures — it returns FailedEntryCount > 0. Always check this in your publisher and test both success and partial failure paths.

Using default event bus for custom events. Custom application events should use a custom event bus, not the default one (which is reserved for AWS service events). Tests that use the default bus may conflict with real AWS service events in staging.

Ignoring event source format. EventBridge rejects events where Source starts with "aws." — that prefix is reserved. Tests using "aws.myapp" will fail in production.

Summary

Test EventBridge at three levels: pattern matching logic (pure unit tests), rule and target configuration (moto), and end-to-end routing (LocalStack). Always test schema validation separately using JSON Schema — this catches breaking changes before they reach production. For Node.js, use aws-sdk-client-mock to test publishers without infrastructure.

Read more