10.2EventBridge: Event Routing Tests

import json
import boto3
import pytest

class TestEventBridgeIntegration:
    """Test AWS EventBridge event routing and processing."""

    @pytest.fixture
    def eventbridge(self):
        return boto3.client('events', region_name='us-east-1')

    @pytest.fixture
    def sqs_queues(self):
        sqs = boto3.client('sqs', region_name='us-east-1')
        return {
            "billing": sqs.get_queue_url(QueueName='billing-queue')['QueueUrl'],
            "shipping": sqs.get_queue_url(QueueName='shipping-queue')['QueueUrl'],
            "fraud-check": sqs.get_queue_url(QueueName='fraud-check-queue')['QueueUrl'],
        }

    def test_order_event_routes_to_correct_targets(self, eventbridge, sqs_queues):
        """Verify that order events route to billing and shipping queues."""
        event = {
            "Source": "orders.service",
            "DetailType": "OrderPlaced",
            "Detail": json.dumps({
                "order_id": "ord-route-test",
                "total": 150.00,
                "shipping_method": "express"
            })
        }
        eventbridge.put_events(Entries=[event])

        # Check billing queue received the event
        billing_msg = poll_sqs(sqs_queues["billing"], timeout=10)
        assert billing_msg is not None, (
            "Billing queue did not receive OrderPlaced event"
        )
        assert json.loads(billing_msg["Body"])["detail"]["order_id"] == "ord-route-test"

        # Check shipping queue received the event
        shipping_msg = poll_sqs(sqs_queues["shipping"], timeout=10)
        assert shipping_msg is not None, (
            "Shipping queue did not receive OrderPlaced event"
        )

    def test_event_filtering_rules(self, eventbridge, sqs_queues):
        """High-value orders should also route to the fraud-check queue."""
        high_value_event = {
            "Source": "orders.service",
            "DetailType": "OrderPlaced",
            "Detail": json.dumps({
                "order_id": "ord-hv",
                "total": 5000.00
            })
        }
        low_value_event = {
            "Source": "orders.service",
            "DetailType": "OrderPlaced",
            "Detail": json.dumps({
                "order_id": "ord-lv",
                "total": 25.00
            })
        }

        eventbridge.put_events(Entries=[high_value_event, low_value_event])

        # Fraud queue should only get the high-value order
        fraud_msgs = poll_sqs_all(sqs_queues["fraud-check"], timeout=10)
        fraud_order_ids = [
            json.loads(m["Body"])["detail"]["order_id"] for m in fraud_msgs
        ]
        assert "ord-hv" in fraud_order_ids, "High-value order missing from fraud queue"
        assert "ord-lv" not in fraud_order_ids, "Low-value order incorrectly in fraud queue"

    def test_event_schema_validation(self, eventbridge):
        """Events with missing required fields should be rejected or routed to DLQ."""
        invalid_event = {
            "Source": "orders.service",
            "DetailType": "OrderPlaced",
            "Detail": json.dumps({
                # Missing order_id and total
                "shipping_method": "standard"
            })
        }
        response = eventbridge.put_events(Entries=[invalid_event])
        # EventBridge accepts all events but routing rules may filter
        # Check that the event did not reach any processing queue
        # or was routed to a validation-error queue