test: validate broker adapters against Docker services

.github/workflows/ci.yml

@@ -70,3 +70,39 @@ jobs:
             echo "- Checks: black, ruff, mypy, pytest"
             echo "- Benchmark artifact: benchmark-results.json"
           } >> "$GITHUB_STEP_SUMMARY"
+
+
+  broker-integration:
+    name: Broker adapter integration
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+          cache-dependency-path: |
+            requirements.txt
+            requirements-dev.txt
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install -r requirements.txt -r requirements-dev.txt
+
+      - name: Start Kafka, Redis, and RabbitMQ
+        run: docker compose -f tests/integration/docker-compose.yml up --detach
+
+      - name: Allow broker startup
+        run: sleep 45
+
+      - name: Run broker integration tests
+        run: pytest -m integration
+
+      - name: Stop broker services
+        if: always()
+        run: docker compose -f tests/integration/docker-compose.yml down --volumes

README.md

@@ -241,7 +241,7 @@ Known remaining gaps for a full production release:
 
 - Coverage is 54%; next priority is adding API router, consumer, broker adapter, and service integration tests.
 - `docker-compose.yml` still references service directories that are architectural placeholders.
-- Broker adapters require live Kafka, Redis, or RabbitMQ integration environments for end-to-end validation.
+- Kafka, Redis Streams, and RabbitMQ adapters pass Docker-backed publish/consume round trips in the dedicated CI job. The suite is isolated from the Docker-free unit-test path.
 - Kubernetes manifests should be parameterized with real image names and deployment environments.
 - Authentication, authorization, secrets management, and PII controls need implementation before production use.
 # [![CI](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/ci.yml/badge.svg?branch=docs%2Fportfolio-readme-production-scalable-event-driven-ride-sharing-platform)](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/ci.yml) [![Hygiene](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/hygiene-matrix.yml/badge.svg?branch=docs%2Fportfolio-readme-production-scalable-event-driven-ride-sharing-platform)](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/hygiene-matrix.yml)

pyproject.toml

@@ -5,7 +5,10 @@ line-length = 100
 pythonpath = ["."]
 testpaths = [".", "tests"]
 python_files = ["test_*.py"]
-addopts = "-ra"
+addopts = "-ra -m 'not integration'"
+markers = [
+    "integration: requires Docker-hosted Kafka, Redis, and RabbitMQ services",
+]
 
 [tool.coverage.run]
 source = ["."]

redis_bus.py

@@ -1,6 +1,6 @@
 import json
 
-import aioredis
+import redis.asyncio as redis
 
 from .base import EventBus
 
@@ -16,15 +16,15 @@ def __init__(self, redis_url="redis://localhost:6379"):
         self.redis = None
 
     async def connect(self):
-        self.redis = await aioredis.from_url(self.redis_url, decode_responses=True)
+        self.redis = redis.from_url(self.redis_url, decode_responses=True)
 
     async def publish(self, topic: str, message: dict):
         await self.redis.xadd(topic, {"data": json.dumps(message)})
 
     async def subscribe(self, topic: str, handler):
         last_id = "$"
         while True:
-            streams = await self.redis.xread({topic: last_id}, timeout=5000)
+            streams = await self.redis.xread({topic: last_id}, block=5000)
             if streams:
                 _, messages = streams[0]
                 for msg_id, fields in messages:
@@ -34,4 +34,4 @@ async def subscribe(self, topic: str, handler):
 
     async def close(self):
         if self.redis:
-            await self.redis.close()
+            await self.redis.aclose()

requirements-dev.txt

@@ -9,6 +9,10 @@ pytest
 pytest-cov
 pytest-asyncio
 
+# Broker integration testing (development/CI only)
+aiokafka>=0.11,<1.0
+aio-pika>=9,<10
+
 # Type-checking
 types-requests
 

tests/integration/docker-compose.yml

@@ -0,0 +1,34 @@
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.6.1
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  kafka:
+    image: confluentinc/cp-kafka:7.6.1
+    depends_on:
+      - zookeeper
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:29092
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+
+  redis:
+    image: redis:7.4-alpine
+    ports:
+      - "6379:6379"
+
+  rabbitmq:
+    image: rabbitmq:3.13-alpine
+    ports:
+      - "5672:5672"

tests/integration/test_broker_adapters.py

@@ -0,0 +1,119 @@
+"""Round-trip integration tests against Docker-hosted broker implementations."""
+
+import asyncio
+import importlib.util
+import sys
+import types
+import uuid
+from contextlib import suppress
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+BASE_ADAPTER = ROOT / "src" / "event-bus" / "base.py"
+
+
+def load_adapter_module(module_name: str):
+    """Load root adapters beneath a temporary package so their relative base import resolves."""
+    package_name = f"_integration_adapter_{uuid.uuid4().hex}"
+    package = types.ModuleType(package_name)
+    package.__path__ = [str(ROOT)]
+    sys.modules[package_name] = package
+
+    base_spec = importlib.util.spec_from_file_location(f"{package_name}.base", BASE_ADAPTER)
+    base_module = importlib.util.module_from_spec(base_spec)
+    sys.modules[base_spec.name] = base_module
+    base_spec.loader.exec_module(base_module)
+
+    adapter_spec = importlib.util.spec_from_file_location(
+        f"{package_name}.{module_name}", ROOT / f"{module_name}.py"
+    )
+    adapter_module = importlib.util.module_from_spec(adapter_spec)
+    sys.modules[adapter_spec.name] = adapter_module
+    adapter_spec.loader.exec_module(adapter_module)
+    return adapter_module
+
+
+async def stop_subscription(task: asyncio.Task) -> None:
+    task.cancel()
+    with suppress(asyncio.CancelledError):
+        await task
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_kafka_adapter_round_trip():
+    adapter = load_adapter_module("kafka_bus")
+    topic = f"ride-location-{uuid.uuid4().hex}"
+    payload = {"driver_id": "driver-1", "status": "available"}
+    received = asyncio.Event()
+    messages = []
+
+    async def handler(message):
+        messages.append(message)
+        received.set()
+
+    bus = adapter.KafkaEventBus(bootstrap_servers="localhost:9092", group_id=f"test-{uuid.uuid4()}")
+    await bus.connect()
+    subscription = asyncio.create_task(bus.subscribe(topic, handler))
+    try:
+        await asyncio.sleep(1)
+        await bus.publish(topic, payload)
+        await asyncio.wait_for(received.wait(), timeout=30)
+        assert messages == [payload]
+    finally:
+        await stop_subscription(subscription)
+        await bus.close()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_redis_adapter_round_trip():
+    adapter = load_adapter_module("redis_bus")
+    topic = f"ride-location-{uuid.uuid4().hex}"
+    payload = {"driver_id": "driver-2", "status": "en_route"}
+    received = asyncio.Event()
+    messages = []
+
+    async def handler(message):
+        messages.append(message)
+        received.set()
+
+    bus = adapter.RedisEventBus(redis_url="redis://localhost:6379/15")
+    await bus.connect()
+    subscription = asyncio.create_task(bus.subscribe(topic, handler))
+    try:
+        await asyncio.sleep(1)
+        await bus.publish(topic, payload)
+        await asyncio.wait_for(received.wait(), timeout=30)
+        assert messages == [payload]
+    finally:
+        await stop_subscription(subscription)
+        await bus.close()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_rabbitmq_adapter_round_trip():
+    adapter = load_adapter_module("rabbitmq_bus")
+    topic = f"ride-location-{uuid.uuid4().hex}"
+    payload = {"driver_id": "driver-3", "status": "on_trip"}
+    received = asyncio.Event()
+    messages = []
+
+    async def handler(message):
+        messages.append(message)
+        received.set()
+
+    bus = adapter.RabbitMQEventBus(url="amqp://guest:guest@localhost/")
+    await bus.connect()
+    subscription = asyncio.create_task(bus.subscribe(topic, handler))
+    try:
+        await asyncio.sleep(1)
+        await bus.publish(topic, payload)
+        await asyncio.wait_for(received.wait(), timeout=30)
+        assert messages == [payload]
+    finally:
+        await stop_subscription(subscription)
+        await bus.close()