9.2Basic Consumer Test: Event Processing

import pytest
from confluent_kafka import Producer, Consumer
import json
import time

class TestOrderEventConsumer:
    """Test the order event consumer that processes Kafka messages."""

    @pytest.fixture
    def kafka_producer(self):
        """Create a test Kafka producer."""
        return Producer({
            'bootstrap.servers': 'localhost:9092',
            'client.id': 'test-producer'
        })

    @pytest.fixture
    def order_consumer(self):
        """Create the order consumer under test."""
        consumer = OrderEventConsumer(
            bootstrap_servers='localhost:9092',
            group_id='test-group',
            topic='order-events'
        )
        yield consumer
        consumer.close()

    def test_order_created_event_processed(self, kafka_producer, order_consumer, db):
        """When an order.created event is published, the consumer stores it."""
        event = {
            "event_type": "order.created",
            "timestamp": "2026-02-09T10:00:00Z",
            "data": {
                "order_id": "ord-123",
                "customer_id": "cust-456",
                "total": 99.99,
                "items": [{"product_id": "prod-789", "quantity": 2}]
            }
        }
        kafka_producer.produce(
            'order-events',
            key="ord-123",
            value=json.dumps(event).encode('utf-8')
        )
        kafka_producer.flush()

        # Wait for consumer to process (with timeout)
        deadline = time.time() + 10
        while time.time() < deadline:
            order_consumer.poll(timeout=1.0)
            order = db.get_order("ord-123")
            if order:
                break
        else:
            pytest.fail("Consumer did not process order.created within 10s")

        assert order.customer_id == "cust-456"
        assert order.total == 99.99
        assert len(order.items) == 1