diff --git a/src/omnibase_infra/runtime/__init__.py b/src/omnibase_infra/runtime/__init__.py index a2b7a5d05e..12d89066a4 100644 --- a/src/omnibase_infra/runtime/__init__.py +++ b/src/omnibase_infra/runtime/__init__.py @@ -243,6 +243,9 @@ load_event_bus_subcontract, ) +# Request-response wiring (OMN-1742) +from omnibase_infra.runtime.request_response_wiring import RequestResponseWiring + # Runtime contract config loader (OMN-1519) from omnibase_infra.runtime.runtime_contract_config_loader import ( RuntimeContractConfigLoader, @@ -429,6 +432,8 @@ # Event bus subcontract wiring (OMN-1621) "EventBusSubcontractWiring", "load_event_bus_subcontract", + # Request-response wiring (OMN-1742) + "RequestResponseWiring", # Runtime contract config loader (OMN-1519) "RuntimeContractConfigLoader", # Security constants and configuration (OMN-1519) diff --git a/src/omnibase_infra/runtime/request_response_wiring.py b/src/omnibase_infra/runtime/request_response_wiring.py new file mode 100644 index 0000000000..fe91a0cf3a --- /dev/null +++ b/src/omnibase_infra/runtime/request_response_wiring.py @@ -0,0 +1,785 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2025 OmniNode Team +"""Request-response wiring for correlation-based RPC-style Kafka communication. + +This module provides the RequestResponseWiring class for implementing request-response +patterns over Kafka. Unlike the standard EventBusSubcontractWiring (designed for 24/7 +consumers), this wiring supports correlation-based request-response flows where a +publisher sends a request and awaits a correlated response. + +Architecture: + The RequestResponseWiring class is responsible for: + 1. Reading ModelRequestResponseConfig from contracts + 2. Creating dedicated consumers for reply topics (completed + failed) + 3. Managing correlation ID tracking with in-flight futures + 4. Injecting correlation IDs if not present in outgoing requests + 5. Matching incoming responses to pending requests via correlation ID + 6. Handling timeouts with InfraTimeoutError + 7. Circuit breaker protection for publish failures + + This follows ARCH-002: "Runtime owns all Kafka plumbing." Nodes and handlers + declare request-response requirements in contracts but never directly interact + with Kafka consumers or producers. + +Boot Nonce: + A per-process boot nonce (8-character hex string from UUID4) is generated once + at module load time. This ensures consumer groups are unique per process instance, + preventing message stealing between concurrent processes. + +Consumer Group Naming: + Consumer groups are named as: {environment}.rr.{instance_name}.{boot_nonce} + Example: "dev.rr.code-analysis.a1b2c3d4" + + This ensures: + - Each process instance has its own consumer group + - Multiple instances don't steal each other's responses + - Process restarts get new consumer groups + +Correlation ID Handling: + When sending requests, the wiring: + 1. Checks if correlation_id exists at the configured location (default: body.correlation_id) + 2. If missing, injects a new UUID4 correlation_id into the payload + 3. Returns the correlation_id in the response for tracing + +Error Handling: + - Timeout: Raises InfraTimeoutError (NOT InfraUnavailableError) + - Circuit breaker open: Raises InfraUnavailableError + - Publish failures: Recorded by circuit breaker, wrapped in appropriate error + +Related: + - OMN-1742: Request-response wiring for Kafka RPC patterns + - ModelRequestResponseConfig: Contract model for request-response configuration + - EventBusSubcontractWiring: Standard 24/7 consumer wiring (different pattern) + +.. versionadded:: 0.3.1 +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from dataclasses import dataclass, field +from typing import TYPE_CHECKING +from uuid import UUID, uuid4 + +from aiokafka import AIOKafkaConsumer + +from omnibase_core.models.contracts.subcontracts import ( + ModelCorrelationConfig, + ModelRequestResponseConfig, + ModelRequestResponseInstance, +) +from omnibase_core.protocols.event_bus.protocol_event_bus_publisher import ( + ProtocolEventBusPublisher, +) +from omnibase_infra.enums import EnumInfraTransportType +from omnibase_infra.errors import ( + InfraTimeoutError, + InfraUnavailableError, + ModelInfraErrorContext, + ModelTimeoutErrorContext, + ProtocolConfigurationError, +) +from omnibase_infra.mixins import MixinAsyncCircuitBreaker + +if TYPE_CHECKING: + from aiokafka import ConsumerRecord + +logger = logging.getLogger(__name__) + +# Boot nonce: Generated ONCE per process at module load time. +# Used to create unique consumer groups per process instance. +_BOOT_NONCE: str = uuid4().hex[:8] + + +@dataclass +class RequestResponseInstanceState: + """Internal state for a single request-response instance. + + Tracks pending requests, consumer task, and consumer instance for + a configured request-response pattern. + """ + + name: str + request_topic: str + completed_topic: str + failed_topic: str + timeout_seconds: int + correlation_config: ModelCorrelationConfig + consumer_group: str + pending: dict[str, asyncio.Future[dict[str, object]]] = field(default_factory=dict) + consumer: AIOKafkaConsumer | None = None + consumer_task: asyncio.Task[None] | None = None + + +class RequestResponseWiring(MixinAsyncCircuitBreaker): + """Wires request-response patterns to Kafka for correlation-based RPC. + + This class implements the request-response pattern over Kafka, where: + 1. A request is published to a request topic + 2. The wiring awaits a correlated response on reply topics + 3. Correlation is tracked via correlation_id in the message payload + + Unlike EventBusSubcontractWiring (designed for 24/7 consumers), this wiring + creates ephemeral consumers that match responses to pending requests. + + Consumer Startup: + Consumers are started eagerly when wire_request_response() is called. + This ensures responses can be received immediately after the first request. + + Correlation ID Injection: + If the outgoing payload lacks a correlation_id at the configured location, + the wiring injects a new UUID4. The correlation_id is always returned + in the response for tracing. + + Timeout Handling: + If no response is received within the configured timeout (default: 30s), + InfraTimeoutError is raised. Note: This is a timeout error, NOT + InfraUnavailableError which is reserved for circuit breaker states. + + Circuit Breaker: + Publish failures are tracked by the circuit breaker. When the circuit + opens, InfraUnavailableError is raised immediately without attempting + to publish. + + Thread Safety: + This class is designed for single-threaded async use. All operations + should be performed from a single async context. + + Example: + ```python + from omnibase_infra.runtime import RequestResponseWiring + from omnibase_core.models.contracts.subcontracts import ( + ModelRequestResponseConfig, + ModelRequestResponseInstance, + ModelReplyTopics, + ) + + # Create wiring + wiring = RequestResponseWiring( + event_bus=event_bus, + environment="dev", + app_name="my-service", + ) + + # Wire from config + config = ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="code-analysis", + request_topic="onex.cmd.intelligence.analyze-code.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.intelligence.code-analyzed.v1", + failed="onex.evt.intelligence.code-analysis-failed.v1", + ), + timeout_seconds=30, + ) + ] + ) + await wiring.wire_request_response(config) + + # Send request and await response + response = await wiring.send_request( + instance_name="code-analysis", + payload={"code": "print('hello')"}, + ) + + # Cleanup on shutdown + await wiring.cleanup() + ``` + + Attributes: + _event_bus: Event bus for publishing requests + _environment: Environment prefix for topics (e.g., 'dev', 'prod') + _app_name: Application name for consumer group identification + _instances: Dict mapping instance names to their state + _bootstrap_servers: Kafka bootstrap servers from event bus + + .. versionadded:: 0.3.1 + """ + + def __init__( + self, + event_bus: ProtocolEventBusPublisher, + environment: str, + app_name: str, + bootstrap_servers: str | None = None, + ) -> None: + """Initialize request-response wiring. + + Args: + event_bus: Event bus for publishing requests. Must implement + ProtocolEventBusPublisher interface. + environment: Environment prefix for topics (e.g., 'dev', 'prod'). + Used to resolve topic suffixes to full topic names. + app_name: Application name for logging and consumer group naming. + bootstrap_servers: Kafka bootstrap servers. If not provided, attempts + to read from event_bus._bootstrap_servers or environment variable. + + Raises: + ValueError: If environment is empty or whitespace-only. + ProtocolConfigurationError: If bootstrap_servers cannot be determined. + """ + if not environment or not environment.strip(): + raise ValueError("environment must be a non-empty string") + if not app_name or not app_name.strip(): + raise ValueError("app_name must be a non-empty string") + + self._event_bus = event_bus + self._environment = environment + self._app_name = app_name + self._instances: dict[str, RequestResponseInstanceState] = {} + self._logger = logging.getLogger(__name__) + + # Resolve bootstrap servers + if bootstrap_servers: + self._bootstrap_servers = bootstrap_servers + elif hasattr(event_bus, "_bootstrap_servers"): + self._bootstrap_servers = event_bus._bootstrap_servers # type: ignore[union-attr] + else: + import os + + self._bootstrap_servers = os.environ.get( + "KAFKA_BOOTSTRAP_SERVERS", "localhost:9092" + ) + + # Initialize circuit breaker for publish protection + self._init_circuit_breaker( + threshold=5, + reset_timeout=60.0, + service_name=f"request-response.{app_name}", + transport_type=EnumInfraTransportType.KAFKA, + half_open_successes=1, + ) + + self._logger.debug( + "RequestResponseWiring initialized: environment=%s, app_name=%s, " + "boot_nonce=%s, bootstrap_servers=%s", + environment, + app_name, + _BOOT_NONCE, + self._bootstrap_servers, + ) + + def resolve_topic(self, topic_suffix: str) -> str: + """Resolve topic suffix to full topic name with environment prefix. + + Args: + topic_suffix: ONEX format topic suffix + (e.g., 'onex.cmd.intelligence.analyze-code.v1') + + Returns: + Full topic name with environment prefix + (e.g., 'dev.onex.cmd.intelligence.analyze-code.v1') + """ + return f"{self._environment}.{topic_suffix}" + + async def wire_request_response( + self, + config: ModelRequestResponseConfig, + ) -> None: + """Wire request-response instances from configuration. + + Creates consumers for each instance's reply topics and starts them + eagerly. Consumers run in background tasks, matching incoming responses + to pending requests via correlation ID. + + Consumer Group Naming: + Consumer groups are named as: {environment}.rr.{instance_name}.{boot_nonce} + The boot_nonce ensures each process instance has unique consumer groups. + + Args: + config: Request-response configuration with instance definitions. + + Raises: + ProtocolConfigurationError: If instance name conflicts with existing. + InfraConnectionError: If Kafka connection fails during consumer start. + """ + for instance in config.instances: + await self._wire_instance(instance) + + async def _wire_instance(self, instance: ModelRequestResponseInstance) -> None: + """Wire a single request-response instance. + + Args: + instance: Instance configuration to wire. + + Raises: + ProtocolConfigurationError: If instance name already wired. + """ + if instance.name in self._instances: + raise ProtocolConfigurationError( + f"Request-response instance '{instance.name}' already wired", + context=ModelInfraErrorContext.with_correlation( + transport_type=EnumInfraTransportType.KAFKA, + operation="wire_request_response", + ), + instance_name=instance.name, + ) + + # Build consumer group: {environment}.rr.{instance_name}.{boot_nonce} + consumer_group = f"{self._environment}.rr.{instance.name}.{_BOOT_NONCE}" + + # Resolve topics with environment prefix + request_topic = self.resolve_topic(instance.request_topic) + completed_topic = self.resolve_topic(instance.reply_topics.completed) + failed_topic = self.resolve_topic(instance.reply_topics.failed) + + # Default correlation config if not specified + correlation_config = instance.correlation or ModelCorrelationConfig() + + # Create instance state + rr_instance = RequestResponseInstanceState( + name=instance.name, + request_topic=request_topic, + completed_topic=completed_topic, + failed_topic=failed_topic, + timeout_seconds=instance.timeout_seconds, + correlation_config=correlation_config, + consumer_group=consumer_group, + ) + + # Create consumer for reply topics + consumer = AIOKafkaConsumer( + completed_topic, + failed_topic, + bootstrap_servers=self._bootstrap_servers, + group_id=consumer_group, + auto_offset_reset=instance.auto_offset_reset, + enable_auto_commit=True, + ) + + rr_instance.consumer = consumer + + # Start consumer eagerly + await consumer.start() + self._logger.info( + "Started request-response consumer: instance=%s, " + "consumer_group=%s, topics=[%s, %s]", + instance.name, + consumer_group, + completed_topic, + failed_topic, + ) + + # Start background task to process responses + consumer_task = asyncio.create_task( + self._consume_responses(rr_instance), + name=f"rr-consumer-{instance.name}", + ) + rr_instance.consumer_task = consumer_task + + # Store instance + self._instances[instance.name] = rr_instance + + async def _consume_responses(self, instance: RequestResponseInstanceState) -> None: + """Background task that consumes responses and resolves pending futures. + + Runs continuously until cleanup() is called. Matches incoming responses + to pending requests via correlation ID. + + Args: + instance: The request-response instance to consume for. + """ + consumer = instance.consumer + if consumer is None: + return + + try: + async for message in consumer: + await self._handle_response_message(instance, message) + except asyncio.CancelledError: + self._logger.debug( + "Consumer task cancelled for instance: %s", + instance.name, + ) + raise + except Exception as e: + self._logger.exception( + "Unexpected error in consumer task for instance %s: %s", + instance.name, + e, + ) + + async def _handle_response_message( + self, + instance: RequestResponseInstanceState, + message: ConsumerRecord, + ) -> None: + """Handle a single response message from Kafka. + + Extracts correlation ID and resolves the corresponding pending future. + + Args: + instance: The request-response instance. + message: The Kafka message received. + """ + try: + # Deserialize message value + if message.value is None: + self._logger.warning( + "Received empty message on topic %s, skipping", + message.topic, + ) + return + + response_data: dict[str, object] = json.loads(message.value.decode("utf-8")) + + # Extract correlation ID based on config + correlation_id = self._extract_correlation_id( + response_data, + instance.correlation_config, + ) + + if correlation_id is None: + self._logger.warning( + "Response missing correlation_id: topic=%s, instance=%s", + message.topic, + instance.name, + ) + return + + correlation_key = str(correlation_id) + + # Look up pending future + future = instance.pending.pop(correlation_key, None) + if future is None: + self._logger.debug( + "Orphan response received (no pending request): " + "correlation_id=%s, topic=%s, instance=%s", + correlation_key, + message.topic, + instance.name, + ) + return + + # Determine if this is a success or failure based on topic + is_failure = message.topic == instance.failed_topic + + if is_failure: + # Set exception for failed responses + error_message = response_data.get("error", "Request failed") + future.set_exception(RuntimeError(f"Request failed: {error_message}")) + else: + # Set result for successful responses + # Include correlation_id in response for tracing + response_data["_correlation_id"] = correlation_key + future.set_result(response_data) + + self._logger.debug( + "Resolved pending request: correlation_id=%s, topic=%s, " + "is_failure=%s, instance=%s", + correlation_key, + message.topic, + is_failure, + instance.name, + ) + + except json.JSONDecodeError as e: + self._logger.warning( + "Failed to decode response JSON: topic=%s, error=%s", + message.topic, + e, + ) + except Exception as e: + self._logger.exception( + "Error handling response message: topic=%s, error=%s", + message.topic, + e, + ) + + def _extract_correlation_id( + self, + data: dict[str, object], + config: ModelCorrelationConfig, + ) -> UUID | None: + """Extract correlation ID from response data based on configuration. + + Args: + data: Response data dictionary. + config: Correlation configuration specifying location and field. + + Returns: + Correlation ID as UUID if found, None otherwise. + """ + value: object | None = None + + if config.location == "body": + value = data.get(config.field) + elif config.location == "headers": + # Headers would be in message headers, not body + # For now, we only support body location + self._logger.warning( + "Header-based correlation not implemented, falling back to body" + ) + value = data.get(config.field) + + if value is None: + return None + + # Parse to UUID - correlation IDs are always UUIDs + try: + return UUID(str(value)) + except ValueError: + self._logger.warning( + "Invalid correlation_id format (not a UUID): %s", + value, + ) + return None + + async def send_request( + self, + instance_name: str, + payload: dict[str, object], + timeout_seconds: int | None = None, + ) -> dict[str, object]: + """Send a request and await the correlated response. + + Publishes a request to the instance's request topic and waits for + a response on the reply topics. If the payload lacks a correlation_id + at the configured location, one is injected. + + Correlation ID Handling: + - If correlation_id exists in payload: Use existing value + - If missing: Inject new UUID4 into payload + - Always: Return correlation_id in response (as _correlation_id) + + Args: + instance_name: Name of the wired request-response instance. + payload: Request payload dictionary. Modified in place to add + correlation_id if not present. + timeout_seconds: Override timeout for this request. If None, + uses the instance's configured timeout (default: 30s). + + Returns: + Response dictionary from the reply topic. Includes _correlation_id + field for tracing. + + Raises: + ProtocolConfigurationError: If instance_name is not wired. + InfraTimeoutError: If no response received within timeout. + InfraUnavailableError: If circuit breaker is open. + RuntimeError: If request failed (response on failed topic). + """ + # Get instance + instance = self._instances.get(instance_name) + if instance is None: + raise ProtocolConfigurationError( + f"Request-response instance '{instance_name}' not wired", + context=ModelInfraErrorContext.with_correlation( + transport_type=EnumInfraTransportType.KAFKA, + operation="send_request", + ), + instance_name=instance_name, + ) + + # Determine timeout + timeout = ( + timeout_seconds if timeout_seconds is not None else instance.timeout_seconds + ) + + # Extract or inject correlation_id + correlation_id = self._ensure_correlation_id( + payload, + instance.correlation_config, + ) + correlation_key = str(correlation_id) + + # Create future for response + future: asyncio.Future[dict[str, object]] = ( + asyncio.get_running_loop().create_future() + ) + instance.pending[correlation_key] = future + + try: + # Check circuit breaker before publish + async with self._circuit_breaker_lock: + await self._check_circuit_breaker( + operation="send_request", + correlation_id=correlation_id, + ) + + # Publish request + await self._publish_request(instance, payload, correlation_id) + + # Wait for response with timeout + try: + response = await asyncio.wait_for(future, timeout=timeout) + + # Record success in circuit breaker + async with self._circuit_breaker_lock: + await self._reset_circuit_breaker() + + return response + + except TimeoutError: + # Remove pending future on timeout + instance.pending.pop(correlation_key, None) + + # Raise InfraTimeoutError (NOT InfraUnavailableError) + timeout_context = ModelTimeoutErrorContext( + transport_type=EnumInfraTransportType.KAFKA, + operation="send_request", + target_name=instance.request_topic, + correlation_id=correlation_id, + timeout_seconds=float(timeout), + ) + raise InfraTimeoutError( + f"Request-response timeout after {timeout}s: " + f"instance={instance_name}, correlation_id={correlation_key}", + context=timeout_context, + ) from None + + except InfraUnavailableError: + # Circuit breaker open - re-raise without modification + instance.pending.pop(correlation_key, None) + raise + + except Exception: + # Record failure in circuit breaker + async with self._circuit_breaker_lock: + await self._record_circuit_failure( + operation="send_request", + correlation_id=correlation_id, + ) + + # Clean up pending future + instance.pending.pop(correlation_key, None) + raise + + def _ensure_correlation_id( + self, + payload: dict[str, object], + config: ModelCorrelationConfig, + ) -> UUID: + """Ensure correlation_id exists in payload, injecting if missing. + + Args: + payload: Request payload dictionary. Modified in place. + config: Correlation configuration. + + Returns: + The correlation ID as UUID (existing parsed or newly generated). + """ + existing = payload.get(config.field) + + if existing is not None: + # Parse existing to UUID - correlation IDs are always UUIDs + correlation_id = UUID(str(existing)) + else: + # Generate new UUID + correlation_id = uuid4() + payload[config.field] = str(correlation_id) + + return correlation_id + + async def _publish_request( + self, + instance: RequestResponseInstanceState, + payload: dict[str, object], + correlation_id: UUID, + ) -> None: + """Publish request to the instance's request topic. + + Args: + instance: Request-response instance. + payload: Request payload. + correlation_id: Correlation ID for logging and message key. + """ + # Serialize payload + value = json.dumps(payload).encode("utf-8") + + # Publish via event bus - convert UUID to string at serialization boundary + await self._event_bus.publish( + topic=instance.request_topic, + key=str(correlation_id).encode("utf-8"), + value=value, + ) + + self._logger.debug( + "Published request: topic=%s, correlation_id=%s, instance=%s", + instance.request_topic, + correlation_id, + instance.name, + ) + + async def cleanup(self) -> None: + """Clean up all request-response instances. + + Cancels all consumer tasks, stops consumers, and clears pending futures + with exceptions. Should be called during runtime shutdown. + + This method is safe to call multiple times - subsequent calls are no-ops. + """ + cleanup_count = len(self._instances) + if cleanup_count == 0: + return + + for instance_name, instance in list(self._instances.items()): + await self._cleanup_instance(instance) + + self._instances.clear() + self._logger.info( + "Cleaned up %d request-response instance(s)", + cleanup_count, + ) + + async def _cleanup_instance(self, instance: RequestResponseInstanceState) -> None: + """Clean up a single request-response instance. + + Args: + instance: Instance to clean up. + """ + # Cancel consumer task + if instance.consumer_task is not None and not instance.consumer_task.done(): + instance.consumer_task.cancel() + try: + await instance.consumer_task + except asyncio.CancelledError: + pass + + # Stop consumer + if instance.consumer is not None: + try: + await instance.consumer.stop() + except Exception as e: + self._logger.warning( + "Error stopping consumer for instance %s: %s", + instance.name, + e, + ) + + # Fail all pending futures + cleanup_error = RuntimeError( + f"Request-response instance '{instance.name}' was cleaned up" + ) + for correlation_key, future in instance.pending.items(): + if not future.done(): + future.set_exception(cleanup_error) + self._logger.debug( + "Failed pending request on cleanup: correlation_id=%s, instance=%s", + correlation_key, + instance.name, + ) + + instance.pending.clear() + self._logger.debug( + "Cleaned up instance: %s", + instance.name, + ) + + def get_boot_nonce(self) -> str: + """Return the boot nonce for this process. + + Useful for debugging and logging consumer group identification. + + Returns: + 8-character hex string unique to this process instance. + """ + return _BOOT_NONCE + + +__all__: list[str] = [ + "RequestResponseWiring", +] diff --git a/src/omnibase_infra/validation/infra_validators.py b/src/omnibase_infra/validation/infra_validators.py index a1fe68f6c7..834aba9b60 100644 --- a/src/omnibase_infra/validation/infra_validators.py +++ b/src/omnibase_infra/validation/infra_validators.py @@ -449,6 +449,7 @@ def get_architecture_exemptions() -> list[ExemptionPattern]: # - 117 (2026-02-01): OMN-1783 PostgresRepositoryRuntime (+2 unions) # call() return type: list[dict] | dict | None # _execute_with_timeout() return type: list[dict] | dict | None +# Note: OMN-1742 RequestResponseWiring uses UUID | None (optional, not counted) INFRA_MAX_UNIONS = 117 # Maximum allowed architecture violations in infrastructure code. diff --git a/tests/integration/runtime/test_request_response_wiring_integration.py b/tests/integration/runtime/test_request_response_wiring_integration.py new file mode 100644 index 0000000000..39eabc0806 --- /dev/null +++ b/tests/integration/runtime/test_request_response_wiring_integration.py @@ -0,0 +1,1066 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2025 OmniNode Team +"""Integration tests for RequestResponseWiring with in-memory event bus. + +These tests verify the complete request-response flow including: +1. Correlation ID injection and matching +2. Response routing (completed vs failed topics) +3. Concurrent request isolation +4. Timeout handling + +Architecture: + Since RequestResponseWiring uses AIOKafkaConsumer directly for reply topics, + these tests mock the consumer while using a coordinated responder to simulate + the full request-response cycle. This avoids Kafka infrastructure requirements + while testing the core correlation and routing logic. + +Related Tickets: + - OMN-1742: Request-response wiring for Kafka RPC patterns + +See Also: + - src/omnibase_infra/runtime/request_response_wiring.py + - tests/integration/runtime/test_event_bus_contract_wiring.py +""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import dataclass, field +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import uuid4 + +import pytest + +from omnibase_core.models.contracts.subcontracts import ( + ModelCorrelationConfig, + ModelReplyTopics, + ModelRequestResponseConfig, + ModelRequestResponseInstance, +) +from omnibase_infra.errors import InfraTimeoutError, ProtocolConfigurationError +from omnibase_infra.event_bus.event_bus_inmemory import EventBusInmemory +from omnibase_infra.runtime.request_response_wiring import RequestResponseWiring + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + +# ============================================================================= +# Mock Consumer Infrastructure +# ============================================================================= + + +@dataclass +class MockConsumerRecord: + """Mock Kafka ConsumerRecord for testing. + + Simulates the structure of aiokafka.ConsumerRecord with the fields + used by RequestResponseWiring._handle_response_message(). + """ + + topic: str + value: bytes | None + key: bytes | None = None + partition: int = 0 + offset: int = 0 + + +class MockAIOKafkaConsumer: + """Mock AIOKafkaConsumer that allows injecting messages programmatically. + + This mock simulates the async iterator pattern of AIOKafkaConsumer, + allowing tests to inject response messages that the RequestResponseWiring + will process. + + Attributes: + topics: Topics the consumer is subscribed to. + message_queue: Queue for injecting messages to be consumed. + started: Whether start() has been called. + stopped: Whether stop() has been called. + """ + + def __init__(self, *topics: str, **kwargs: object) -> None: + """Initialize mock consumer with subscribed topics. + + Args: + *topics: Topics to subscribe to. + **kwargs: Ignored kwargs (bootstrap_servers, group_id, etc.) + """ + self.topics = topics + self.message_queue: asyncio.Queue[MockConsumerRecord | None] = asyncio.Queue() + self.started = False + self.stopped = False + self._kwargs = kwargs + + async def start(self) -> None: + """Mark consumer as started.""" + self.started = True + + async def stop(self) -> None: + """Mark consumer as stopped and inject sentinel to unblock iterator.""" + self.stopped = True + # Inject None to unblock any waiting __anext__ + await self.message_queue.put(None) + + def __aiter__(self) -> MockAIOKafkaConsumer: + """Return self as async iterator.""" + return self + + async def __anext__(self) -> MockConsumerRecord: + """Get next message from queue, raise StopAsyncIteration if stopped.""" + if self.stopped: + raise StopAsyncIteration + message = await self.message_queue.get() + if message is None: + raise StopAsyncIteration + return message + + async def inject_message(self, message: MockConsumerRecord) -> None: + """Inject a message to be consumed. + + Args: + message: The message to inject into the consumer. + """ + await self.message_queue.put(message) + + +# ============================================================================= +# Fixtures +# ============================================================================= + + +@pytest.fixture +def environment() -> str: + """Test environment prefix.""" + return "test" + + +@pytest.fixture +def app_name() -> str: + """Test application name.""" + return "test-rpc-service" + + +@pytest.fixture +async def event_bus() -> AsyncIterator[EventBusInmemory]: + """Create and start an in-memory event bus. + + Yields: + Started EventBusInmemory instance. + """ + bus = EventBusInmemory(environment="test", group="rpc-test") + await bus.start() + yield bus + await bus.close() + + +@pytest.fixture +def request_response_config() -> ModelRequestResponseConfig: + """Create test request-response configuration. + + Returns: + Configuration with a single test-rpc instance. + """ + return ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="test-rpc", + request_topic="onex.cmd.test.request.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.test.completed.v1", + failed="onex.evt.test.failed.v1", + ), + timeout_seconds=5, + ) + ] + ) + + +@pytest.fixture +def multi_instance_config() -> ModelRequestResponseConfig: + """Create configuration with multiple request-response instances. + + Returns: + Configuration with three test instances for concurrent testing. + """ + return ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name=f"test-rpc-{i}", + request_topic=f"onex.cmd.test.request-{i}.v1", + reply_topics=ModelReplyTopics( + completed=f"onex.evt.test.completed-{i}.v1", + failed=f"onex.evt.test.failed-{i}.v1", + ), + timeout_seconds=5, + ) + for i in range(3) + ] + ) + + +# ============================================================================= +# Test Class +# ============================================================================= + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestRequestResponseWiringIntegration: + """Integration tests for RequestResponseWiring with in-memory event bus. + + These tests verify the core request-response functionality without + requiring real Kafka infrastructure. The AIOKafkaConsumer is mocked + to allow programmatic message injection. + """ + + async def test_successful_request_response_flow( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Test complete request-response cycle with correlation matching. + + Verifies: + 1. Request is published to request topic + 2. Response on completed topic resolves the pending future + 3. Correlation ID matches between request and response + 4. Response data is correctly returned + + The test simulates a responder by: + 1. Capturing published requests via event bus subscription + 2. Injecting correlated responses into the mock consumer + """ + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + # EventBusInmemory implements ProtocolEventBusPublisher via duck typing + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(request_response_config) + + assert mock_consumer is not None, "Consumer should be created" + assert mock_consumer.started, "Consumer should be started" + + # Track published requests + published_requests: list[dict[str, object]] = [] + + async def capture_request(msg: object) -> None: + """Capture published request and inject response.""" + value = getattr(msg, "value", b"") + request_data = json.loads(value.decode("utf-8")) + published_requests.append(request_data) + + # Simulate responder: inject response to mock consumer + correlation_id = request_data.get("correlation_id") + response = MockConsumerRecord( + topic=f"{environment}.onex.evt.test.completed.v1", + value=json.dumps( + { + "correlation_id": correlation_id, + "result": {"status": "success", "data": "processed"}, + } + ).encode("utf-8"), + ) + await mock_consumer.inject_message(response) + + # Subscribe to request topic to capture and respond + from omnibase_infra.models import ModelNodeIdentity + + identity = ModelNodeIdentity( + env=environment, + service="responder", + node_name="test-responder", + version="v1", + ) + request_topic = f"{environment}.onex.cmd.test.request.v1" + await event_bus.subscribe(request_topic, identity, capture_request) + + # Send request + request_payload: dict[str, object] = { + "action": "process", + "data": {"input": "test-value"}, + } + response = await wiring.send_request( + instance_name="test-rpc", + payload=request_payload, + ) + + # Verify request was published + assert len(published_requests) == 1 + published = published_requests[0] + assert published["action"] == "process" + assert "correlation_id" in published # Auto-injected + + # Verify response - cast result to dict for type safety + result = response["result"] + assert isinstance(result, dict) + assert result["status"] == "success" + assert result["data"] == "processed" + assert "_correlation_id" in response # Correlation tracking + + # Verify correlation ID matches + assert response["_correlation_id"] == published["correlation_id"] + + # Cleanup + await wiring.cleanup() + assert mock_consumer.stopped, "Consumer should be stopped on cleanup" + + async def test_error_response_flow( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Test error response handling via failed topic. + + Verifies: + 1. Response on failed topic raises RuntimeError + 2. Error message from response is included in exception + 3. Correlation ID tracking works for error responses + """ + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + # EventBusInmemory implements ProtocolEventBusPublisher via duck typing + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(request_response_config) + assert mock_consumer is not None + + async def capture_and_fail(msg: object) -> None: + """Capture request and inject error response.""" + value = getattr(msg, "value", b"") + request_data = json.loads(value.decode("utf-8")) + correlation_id = request_data.get("correlation_id") + + # Inject error response on failed topic + error_response = MockConsumerRecord( + topic=f"{environment}.onex.evt.test.failed.v1", + value=json.dumps( + { + "correlation_id": correlation_id, + "error": "Processing failed: invalid input format", + } + ).encode("utf-8"), + ) + await mock_consumer.inject_message(error_response) + + from omnibase_infra.models import ModelNodeIdentity + + identity = ModelNodeIdentity( + env=environment, + service="responder", + node_name="error-responder", + version="v1", + ) + request_topic = f"{environment}.onex.cmd.test.request.v1" + await event_bus.subscribe(request_topic, identity, capture_and_fail) + + # Send request and expect error + with pytest.raises(RuntimeError, match="Processing failed"): + await wiring.send_request( + instance_name="test-rpc", + payload={"action": "invalid"}, + ) + + await wiring.cleanup() + + async def test_multiple_concurrent_requests( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Test concurrent requests are correctly correlated. + + Verifies: + 1. Multiple concurrent requests are tracked independently + 2. Each response matches its originating request via correlation_id + 3. No cross-contamination between concurrent requests + """ + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + # EventBusInmemory implements ProtocolEventBusPublisher via duck typing + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(request_response_config) + assert mock_consumer is not None + + # Track requests and their order + request_order: list[str] = [] + + async def capture_and_respond_with_delay(msg: object) -> None: + """Capture request, add delay variation, then respond.""" + value = getattr(msg, "value", b"") + request_data = json.loads(value.decode("utf-8")) + correlation_id = request_data.get("correlation_id") + request_id = request_data.get("request_id") + + request_order.append(str(request_id)) + + # Vary response delay based on request_id to test out-of-order handling + delay = 0.01 * (3 - int(str(request_id))) + await asyncio.sleep(delay) + + # Respond with request_id included for verification + response = MockConsumerRecord( + topic=f"{environment}.onex.evt.test.completed.v1", + value=json.dumps( + { + "correlation_id": correlation_id, + "result": { + "request_id": request_id, + "processed_at": "timestamp", + }, + } + ).encode("utf-8"), + ) + await mock_consumer.inject_message(response) + + from omnibase_infra.models import ModelNodeIdentity + + identity = ModelNodeIdentity( + env=environment, + service="responder", + node_name="concurrent-responder", + version="v1", + ) + request_topic = f"{environment}.onex.cmd.test.request.v1" + await event_bus.subscribe( + request_topic, identity, capture_and_respond_with_delay + ) + + # Send 3 concurrent requests + async def send_request(request_id: int) -> dict[str, object]: + return await wiring.send_request( + instance_name="test-rpc", + payload={"request_id": request_id, "data": f"payload-{request_id}"}, + ) + + # Launch all requests concurrently + results = await asyncio.gather( + send_request(0), + send_request(1), + send_request(2), + ) + + # Verify all 3 requests were processed + assert len(results) == 3 + assert len(request_order) == 3 + + # Verify each response matches its request (no cross-contamination) + for i, result in enumerate(results): + result_data = result["result"] + assert isinstance(result_data, dict) + assert result_data["request_id"] == i, ( + f"Response {i} should contain request_id {i}, " + f"got {result_data['request_id']}" + ) + + await wiring.cleanup() + + async def test_correlation_id_preserved_when_provided( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Test that existing correlation_id is preserved, not overwritten. + + Verifies: + 1. When payload contains correlation_id, it is used + 2. Wiring does not inject a new correlation_id + 3. Response tracking uses the provided correlation_id + """ + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + # EventBusInmemory implements ProtocolEventBusPublisher via duck typing + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(request_response_config) + assert mock_consumer is not None + + # Pre-defined correlation_id + provided_correlation_id = str(uuid4()) + captured_correlation_id: str | None = None + + async def capture_and_verify_correlation(msg: object) -> None: + """Capture request and verify correlation_id.""" + nonlocal captured_correlation_id + value = getattr(msg, "value", b"") + request_data = json.loads(value.decode("utf-8")) + captured_correlation_id = request_data.get("correlation_id") + + response = MockConsumerRecord( + topic=f"{environment}.onex.evt.test.completed.v1", + value=json.dumps( + { + "correlation_id": captured_correlation_id, + "result": {"verified": True}, + } + ).encode("utf-8"), + ) + await mock_consumer.inject_message(response) + + from omnibase_infra.models import ModelNodeIdentity + + identity = ModelNodeIdentity( + env=environment, + service="responder", + node_name="correlation-verifier", + version="v1", + ) + request_topic = f"{environment}.onex.cmd.test.request.v1" + await event_bus.subscribe( + request_topic, identity, capture_and_verify_correlation + ) + + # Send request with pre-existing correlation_id + response = await wiring.send_request( + instance_name="test-rpc", + payload={ + "correlation_id": provided_correlation_id, + "action": "test", + }, + ) + + # Verify the provided correlation_id was used + assert captured_correlation_id == provided_correlation_id + assert response["_correlation_id"] == provided_correlation_id + + await wiring.cleanup() + + async def test_timeout_raises_infra_timeout_error( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + ) -> None: + """Test that timeout raises InfraTimeoutError, not InfraUnavailableError. + + Verifies: + 1. When no response is received within timeout, InfraTimeoutError is raised + 2. The error includes timeout details and correlation_id + 3. Pending future is cleaned up after timeout + """ + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + # Config with very short timeout for testing + short_timeout_config = ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="timeout-test", + request_topic="onex.cmd.test.timeout.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.test.timeout-completed.v1", + failed="onex.evt.test.timeout-failed.v1", + ), + timeout_seconds=1, # Short timeout + ) + ] + ) + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(short_timeout_config) + assert mock_consumer is not None + + # Do NOT set up a responder - let it timeout + + with pytest.raises(InfraTimeoutError) as exc_info: + await wiring.send_request( + instance_name="timeout-test", + payload={"action": "will-timeout"}, + timeout_seconds=1, # Short timeout (int required) + ) + + # Verify error details + error = exc_info.value + assert "timeout" in str(error).lower() + assert "1s" in str(error) or "1 " in str(error) # Timeout value + + await wiring.cleanup() + + async def test_unwired_instance_raises_protocol_error( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Test that sending to unwired instance raises ProtocolConfigurationError. + + Verifies: + 1. Attempting to send to non-existent instance raises clear error + 2. Error message includes the invalid instance name + """ + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + # Wire only "test-rpc" instance + await wiring.wire_request_response(request_response_config) + + # Try to send to non-existent instance + with pytest.raises( + ProtocolConfigurationError, match=r"not-wired.*not wired" + ): + await wiring.send_request( + instance_name="not-wired", + payload={"action": "test"}, + ) + + await wiring.cleanup() + + async def test_cleanup_cancels_pending_requests( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Test that cleanup properly cancels all pending request futures. + + Verifies: + 1. Pending requests fail with RuntimeError on cleanup + 2. Consumer is stopped + 3. Cleanup is idempotent (safe to call multiple times) + """ + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(request_response_config) + assert mock_consumer is not None + + # Start a request but don't provide a response + request_task = asyncio.create_task( + wiring.send_request( + instance_name="test-rpc", + payload={"action": "will-be-cancelled"}, + timeout_seconds=10, # Long timeout + ) + ) + + # Give the task time to register the pending future + await asyncio.sleep(0.05) + + # Cleanup should cancel pending requests + await wiring.cleanup() + + # Request should fail with RuntimeError + with pytest.raises(RuntimeError, match="cleaned up"): + await request_task + + # Verify cleanup is idempotent + await wiring.cleanup() # Should not raise + + async def test_boot_nonce_uniqueness( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + ) -> None: + """Test that boot nonce is consistent within a process. + + Verifies: + 1. get_boot_nonce() returns consistent value + 2. Boot nonce is used in consumer group naming + """ + mock_consumer: MockAIOKafkaConsumer | None = None + captured_group_id: str | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer, captured_group_id + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + captured_group_id = str(kwargs.get("group_id", "")) + return mock_consumer + + config = ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="nonce-test", + request_topic="onex.cmd.test.nonce.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.test.nonce-completed.v1", + failed="onex.evt.test.nonce-failed.v1", + ), + timeout_seconds=5, + ) + ] + ) + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(config) + + boot_nonce = wiring.get_boot_nonce() + + # Verify boot nonce format (8-char hex) + assert len(boot_nonce) == 8 + assert all(c in "0123456789abcdef" for c in boot_nonce) + + # Verify boot nonce is in consumer group + assert captured_group_id is not None + assert boot_nonce in captured_group_id + assert f"{environment}.rr.nonce-test.{boot_nonce}" == captured_group_id + + await wiring.cleanup() + + async def test_topic_resolution_includes_environment( + self, + event_bus: EventBusInmemory, + environment: str, + app_name: str, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Test that topics are resolved with environment prefix. + + Verifies: + 1. Request topic includes environment prefix + 2. Reply topics include environment prefix + """ + mock_consumer: MockAIOKafkaConsumer | None = None + subscribed_topics: tuple[str, ...] = () + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer, subscribed_topics + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + subscribed_topics = topics + return mock_consumer + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(request_response_config) + + # Verify resolve_topic method + resolved = wiring.resolve_topic("onex.cmd.test.request.v1") + assert resolved == f"{environment}.onex.cmd.test.request.v1" + + # Verify consumer subscribed to resolved topics + assert f"{environment}.onex.evt.test.completed.v1" in subscribed_topics + assert f"{environment}.onex.evt.test.failed.v1" in subscribed_topics + + await wiring.cleanup() + + +# ============================================================================= +# Edge Case Tests +# ============================================================================= + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestRequestResponseWiringEdgeCases: + """Edge case tests for RequestResponseWiring.""" + + async def test_empty_response_body_handled( + self, + environment: str, + app_name: str, + ) -> None: + """Test that empty/null response body is handled gracefully. + + Verifies: + 1. Empty response body is logged but doesn't crash + 2. No exception is raised for empty messages + """ + event_bus = EventBusInmemory(environment="test", group="edge-test") + await event_bus.start() + + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + config = ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="empty-test", + request_topic="onex.cmd.test.empty.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.test.empty-completed.v1", + failed="onex.evt.test.empty-failed.v1", + ), + timeout_seconds=1, + ) + ] + ) + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(config) + assert mock_consumer is not None + + # Inject empty message (should be skipped, not crash) + empty_response = MockConsumerRecord( + topic=f"{environment}.onex.evt.test.empty-completed.v1", + value=None, + ) + await mock_consumer.inject_message(empty_response) + + # Give consumer time to process + await asyncio.sleep(0.05) + + # Should not have crashed - cleanup works + await wiring.cleanup() + await event_bus.close() + + async def test_orphan_response_handled( + self, + environment: str, + app_name: str, + ) -> None: + """Test that responses without pending requests are handled gracefully. + + Verifies: + 1. Orphan responses (no matching correlation_id) are logged + 2. No exception is raised for orphan responses + """ + event_bus = EventBusInmemory(environment="test", group="orphan-test") + await event_bus.start() + + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + config = ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="orphan-test", + request_topic="onex.cmd.test.orphan.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.test.orphan-completed.v1", + failed="onex.evt.test.orphan-failed.v1", + ), + timeout_seconds=1, + ) + ] + ) + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + await wiring.wire_request_response(config) + assert mock_consumer is not None + + # Inject response with unknown correlation_id (orphan) + orphan_response = MockConsumerRecord( + topic=f"{environment}.onex.evt.test.orphan-completed.v1", + value=json.dumps( + { + "correlation_id": str(uuid4()), # No matching request + "result": {"orphan": True}, + } + ).encode("utf-8"), + ) + await mock_consumer.inject_message(orphan_response) + + # Give consumer time to process + await asyncio.sleep(0.05) + + # Should not have crashed - cleanup works + await wiring.cleanup() + await event_bus.close() + + async def test_duplicate_instance_wiring_raises_error( + self, + environment: str, + app_name: str, + ) -> None: + """Test that wiring duplicate instance names raises error. + + Verifies: + 1. Attempting to wire same instance name twice raises ProtocolConfigurationError + 2. Error message identifies the duplicate instance + """ + event_bus = EventBusInmemory(environment="test", group="duplicate-test") + await event_bus.start() + + mock_consumer: MockAIOKafkaConsumer | None = None + + def consumer_factory(*topics: str, **kwargs: object) -> MockAIOKafkaConsumer: + nonlocal mock_consumer + mock_consumer = MockAIOKafkaConsumer(*topics, **kwargs) + return mock_consumer + + config = ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="duplicate-test", + request_topic="onex.cmd.test.dup.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.test.dup-completed.v1", + failed="onex.evt.test.dup-failed.v1", + ), + timeout_seconds=5, + ) + ] + ) + + with patch( + "omnibase_infra.runtime.request_response_wiring.AIOKafkaConsumer", + side_effect=consumer_factory, + ): + wiring = RequestResponseWiring( + event_bus=event_bus, # type: ignore[arg-type] + environment=environment, + app_name=app_name, + bootstrap_servers="localhost:9092", + ) + + # First wiring succeeds + await wiring.wire_request_response(config) + + # Second wiring with same instance name should fail + with pytest.raises( + ProtocolConfigurationError, match=r"duplicate-test.*already wired" + ): + await wiring.wire_request_response(config) + + await wiring.cleanup() + await event_bus.close() diff --git a/tests/unit/runtime/test_request_response_wiring.py b/tests/unit/runtime/test_request_response_wiring.py new file mode 100644 index 0000000000..49c7ec4c65 --- /dev/null +++ b/tests/unit/runtime/test_request_response_wiring.py @@ -0,0 +1,1137 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2025 OmniNode Team +""" +Comprehensive unit tests for RequestResponseWiring. + +This test suite validates: +- Correlation ID handling (injection, preservation, return in result) +- Timeout behavior (InfraTimeoutError, context, cleanup) +- Consumer group naming format and boot nonce consistency +- Pending map management (success, failure, orphan responses) +- Cleanup behavior (idempotency, task cancellation, future failures) +- Circuit breaker integration (threshold, unavailable error) + +Test Organization: + - TestRequestResponseWiringInit: Initialization and validation + - TestCorrelationIdHandling: Correlation ID injection/preservation + - TestTimeoutBehavior: Timeout handling and error context + - TestConsumerGroup: Consumer group format and boot nonce + - TestPendingMapHandling: Response resolution and orphan handling + - TestCleanup: Cleanup behavior and idempotency + - TestCircuitBreaker: Circuit breaker integration + +Related: + - OMN-1742: Request-response wiring for Kafka RPC patterns + - RequestResponseWiring: Implementation under test + +.. versionadded:: 0.3.1 +""" + +from __future__ import annotations + +import asyncio +import json +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import UUID, uuid4 + +import pytest + +from omnibase_core.models.contracts.subcontracts import ( + ModelCorrelationConfig, + ModelReplyTopics, + ModelRequestResponseConfig, + ModelRequestResponseInstance, +) +from omnibase_infra.enums import EnumInfraTransportType +from omnibase_infra.errors import ( + InfraTimeoutError, + InfraUnavailableError, + ProtocolConfigurationError, +) + +if TYPE_CHECKING: + from omnibase_infra.runtime.request_response_wiring import RequestResponseWiring + + +# ============================================================================= +# Test Fixtures +# ============================================================================= + + +@pytest.fixture +def mock_event_bus() -> MagicMock: + """Mock event bus with publish/subscribe capabilities.""" + bus = MagicMock() + bus.publish = AsyncMock() + bus._bootstrap_servers = "localhost:9092" + return bus + + +@pytest.fixture +def request_response_config() -> ModelRequestResponseConfig: + """Sample ModelRequestResponseConfig for testing.""" + return ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="routing", + request_topic="onex.cmd.routing.request.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.routing.completed.v1", + failed="onex.evt.routing.failed.v1", + ), + timeout_seconds=5, # Short for tests + ) + ] + ) + + +@pytest.fixture +def request_response_config_multiple() -> ModelRequestResponseConfig: + """Config with multiple request-response instances.""" + return ModelRequestResponseConfig( + instances=[ + ModelRequestResponseInstance( + name="routing", + request_topic="onex.cmd.routing.request.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.routing.completed.v1", + failed="onex.evt.routing.failed.v1", + ), + timeout_seconds=5, + ), + ModelRequestResponseInstance( + name="analysis", + request_topic="onex.cmd.intelligence.analyze.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.intelligence.analyzed.v1", + failed="onex.evt.intelligence.failed.v1", + ), + timeout_seconds=30, + ), + ] + ) + + +@pytest.fixture +async def wiring(mock_event_bus: MagicMock) -> RequestResponseWiring: + """Configured RequestResponseWiring instance (without consumers started).""" + from omnibase_infra.runtime.request_response_wiring import RequestResponseWiring + + return RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + +# ============================================================================= +# Initialization Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestRequestResponseWiringInit: + """Test RequestResponseWiring initialization.""" + + async def test_init_with_valid_params(self, mock_event_bus: MagicMock) -> None: + """Test successful initialization with valid parameters.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="dev", + app_name="my-service", + bootstrap_servers="localhost:9092", + ) + + assert wiring._environment == "dev" + assert wiring._app_name == "my-service" + assert wiring._bootstrap_servers == "localhost:9092" + assert len(wiring._instances) == 0 + + async def test_init_raises_on_empty_environment( + self, mock_event_bus: MagicMock + ) -> None: + """Test that empty environment raises ValueError.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + with pytest.raises(ValueError, match="environment must be a non-empty string"): + RequestResponseWiring( + event_bus=mock_event_bus, + environment="", + app_name="test-app", + ) + + async def test_init_raises_on_empty_app_name( + self, mock_event_bus: MagicMock + ) -> None: + """Test that empty app_name raises ValueError.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + with pytest.raises(ValueError, match="app_name must be a non-empty string"): + RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="", + ) + + async def test_init_uses_event_bus_bootstrap_servers( + self, mock_event_bus: MagicMock + ) -> None: + """Test bootstrap servers fallback to event_bus attribute.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + mock_event_bus._bootstrap_servers = "kafka.example.com:9092" + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + # No bootstrap_servers provided + ) + + assert wiring._bootstrap_servers == "kafka.example.com:9092" + + async def test_resolve_topic_adds_environment_prefix( + self, wiring: RequestResponseWiring + ) -> None: + """Test that resolve_topic adds environment prefix correctly.""" + topic = wiring.resolve_topic("onex.cmd.routing.request.v1") + assert topic == "test.onex.cmd.routing.request.v1" + + +# ============================================================================= +# Correlation ID Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestCorrelationIdHandling: + """Test correlation ID injection, preservation, and return.""" + + async def test_send_request_injects_correlation_id_when_missing( + self, + mock_event_bus: MagicMock, + request_response_config: ModelRequestResponseConfig, + ) -> None: + """Verify UUID4 is injected when correlation_id is missing.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + # Manually create instance state without starting consumer + instance_name = "routing" + wiring._instances[instance_name] = MagicMock( + name=instance_name, + request_topic="test.onex.cmd.routing.request.v1", + completed_topic="test.onex.evt.routing.completed.v1", + failed_topic="test.onex.evt.routing.failed.v1", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.rr.routing.abc12345", + pending={}, + ) + + # Prepare payload without correlation_id + payload: dict[str, object] = {"data": "test"} + + # Call _ensure_correlation_id to verify injection + correlation_id = wiring._ensure_correlation_id( + payload, + ModelCorrelationConfig(), + ) + + # Verify correlation_id was injected + assert correlation_id is not None + assert "correlation_id" in payload + # Verify it's a valid UUID4 string + injected_uuid = UUID(str(payload["correlation_id"])) + assert injected_uuid.version == 4 + + async def test_send_request_preserves_existing_correlation_id( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify existing correlation_id is not overwritten.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + # Prepare payload with existing correlation_id + existing_id = uuid4() + payload: dict[str, object] = { + "data": "test", + "correlation_id": str(existing_id), + } + + # Call _ensure_correlation_id + result_id = wiring._ensure_correlation_id( + payload, + ModelCorrelationConfig(), + ) + + # Verify existing correlation_id is preserved + assert str(result_id) == str(existing_id) + assert payload["correlation_id"] == str(existing_id) + + async def test_correlation_id_returned_in_result( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify result contains correlation_id for tracing.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + ) + + # Create a mock instance + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Add pending future + correlation_key = str(uuid4()) + future: asyncio.Future[dict[str, object]] = ( + asyncio.get_running_loop().create_future() + ) + instance.pending[correlation_key] = future + + # Simulate setting result with correlation_id (as done in _handle_response_message) + response_data: dict[str, object] = {"result": "success"} + response_data["_correlation_id"] = correlation_key + future.set_result(response_data) + + # Verify result contains _correlation_id + result = await future + assert "_correlation_id" in result + assert result["_correlation_id"] == correlation_key + + +# ============================================================================= +# Timeout Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestTimeoutBehavior: + """Test timeout handling and error context.""" + + async def test_send_request_raises_infra_timeout_error_on_timeout( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify InfraTimeoutError is raised (not InfraUnavailableError).""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + # Create mock instance with very short timeout + instance_name = "routing" + instance = RequestResponseInstanceState( + name=instance_name, + request_topic="test.onex.cmd.routing.request.v1", + completed_topic="test.onex.evt.routing.completed.v1", + failed_topic="test.onex.evt.routing.failed.v1", + timeout_seconds=1, # Very short timeout + correlation_config=ModelCorrelationConfig(), + consumer_group="test.rr.routing.abc12345", + ) + wiring._instances[instance_name] = instance + + # Send request with timeout (no response will come) + payload: dict[str, object] = {"data": "test"} + + with pytest.raises(InfraTimeoutError) as exc_info: + # Use 0.1s timeout for fast test + await wiring.send_request(instance_name, payload, timeout_seconds=1) + + # Verify it's InfraTimeoutError, NOT InfraUnavailableError + assert isinstance(exc_info.value, InfraTimeoutError) + assert not isinstance(exc_info.value, InfraUnavailableError) + assert "timeout" in exc_info.value.message.lower() + + async def test_timeout_error_has_correct_context( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify ModelTimeoutErrorContext fields are correct.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance_name = "routing" + request_topic = "test.onex.cmd.routing.request.v1" + instance = RequestResponseInstanceState( + name=instance_name, + request_topic=request_topic, + completed_topic="test.onex.evt.routing.completed.v1", + failed_topic="test.onex.evt.routing.failed.v1", + timeout_seconds=1, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.rr.routing.abc12345", + ) + wiring._instances[instance_name] = instance + + payload: dict[str, object] = {"data": "test"} + + with pytest.raises(InfraTimeoutError) as exc_info: + await wiring.send_request(instance_name, payload, timeout_seconds=1) + + error = exc_info.value + context = error.model.context + + # Verify context fields + assert context["transport_type"] == EnumInfraTransportType.KAFKA + assert context["operation"] == "send_request" + assert context["target_name"] == request_topic + assert "timeout_seconds" in context + assert context["timeout_seconds"] == 1.0 + + async def test_timeout_cleans_up_pending_future( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify pending map doesn't leak on timeout.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance_name = "routing" + instance = RequestResponseInstanceState( + name=instance_name, + request_topic="test.onex.cmd.routing.request.v1", + completed_topic="test.onex.evt.routing.completed.v1", + failed_topic="test.onex.evt.routing.failed.v1", + timeout_seconds=1, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.rr.routing.abc12345", + ) + wiring._instances[instance_name] = instance + + # Verify pending map is empty initially + assert len(instance.pending) == 0 + + payload: dict[str, object] = {"data": "test"} + + with pytest.raises(InfraTimeoutError): + await wiring.send_request(instance_name, payload, timeout_seconds=1) + + # Verify pending map is empty after timeout (cleaned up) + assert len(instance.pending) == 0 + + +# ============================================================================= +# Consumer Group Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestConsumerGroup: + """Test consumer group naming and boot nonce.""" + + async def test_consumer_group_format( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify consumer group format: {environment}.rr.{instance_name}.{boot_nonce}.""" + from omnibase_infra.runtime.request_response_wiring import ( + _BOOT_NONCE, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="dev", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + # Create instance manually to check consumer group + instance_name = "code-analysis" + expected_group = f"dev.rr.{instance_name}.{_BOOT_NONCE}" + + # Verify the format matches expected pattern + # The actual consumer group is built in _wire_instance + # We test the format by checking expected pattern + assert _BOOT_NONCE is not None + assert len(_BOOT_NONCE) == 8 + assert all(c in "0123456789abcdef" for c in _BOOT_NONCE) + assert expected_group.startswith("dev.rr.code-analysis.") + + async def test_boot_nonce_is_consistent_across_instances( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify same boot_nonce for all instances in process.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + wiring1 = RequestResponseWiring( + event_bus=mock_event_bus, + environment="dev", + app_name="app1", + bootstrap_servers="localhost:9092", + ) + + wiring2 = RequestResponseWiring( + event_bus=mock_event_bus, + environment="prod", + app_name="app2", + bootstrap_servers="localhost:9092", + ) + + # Both wirings should have the same boot nonce + assert wiring1.get_boot_nonce() == wiring2.get_boot_nonce() + assert len(wiring1.get_boot_nonce()) == 8 + + async def test_boot_nonce_is_8_char_hex( + self, + wiring: RequestResponseWiring, + ) -> None: + """Verify boot nonce is 8-character hex string.""" + nonce = wiring.get_boot_nonce() + + assert len(nonce) == 8 + # Verify it's valid hex + int(nonce, 16) # Will raise if not valid hex + + +# ============================================================================= +# Pending Map Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestPendingMapHandling: + """Test pending map response resolution.""" + + async def test_successful_response_resolves_future(self) -> None: + """Verify completed topic resolves future with result.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Create pending future + correlation_key = str(uuid4()) + future: asyncio.Future[dict[str, object]] = ( + asyncio.get_running_loop().create_future() + ) + instance.pending[correlation_key] = future + + # Resolve future (simulating response handling) + response_data: dict[str, object] = {"result": "success", "data": [1, 2, 3]} + response_data["_correlation_id"] = correlation_key + future.set_result(response_data) + + # Verify future resolved correctly + result = await future + assert result["result"] == "success" + assert result["data"] == [1, 2, 3] + assert result["_correlation_id"] == correlation_key + + async def test_failed_response_rejects_future(self) -> None: + """Verify failed topic sets exception on future.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Create pending future + correlation_key = str(uuid4()) + future: asyncio.Future[dict[str, object]] = ( + asyncio.get_running_loop().create_future() + ) + instance.pending[correlation_key] = future + + # Reject future (simulating failed response) + error_message = "Request processing failed" + future.set_exception(RuntimeError(f"Request failed: {error_message}")) + + # Verify future rejected with exception + with pytest.raises(RuntimeError) as exc_info: + await future + + assert "Request failed" in str(exc_info.value) + + async def test_orphan_response_is_logged_not_crashed( + self, + mock_event_bus: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Verify missing correlation_id doesn't crash consumer.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Create mock message with unknown correlation_id + mock_message = MagicMock() + mock_message.value = json.dumps( + { + "correlation_id": str(uuid4()), # Unknown correlation ID + "result": "success", + } + ).encode("utf-8") + mock_message.topic = "test.completed" + + # Handle response - should not crash, just log + await wiring._handle_response_message(instance, mock_message) + + # Instance pending map should remain empty (no matching future) + assert len(instance.pending) == 0 + + +# ============================================================================= +# Cleanup Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestCleanup: + """Test cleanup behavior and idempotency.""" + + async def test_cleanup_cancels_consumer_tasks( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify consumer tasks are cancelled on cleanup.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + # Create mock consumer task + async def mock_consumer() -> None: + await asyncio.sleep(100) # Long running task + + consumer_task = asyncio.create_task(mock_consumer()) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + instance.consumer_task = consumer_task + instance.consumer = MagicMock() + instance.consumer.stop = AsyncMock() + + wiring._instances["test"] = instance + + # Cleanup should cancel task + await wiring.cleanup() + + # Verify task was cancelled + assert consumer_task.cancelled() or consumer_task.done() + assert len(wiring._instances) == 0 + + async def test_cleanup_fails_pending_futures( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify waiting futures get exceptions on cleanup.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + instance.consumer = MagicMock() + instance.consumer.stop = AsyncMock() + + # Add pending futures + future1: asyncio.Future[dict[str, object]] = ( + asyncio.get_running_loop().create_future() + ) + future2: asyncio.Future[dict[str, object]] = ( + asyncio.get_running_loop().create_future() + ) + instance.pending["id1"] = future1 + instance.pending["id2"] = future2 + + wiring._instances["test"] = instance + + # Cleanup should fail pending futures + await wiring.cleanup() + + # Verify futures have exceptions + assert future1.done() + assert future2.done() + + with pytest.raises(RuntimeError) as exc_info: + future1.result() + assert "cleaned up" in str(exc_info.value) + + with pytest.raises(RuntimeError) as exc_info: + future2.result() + assert "cleaned up" in str(exc_info.value) + + async def test_cleanup_is_idempotent( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify cleanup can be called multiple times safely.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + instance.consumer = MagicMock() + instance.consumer.stop = AsyncMock() + + wiring._instances["test"] = instance + + # Call cleanup multiple times + await wiring.cleanup() + await wiring.cleanup() + await wiring.cleanup() + + # Should not raise and instances should be empty + assert len(wiring._instances) == 0 + + +# ============================================================================= +# Circuit Breaker Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestCircuitBreaker: + """Test circuit breaker integration.""" + + async def test_circuit_breaker_opens_after_threshold_failures( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify 5 failures opens circuit.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance_name = "routing" + instance = RequestResponseInstanceState( + name=instance_name, + request_topic="test.onex.cmd.routing.request.v1", + completed_topic="test.onex.evt.routing.completed.v1", + failed_topic="test.onex.evt.routing.failed.v1", + timeout_seconds=1, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.rr.routing.abc12345", + ) + wiring._instances[instance_name] = instance + + # Make publish fail + mock_event_bus.publish.side_effect = Exception("Publish failed") + + # Record 5 failures (threshold is 5 by default) + for _ in range(5): + try: + payload: dict[str, object] = {"data": "test"} + await wiring.send_request(instance_name, payload, timeout_seconds=1) + except Exception: + pass # Expected to fail + + # Circuit should now be open + assert wiring._circuit_breaker_open is True + + async def test_circuit_breaker_raises_unavailable_when_open( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify publish blocked when circuit is open.""" + import time + + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance_name = "routing" + instance = RequestResponseInstanceState( + name=instance_name, + request_topic="test.onex.cmd.routing.request.v1", + completed_topic="test.onex.evt.routing.completed.v1", + failed_topic="test.onex.evt.routing.failed.v1", + timeout_seconds=1, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.rr.routing.abc12345", + ) + wiring._instances[instance_name] = instance + + # Manually open the circuit breaker with future open_until time + # to prevent auto-transition to half-open state + wiring._circuit_breaker_open = True + wiring._circuit_breaker_failures = 5 + wiring._circuit_breaker_open_until = time.time() + 3600 # 1 hour in future + + # Attempt to send request should raise InfraUnavailableError + payload: dict[str, object] = {"data": "test"} + + with pytest.raises(InfraUnavailableError) as exc_info: + await wiring.send_request(instance_name, payload, timeout_seconds=1) + + assert "Circuit breaker is open" in exc_info.value.message + + +# ============================================================================= +# Instance Not Wired Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestInstanceNotWired: + """Test error handling for unwired instances.""" + + async def test_send_request_raises_config_error_for_unwired_instance( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify ProtocolConfigurationError when instance not wired.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + # No instances wired + payload: dict[str, object] = {"data": "test"} + + with pytest.raises(ProtocolConfigurationError) as exc_info: + await wiring.send_request("nonexistent", payload) + + assert "not wired" in exc_info.value.message + assert "nonexistent" in exc_info.value.message + + +# ============================================================================= +# Response Message Handling Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestResponseMessageHandling: + """Test _handle_response_message edge cases.""" + + async def test_handle_empty_message_value( + self, + mock_event_bus: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Verify empty message value is logged and skipped.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Create mock message with None value + mock_message = MagicMock() + mock_message.value = None + mock_message.topic = "test.completed" + + # Should not raise, just log warning + await wiring._handle_response_message(instance, mock_message) + + async def test_handle_invalid_json( + self, + mock_event_bus: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Verify invalid JSON is logged and skipped.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Create mock message with invalid JSON + mock_message = MagicMock() + mock_message.value = b"not valid json" + mock_message.topic = "test.completed" + + # Should not raise, just log warning + await wiring._handle_response_message(instance, mock_message) + + async def test_handle_response_missing_correlation_id( + self, + mock_event_bus: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Verify missing correlation_id in response is logged.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + instance = RequestResponseInstanceState( + name="test", + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Create mock message without correlation_id + mock_message = MagicMock() + mock_message.value = json.dumps({"result": "success"}).encode("utf-8") + mock_message.topic = "test.completed" + + # Should not raise, just log warning + await wiring._handle_response_message(instance, mock_message) + + +# ============================================================================= +# Wire Request Response Tests +# ============================================================================= + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestWireRequestResponse: + """Test wire_request_response configuration validation.""" + + async def test_wire_duplicate_instance_raises_config_error( + self, + mock_event_bus: MagicMock, + ) -> None: + """Verify duplicate instance name raises ProtocolConfigurationError.""" + from omnibase_infra.runtime.request_response_wiring import ( + RequestResponseInstanceState, + RequestResponseWiring, + ) + + wiring = RequestResponseWiring( + event_bus=mock_event_bus, + environment="test", + app_name="test-app", + bootstrap_servers="localhost:9092", + ) + + # Pre-register an instance + instance_name = "routing" + wiring._instances[instance_name] = RequestResponseInstanceState( + name=instance_name, + request_topic="test.topic", + completed_topic="test.completed", + failed_topic="test.failed", + timeout_seconds=5, + correlation_config=ModelCorrelationConfig(), + consumer_group="test.group", + ) + + # Try to wire another instance with same name + instance = ModelRequestResponseInstance( + name=instance_name, # Same name + request_topic="onex.cmd.routing.request.v1", + reply_topics=ModelReplyTopics( + completed="onex.evt.routing.completed.v1", + failed="onex.evt.routing.failed.v1", + ), + timeout_seconds=5, + ) + + with pytest.raises(ProtocolConfigurationError) as exc_info: + await wiring._wire_instance(instance) + + assert "already wired" in exc_info.value.message diff --git a/tests/unit/validation/test_validator_defaults.py b/tests/unit/validation/test_validator_defaults.py index a50f90699f..43f9f0a7ac 100644 --- a/tests/unit/validation/test_validator_defaults.py +++ b/tests/unit/validation/test_validator_defaults.py @@ -98,7 +98,7 @@ def test_infra_max_unions_constant(self) -> None: Target: Keep below 150 - if this grows, consider typed patterns from omnibase_core. """ assert INFRA_MAX_UNIONS == 117, ( - "INFRA_MAX_UNIONS should be 115 (non-optional unions only, X | None excluded)" + "INFRA_MAX_UNIONS should be 117 (non-optional unions only, X | None excluded)" ) def test_infra_max_violations_constant(self) -> None: