fixed test suite (extracted infra tests, removed global patching of params, removed mocks in favor of actual implementation; instances are provided by DI); added AsyncExitStack and for lifecycle control instead of manual calls of start/stop

Author: HardMax71

.github/workflows/tests.yml

@@ -139,9 +139,12 @@ jobs:
           MONGODB_PORT: 27017
           # Explicit URL with default credentials
           MONGODB_URL: mongodb://root:[email protected]:27017/?authSource=admin
+          # Optional isolation for Schema Registry subjects (fresh registry in CI, but safe to set)
+          SCHEMA_SUBJECT_PREFIX: "ci.${{ github.run_id }}."
         run: |
           cd backend
           echo "Using BACKEND_BASE_URL=$BACKEND_BASE_URL"
+          echo "Using SCHEMA_SUBJECT_PREFIX=$SCHEMA_SUBJECT_PREFIX"
           echo "MongoDB connection will use default CI credentials"
           python -m pytest tests/integration tests/unit -v --cov=app --cov-branch --cov-report=xml --cov-report=term --cov-report=term-missing
 

backend/.env

@@ -44,7 +44,6 @@ TRACING_SERVICE_VERSION=1.0.0
 TRACING_SAMPLING_RATE=1.0
 
 # Dead Letter Queue Configuration
-KAFKA_DLQ_TOPIC=dead-letter-queue
 DLQ_RETRY_MAX_ATTEMPTS=5
 DLQ_RETRY_BASE_DELAY_SECONDS=60.0
 DLQ_RETRY_MAX_DELAY_SECONDS=3600.0

backend/app/core/dishka_lifespan.py

@@ -1,4 +1,4 @@
-from contextlib import asynccontextmanager
+from contextlib import AsyncExitStack, asynccontextmanager
 from typing import AsyncGenerator
 
 import redis.asyncio as redis
@@ -10,6 +10,7 @@
 from app.core.startup import initialize_metrics_context, initialize_rate_limits
 from app.core.tracing import init_tracing
 from app.db.schema.schema_manager import SchemaManager
+from app.events.event_store_consumer import EventStoreConsumer
 from app.events.schema.schema_registry import SchemaRegistryManager, initialize_event_schemas
 from app.services.sse.kafka_redis_bridge import SSEKafkaRedisBridge
 from app.settings import get_settings
@@ -80,15 +81,14 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 
     # Rate limit middleware added during app creation; service resolved lazily at runtime
 
-    # Start SSE Kafka→Redis bridge to ensure consumers are running before any events are published
-    _ = await container.get(SSEKafkaRedisBridge)
-    logger.info("SSE Kafka→Redis bridge started with consumer pool")
+    # Acquire long-lived services and manage lifecycle via AsyncExitStack
+    sse_bridge = await container.get(SSEKafkaRedisBridge)
+    event_store_consumer = await container.get(EventStoreConsumer)
 
-    # All services initialized by dishka providers
-    logger.info("All services initialized by dishka providers")
-
-    try:
+    async with AsyncExitStack() as stack:
+        await stack.enter_async_context(sse_bridge)
+        logger.info("SSE Kafka→Redis bridge started with consumer pool")
+        await stack.enter_async_context(event_store_consumer)
+        logger.info("EventStoreConsumer started - events will be persisted to MongoDB")
+        logger.info("All services initialized by DI and managed by AsyncExitStack")
         yield
-    finally:
-        # Dishka automatically handles cleanup of all resources!
-        logger.info("Application shutdown complete")

backend/app/core/k8s_clients.py

@@ -0,0 +1,42 @@
+from dataclasses import dataclass
+
+from kubernetes import client as k8s_client
+from kubernetes import config as k8s_config
+
+from app.core.logging import logger
+
+
+@dataclass(frozen=True)
+class K8sClients:
+    api_client: k8s_client.ApiClient
+    v1: k8s_client.CoreV1Api
+    apps_v1: k8s_client.AppsV1Api
+    networking_v1: k8s_client.NetworkingV1Api
+
+
+def create_k8s_clients(kubeconfig_path: str | None = None, in_cluster: bool | None = None) -> K8sClients:
+    if in_cluster:
+        k8s_config.load_incluster_config()
+    elif kubeconfig_path:
+        k8s_config.load_kube_config(config_file=kubeconfig_path)
+    else:
+        k8s_config.load_kube_config()
+
+    configuration = k8s_client.Configuration.get_default_copy()
+    logger.info(f"Kubernetes API host: {configuration.host}")
+    logger.info(f"SSL CA configured: {configuration.ssl_ca_cert is not None}")
+
+    api_client = k8s_client.ApiClient(configuration)
+    return K8sClients(
+        api_client=api_client,
+        v1=k8s_client.CoreV1Api(api_client),
+        apps_v1=k8s_client.AppsV1Api(api_client),
+        networking_v1=k8s_client.NetworkingV1Api(api_client),
+    )
+
+
+def close_k8s_clients(clients: K8sClients) -> None:
+    close = getattr(clients.api_client, "close", None)
+    if callable(close):
+        close()
+

backend/app/core/lifecycle.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from types import TracebackType
+from typing import Optional, Self, Type
+
+
+class LifecycleEnabled:
+    async def start(self) -> None:  # pragma: no cover
+        raise NotImplementedError
+
+    async def stop(self) -> None:  # pragma: no cover
+        raise NotImplementedError
+
+    async def __aenter__(self) -> Self:
+        await self.start()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc: Optional[BaseException],
+        tb: Optional[TracebackType],
+    ) -> None:
+        await self.stop()

backend/app/core/providers.py

@@ -9,6 +9,8 @@
     DatabaseConfig,
     create_database_connection,
 )
+from app.core.k8s_clients import K8sClients, close_k8s_clients, create_k8s_clients
+from app.core.logging import logger
 from app.core.metrics import (
     CoordinatorMetrics,
     DatabaseMetrics,
@@ -126,6 +128,9 @@ async def get_redis_client(self, settings: Settings) -> AsyncIterator[redis.Redi
         )
         # Test connection
         await client.ping()
+        logger.info(
+            f"Redis connected: {settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}"
+        )
         try:
             yield client
         finally:
@@ -221,24 +226,34 @@ async def get_event_store_consumer(
             event_store: EventStore,
             schema_registry: SchemaRegistryManager,
             kafka_producer: UnifiedProducer
-    ) -> AsyncIterator[EventStoreConsumer]:
+    ) -> EventStoreConsumer:
         topics = get_all_topics()
-        consumer = create_event_store_consumer(
+        return create_event_store_consumer(
             event_store=event_store,
             topics=list(topics),
             schema_registry_manager=schema_registry,
             producer=kafka_producer
         )
-        await consumer.start()
+
+    @provide
+    async def get_event_bus_manager(self) -> AsyncIterator[EventBusManager]:
+        manager = EventBusManager()
         try:
-            yield consumer
+            yield manager
         finally:
-            await consumer.stop()
+            await manager.close()
+
+
+class KubernetesProvider(Provider):
+    scope = Scope.APP
 
     @provide
-    def get_event_bus_manager(self) -> EventBusManager:
-        # Don't start the event bus here - let it start lazily when needed
-        return EventBusManager()
+    async def get_k8s_clients(self, settings: Settings) -> AsyncIterator[K8sClients]:
+        clients = create_k8s_clients()
+        try:
+            yield clients
+        finally:
+            close_k8s_clients(clients)
 
 
 class ConnectionProvider(Provider):
@@ -306,18 +321,13 @@ async def get_sse_kafka_redis_bridge(
             settings: Settings,
             event_metrics: EventMetrics,
             sse_redis_bus: SSERedisBus,
-    ) -> AsyncIterator[SSEKafkaRedisBridge]:
-        router = create_sse_kafka_redis_bridge(
+    ) -> SSEKafkaRedisBridge:
+        return create_sse_kafka_redis_bridge(
             schema_registry=schema_registry,
             settings=settings,
             event_metrics=event_metrics,
             sse_bus=sse_redis_bus,
         )
-        await router.start()
-        try:
-            yield router
-        finally:
-            await router.stop()
 
     @provide
     def get_sse_repository(

backend/app/dlq/manager.py

@@ -7,6 +7,7 @@
 from motor.motor_asyncio import AsyncIOMotorCollection, AsyncIOMotorDatabase
 from opentelemetry.trace import SpanKind
 
+from app.core.lifecycle import LifecycleEnabled
 from app.core.logging import logger
 from app.core.metrics.context import get_dlq_metrics
 from app.core.tracing import EventAttributes
@@ -26,7 +27,7 @@
 from app.settings import get_settings
 
 
-class DLQManager:
+class DLQManager(LifecycleEnabled):
     def __init__(
             self,
             database: AsyncIOMotorDatabase,
@@ -69,7 +70,8 @@ async def start(self) -> None:
         if self._running:
             return
 
-        self.consumer.subscribe([self.dlq_topic])
+        topic_name = f"{get_settings().KAFKA_TOPIC_PREFIX}{str(self.dlq_topic)}"
+        self.consumer.subscribe([topic_name])
 
         self._running = True
 
@@ -399,10 +401,11 @@ def create_dlq_manager(
         retry_topic_suffix: str = "-retry",
         default_retry_policy: RetryPolicy | None = None,
 ) -> DLQManager:
+
     settings = get_settings()
     consumer = Consumer({
         'bootstrap.servers': settings.KAFKA_BOOTSTRAP_SERVERS,
-        'group.id': GroupId.DLQ_MANAGER,
+        'group.id': f"{GroupId.DLQ_MANAGER}.{settings.KAFKA_GROUP_SUFFIX}",
         'enable.auto.commit': False,
         'auto.offset.reset': 'earliest',
         'client.id': 'dlq-manager-consumer'

backend/app/events/core/consumer.py

@@ -14,6 +14,7 @@
 from app.domain.enums.kafka import KafkaTopic
 from app.events.schema.schema_registry import SchemaRegistryManager
 from app.infrastructure.kafka.events.base import BaseEvent
+from app.settings import get_settings
 
 from .dispatcher import EventDispatcher
 from .types import ConsumerConfig, ConsumerMetrics, ConsumerState
@@ -37,6 +38,7 @@ def __init__(
         self._event_metrics = get_event_metrics()  # Singleton for Kafka metrics
         self._error_callback: "Callable[[Exception, BaseEvent], Awaitable[None]] | None" = None
         self._consume_task: asyncio.Task[None] | None = None
+        self._topic_prefix = get_settings().KAFKA_TOPIC_PREFIX
 
     async def start(self, topics: list[KafkaTopic]) -> None:
         self._state = (
@@ -49,7 +51,7 @@ async def start(self, topics: list[KafkaTopic]) -> None:
             consumer_config['stats_cb'] = self._handle_stats
 
         self._consumer = Consumer(consumer_config)
-        topic_strings = [str(topic) for topic in topics]
+        topic_strings = [f"{self._topic_prefix}{str(topic)}" for topic in topics]
         self._consumer.subscribe(topic_strings)
         self._running = True
         self._consume_task = asyncio.create_task(self._consume_loop())

backend/app/events/core/producer.py

@@ -7,20 +7,22 @@
 from confluent_kafka import Message, Producer
 from confluent_kafka.error import KafkaError
 
+from app.core.lifecycle import LifecycleEnabled
 from app.core.logging import logger
 from app.core.metrics.context import get_event_metrics
 from app.domain.enums.kafka import KafkaTopic
 from app.events.schema.schema_registry import SchemaRegistryManager
 from app.infrastructure.kafka.events import BaseEvent
 from app.infrastructure.mappers.dlq_mapper import DLQMapper
+from app.settings import get_settings
 
 from .types import ProducerConfig, ProducerMetrics, ProducerState
 
 DeliveryCallback: TypeAlias = Callable[[KafkaError | None, Message], None]
 StatsCallback: TypeAlias = Callable[[dict[str, Any]], None]
 
 
-class UnifiedProducer:
+class UnifiedProducer(LifecycleEnabled):
     def __init__(
             self,
             config: ProducerConfig,
@@ -36,6 +38,8 @@ def __init__(
         self._metrics = ProducerMetrics()
         self._event_metrics = get_event_metrics()  # Singleton for Kafka metrics
         self._poll_task: asyncio.Task | None = None
+        # Topic prefix (for tests/local isolation); cached on init
+        self._topic_prefix = get_settings().KAFKA_TOPIC_PREFIX
 
     @property
     def is_running(self) -> bool:
@@ -190,7 +194,7 @@ async def produce(
         # Serialize value
         serialized_value = self._schema_registry.serialize_event(event_to_produce)
 
-        topic = str(event_to_produce.topic)
+        topic = f"{self._topic_prefix}{str(event_to_produce.topic)}"
         self._producer.produce(
             topic=topic,
             value=serialized_value,
@@ -257,7 +261,7 @@ async def send_to_dlq(
 
             # Send to DLQ topic
             self._producer.produce(
-                topic=str(KafkaTopic.DEAD_LETTER_QUEUE),
+                topic=f"{self._topic_prefix}{str(KafkaTopic.DEAD_LETTER_QUEUE)}",
                 value=serialized_value,
                 key=original_event.event_id.encode() if original_event.event_id else None,
                 headers=[
@@ -269,7 +273,9 @@ async def send_to_dlq(
             )
 
             # Record metrics
-            self._event_metrics.record_kafka_message_produced(str(KafkaTopic.DEAD_LETTER_QUEUE))
+            self._event_metrics.record_kafka_message_produced(
+                f"{self._topic_prefix}{str(KafkaTopic.DEAD_LETTER_QUEUE)}"
+            )
             self._metrics.messages_sent += 1
 
             logger.warning(

backend/app/events/event_store_consumer.py

@@ -2,6 +2,7 @@
 
 from opentelemetry.trace import SpanKind
 
+from app.core.lifecycle import LifecycleEnabled
 from app.core.logging import logger
 from app.core.tracing.utils import trace_span
 from app.domain.enums.events import EventType
@@ -13,7 +14,7 @@
 from app.settings import get_settings
 
 
-class EventStoreConsumer:
+class EventStoreConsumer(LifecycleEnabled):
     """Consumes events from Kafka and stores them in MongoDB."""
 
     def __init__(
@@ -49,7 +50,7 @@ async def start(self) -> None:
         settings = get_settings()
         config = ConsumerConfig(
             bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS,
-            group_id=self.group_id,
+            group_id=f"{self.group_id}.{settings.KAFKA_GROUP_SUFFIX}",
             enable_auto_commit=False,
             max_poll_records=self.batch_size
         )