diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d905975..abd823f 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/README.md b/README.md index 7452c63..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. -- 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) 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 = ["."] diff --git a/redis_bus.py b/redis_bus.py index 0f19b34..c2d4d82 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)}) @@ -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: @@ -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() diff --git a/requirements-dev.txt b/requirements-dev.txt index 103cf9e..a9cf48a 100644 --- a/requirements-dev.txt +++ b/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 diff --git a/tests/integration/docker-compose.yml b/tests/integration/docker-compose.yml new file mode 100644 index 0000000..f1d6ae8 --- /dev/null +++ b/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" 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()