From ead92cddd338826b62903b332df7c758f114232f Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:07:19 -0400 Subject: [PATCH 01/16] test: add Docker broker integration environment --- tests/integration/docker-compose.yml | 32 ++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 tests/integration/docker-compose.yml diff --git a/tests/integration/docker-compose.yml b/tests/integration/docker-compose.yml new file mode 100644 index 0000000..7a9dbae --- /dev/null +++ b/tests/integration/docker-compose.yml @@ -0,0 +1,32 @@ +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_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" From 78de4991fb41c279e2350ba382bf9739ed60b1fa Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:07:38 -0400 Subject: [PATCH 02/16] test: cover real broker adapter round trips --- tests/integration/test_broker_adapters.py | 119 ++++++++++++++++++++++ 1 file changed, 119 insertions(+) create mode 100644 tests/integration/test_broker_adapters.py diff --git a/tests/integration/test_broker_adapters.py b/tests/integration/test_broker_adapters.py new file mode 100644 index 0000000..d395420 --- /dev/null +++ b/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() From e8bf8697ed22559d2054001a6cee8afe8bf0dd54 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:07:45 -0400 Subject: [PATCH 03/16] test: add broker clients for integration suite --- requirements-dev.txt | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/requirements-dev.txt b/requirements-dev.txt index 103cf9e..90ce9eb 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -9,6 +9,11 @@ pytest pytest-cov pytest-asyncio +# Broker integration testing (development/CI only) +aiokafka>=0.11,<1.0 +aio-pika>=9,<10 +aioredis==2.0.1 + # Type-checking types-requests From 267255b788e32a54cbf95fd7f92397907bd2a3fa Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:07:54 -0400 Subject: [PATCH 04/16] test: isolate Docker integration marker from unit suite --- pyproject.toml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index dbcafe1..913068d 100644 --- a/pyproject.toml +++ b/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 = ["."] From c16144f4efe4f3047f3467c7909b55dd258c9d61 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:08:08 -0400 Subject: [PATCH 05/16] ci: run broker integration tests separately --- .github/workflows/ci.yml | 46 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d905975..2a444f0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,3 +70,49 @@ 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: Wait for broker readiness + run: | + for attempt in {1..45}; do + if docker compose -f tests/integration/docker-compose.yml exec -T redis redis-cli ping | grep -q PONG && \ + docker compose -f tests/integration/docker-compose.yml exec -T rabbitmq rabbitmq-diagnostics -q ping && \ + docker compose -f tests/integration/docker-compose.yml exec -T kafka kafka-topics --bootstrap-server kafka:29092 --list; then + exit 0 + fi + sleep 2 + done + docker compose -f tests/integration/docker-compose.yml logs + exit 1 + + - 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 From f2f7713e815275d569e8a4931e520eb56be2e082 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:08:22 -0400 Subject: [PATCH 06/16] docs: describe broker integration validation boundary --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 7452c63..87232bb 100644 --- a/README.md +++ b/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, and RabbitMQ adapters have a Docker-backed integration suite; they are not claimed as validated until the dedicated CI round trips pass on the reviewed commit. - 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) From f72016ec5b80a1f797e7b00fa7bbbb690484cfb9 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:11:34 -0400 Subject: [PATCH 07/16] ci: probe Kafka through its advertised external listener --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2a444f0..4f85eb1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -102,7 +102,7 @@ jobs: for attempt in {1..45}; do if docker compose -f tests/integration/docker-compose.yml exec -T redis redis-cli ping | grep -q PONG && \ docker compose -f tests/integration/docker-compose.yml exec -T rabbitmq rabbitmq-diagnostics -q ping && \ - docker compose -f tests/integration/docker-compose.yml exec -T kafka kafka-topics --bootstrap-server kafka:29092 --list; then + docker compose -f tests/integration/docker-compose.yml exec -T kafka kafka-topics --bootstrap-server localhost:9092 --list; then exit 0 fi sleep 2 From 8ec5c6d088d19002e4277606a41ac0c457adb20a Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:14:02 -0400 Subject: [PATCH 08/16] ci: check Kafka listener readiness without metadata routing --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4f85eb1..04b0f0b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -102,7 +102,7 @@ jobs: for attempt in {1..45}; do if docker compose -f tests/integration/docker-compose.yml exec -T redis redis-cli ping | grep -q PONG && \ docker compose -f tests/integration/docker-compose.yml exec -T rabbitmq rabbitmq-diagnostics -q ping && \ - docker compose -f tests/integration/docker-compose.yml exec -T kafka kafka-topics --bootstrap-server localhost:9092 --list; then + docker compose -f tests/integration/docker-compose.yml exec -T kafka bash -c 'echo > /dev/tcp/localhost/9092'; then exit 0 fi sleep 2 From 6dbe7775eb3d795fbf7c26b3fd38e7eed1941be9 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:16:28 -0400 Subject: [PATCH 09/16] ci: let adapter clients validate broker readiness --- .github/workflows/ci.yml | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 04b0f0b..1b35813 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -97,18 +97,8 @@ jobs: - name: Start Kafka, Redis, and RabbitMQ run: docker compose -f tests/integration/docker-compose.yml up --detach - - name: Wait for broker readiness - run: | - for attempt in {1..45}; do - if docker compose -f tests/integration/docker-compose.yml exec -T redis redis-cli ping | grep -q PONG && \ - docker compose -f tests/integration/docker-compose.yml exec -T rabbitmq rabbitmq-diagnostics -q ping && \ - docker compose -f tests/integration/docker-compose.yml exec -T kafka bash -c 'echo > /dev/tcp/localhost/9092'; then - exit 0 - fi - sleep 2 - done - docker compose -f tests/integration/docker-compose.yml logs - exit 1 + - name: Allow broker startup + run: sleep 30 - name: Run broker integration tests run: pytest -m integration From d08bf25d9d18458606bb475c2eaaf6c4e185d389 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:19:02 -0400 Subject: [PATCH 10/16] test: speed single-broker Kafka group coordination --- tests/integration/docker-compose.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/integration/docker-compose.yml b/tests/integration/docker-compose.yml index 7a9dbae..f1d6ae8 100644 --- a/tests/integration/docker-compose.yml +++ b/tests/integration/docker-compose.yml @@ -18,6 +18,8 @@ services: 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 From e8f670828d15b2596fca6226b5be6776a72e1e5d Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:19:09 -0400 Subject: [PATCH 11/16] ci: allow single-broker Kafka group coordinator startup --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1b35813..abd823f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -98,7 +98,7 @@ jobs: run: docker compose -f tests/integration/docker-compose.yml up --detach - name: Allow broker startup - run: sleep 30 + run: sleep 45 - name: Run broker integration tests run: pytest -m integration From 398e6f044c007fb214c0a1ab7099b9cb1c4a99bc Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:21:31 -0400 Subject: [PATCH 12/16] docs: record verified broker integration status --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 87232bb..0e0e56a 100644 --- a/README.md +++ b/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. -- Kafka, Redis, and RabbitMQ adapters have a Docker-backed integration suite; they are not claimed as validated until the dedicated CI round trips pass on the reviewed commit. +- Kafka and RabbitMQ passed Docker-backed publish/consume round trips in CI; the Redis adapter is not validated because its `aioredis==2.0.1` import fails on Python 3.11 before a broker connection is made. See the broker-integration PR for the exact error and required decision. - 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) From 29ebba9b080888d8b580580fec9fe936f8b4bd8a Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:24:46 -0400 Subject: [PATCH 13/16] fix(redis): use supported Redis asyncio client --- redis_bus.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/redis_bus.py b/redis_bus.py index 0f19b34..46474e2 100644 --- a/redis_bus.py +++ b/redis_bus.py @@ -1,6 +1,6 @@ import json -import aioredis +import redis.asyncio as redis from .base import EventBus @@ -16,7 +16,7 @@ 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)}) @@ -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() From 89181886614b75afa73823205bd3dcc7c17cf76e Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:24:53 -0400 Subject: [PATCH 14/16] fix(redis): remove incompatible legacy test dependency --- requirements-dev.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/requirements-dev.txt b/requirements-dev.txt index 90ce9eb..a9cf48a 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -12,7 +12,6 @@ pytest-asyncio # Broker integration testing (development/CI only) aiokafka>=0.11,<1.0 aio-pika>=9,<10 -aioredis==2.0.1 # Type-checking types-requests From 050bf0b51b1413b3cb3de3a164f89baf6c4db2fe Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:27:06 -0400 Subject: [PATCH 15/16] fix(redis): use redis asyncio stream blocking argument --- redis_bus.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/redis_bus.py b/redis_bus.py index 46474e2..c2d4d82 100644 --- a/redis_bus.py +++ b/redis_bus.py @@ -24,7 +24,7 @@ async def publish(self, topic: str, message: dict): 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: From b8ba1eedf49e6d1b64154c201d12b1abeb905537 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Wed, 5 Aug 2026 23:29:20 -0400 Subject: [PATCH 16/16] docs: record verified broker adapter validation --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 0e0e56a..9884fec 100644 --- a/README.md +++ b/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. -- Kafka and RabbitMQ passed Docker-backed publish/consume round trips in CI; the Redis adapter is not validated because its `aioredis==2.0.1` import fails on Python 3.11 before a broker connection is made. See the broker-integration PR for the exact error and required decision. +- 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)