9.4Testing Asynchronous Patterns

The Polling Pattern

When testing async systems, use polling with a timeout instead of time.sleep():

import time

def wait_for_condition(check_fn, timeout=30, interval=1):
    """Poll a condition until it's true or timeout."""
    deadline = time.time() + timeout
    last_error = None
    while time.time() < deadline:
        try:
            result = check_fn()
            if result:
                return result
        except Exception as e:
            last_error = e
        time.sleep(interval)
    raise TimeoutError(
        f"Condition not met within {timeout}s. Last error: {last_error}"
    )

# Usage:
def test_async_processing():
    publish_event({"orderId": "ORD-123"})

    # Poll DynamoDB until the order appears
    def check_order_processed():
        item = dynamodb.get_item(TableName="orders", Key={"orderId": {"S": "ORD-123"}})
        return item.get("Item", {}).get("status", {}).get("S") == "processed"

    wait_for_condition(check_order_processed, timeout=30)

The Observer Pattern

For complex event flows, attach a test observer that captures events for later assertion:

class EventObserver:
    """Captures events for testing by subscribing to the event stream."""

    def __init__(self, sqs_client, queue_name="test-observer"):
        self.sqs = sqs_client
        self.queue = sqs_client.create_queue(QueueName=queue_name)
        self.queue_url = self.queue["QueueUrl"]
        self.captured = []

    def drain(self, timeout=10):
        """Collect all messages received within the timeout period."""
        deadline = time.time() + timeout
        while time.time() < deadline:
            msgs = self.sqs.receive_message(
                QueueUrl=self.queue_url,
                WaitTimeSeconds=min(5, int(deadline - time.time())),
                MaxNumberOfMessages=10,
            )
            for msg in msgs.get("Messages", []):
                self.captured.append(json.loads(msg["Body"]))
                self.sqs.delete_message(
                    QueueUrl=self.queue_url,
                    ReceiptHandle=msg["ReceiptHandle"],
                )
        return self.captured

    def assert_received(self, event_type, count=1):
        """Assert that a specific event type was received."""
        matching = [e for e in self.captured if e.get("eventType") == event_type]
        assert len(matching) == count, \
            f"Expected {count} '{event_type}' events, found {len(matching)}"

These patterns work across event systems -- SQS, EventBridge, Kafka, RabbitMQ. The specific API calls change, but the testing strategy remains the same: publish, observe, and verify state.