diff --git a/pyproject.toml b/pyproject.toml index 8dabe0a6d6..170ce3658a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,10 @@ python = "^3.12" # NOTE: Tags are acceptable for development; consider commit SHAs for production deployments. # # omnibase-core v0.6.2: Latest release with updated models and enums. +# Note (OMN-1257): In omnibase_core v0.6.2+, ModelONEXContainer.service_registry +# returns None when enable_service_registry=False was passed to constructor or +# ServiceRegistry module is unavailable. container_wiring.py now raises +# ServiceRegistryUnavailableError with actionable hints instead of cryptic AttributeError. omnibase-core = "^0.6.2" omnibase-spi = "^0.4.1" diff --git a/scripts/dlq_replay.py b/scripts/dlq_replay.py index bd260eba24..d99b5c06db 100644 --- a/scripts/dlq_replay.py +++ b/scripts/dlq_replay.py @@ -47,7 +47,7 @@ from pydantic import BaseModel, Field, ValidationInfo, field_validator from omnibase_infra.dlq import ( - DLQTrackingService, + DLQReplayTracker, EnumReplayStatus, ModelDlqReplayRecord, ModelDlqTrackingConfig, @@ -1023,7 +1023,7 @@ def __init__(self, config: ModelReplayConfig) -> None: self.consumer = DLQConsumer(config) self.producer = DLQProducer(config) self.results: list[ModelReplayResult] = [] - self._tracking_service: DLQTrackingService | None = None + self._tracking_service: DLQReplayTracker | None = None @property def is_tracking_enabled(self) -> bool: @@ -1050,7 +1050,7 @@ async def start(self) -> None: dsn = self.config.build_tracking_dsn() if dsn: tracking_config = ModelDlqTrackingConfig(dsn=dsn) - self._tracking_service = DLQTrackingService(tracking_config) + self._tracking_service = DLQReplayTracker(tracking_config) try: await self._tracking_service.initialize() logger.info("DLQ tracking service initialized") diff --git a/src/omnibase_infra/dlq/__init__.py b/src/omnibase_infra/dlq/__init__.py index 2e041441b4..7cef5c5790 100644 --- a/src/omnibase_infra/dlq/__init__.py +++ b/src/omnibase_infra/dlq/__init__.py @@ -23,7 +23,6 @@ Tracker: - DLQReplayTracker: PostgreSQL tracker for replay history - ServiceDlqTracking: ONEX naming convention alias - - DLQTrackingService: Backwards compatibility alias Example - Recording Replay Attempts: >>> from omnibase_infra.dlq import ( @@ -82,19 +81,15 @@ ServiceDlqTracking, ) -# Backwards compatibility alias -DLQTrackingService = DLQReplayTracker - __all__: list[str] = [ # Constants "PATTERN_TABLE_NAME", # Regex pattern string for table name validation "REGEX_TABLE_NAME", # Pre-compiled regex for runtime validation - # Tracker (multiple names for flexibility) + # Tracker "DLQReplayTracker", # Primary class name - "DLQTrackingService", # Backwards compatibility alias + "ServiceDlqTracking", # ONEX naming convention (service_.py → Service) # Models "EnumReplayStatus", "ModelDlqReplayRecord", "ModelDlqTrackingConfig", - "ServiceDlqTracking", # ONEX naming convention (service_.py → Service) ] diff --git a/src/omnibase_infra/errors/__init__.py b/src/omnibase_infra/errors/__init__.py index 3ba7620a24..7695259700 100644 --- a/src/omnibase_infra/errors/__init__.py +++ b/src/omnibase_infra/errors/__init__.py @@ -91,6 +91,7 @@ ContainerValidationError, ContainerWiringError, ServiceRegistrationError, + ServiceRegistryUnavailableError, ServiceResolutionError, ) from omnibase_infra.errors.error_infra import ( @@ -128,6 +129,7 @@ "RuntimeHostError", "SecretResolutionError", "ServiceRegistrationError", + "ServiceRegistryUnavailableError", "ServiceResolutionError", "UnknownHandlerTypeError", ] diff --git a/src/omnibase_infra/errors/error_container_wiring.py b/src/omnibase_infra/errors/error_container_wiring.py index 9ada5ceb4b..5de9a4c4e3 100644 --- a/src/omnibase_infra/errors/error_container_wiring.py +++ b/src/omnibase_infra/errors/error_container_wiring.py @@ -176,9 +176,69 @@ def __init__( ) +class ServiceRegistryUnavailableError(ContainerValidationError): + """Raised when container.service_registry is None. + + This error indicates that the ModelONEXContainer was initialized + without a service registry, either because: + - enable_service_registry=False was passed + - The ServiceRegistry module is not installed/available + - Container initialization failed silently + + This is a distinct error from generic ContainerValidationError to provide + clearer diagnostics and actionable error messages for service_registry issues. + + Example: + >>> raise ServiceRegistryUnavailableError( + ... "Container service_registry is None", + ... operation="register PolicyRegistry", + ... hint="Check that enable_service_registry=True", + ... ) + """ + + def __init__( + self, + message: str, + *, + operation: str | None = None, + hint: str | None = None, + context: ModelInfraErrorContext | None = None, + **extra_context: object, + ) -> None: + """Initialize ServiceRegistryUnavailableError. + + Args: + message: Primary error message. + operation: The operation that was attempted (e.g., 'register_instance'). + hint: Actionable hint for fixing the issue. + context: Bundled infrastructure context. + **extra_context: Additional context information. + """ + # Build full message with operation and hint + full_message = message + if operation: + full_message = f"{message} (operation: {operation})" + extra_context["operation"] = operation + + default_hint = ( + "Ensure ModelONEXContainer is created with enable_service_registry=True " + "and that the ServiceRegistry module is installed." + ) + actual_hint = hint or default_hint + full_message = f"{full_message}\nHint: {actual_hint}" + extra_context["hint"] = actual_hint + + super().__init__( + message=full_message, + context=context, + **extra_context, + ) + + __all__ = [ "ContainerValidationError", "ContainerWiringError", "ServiceRegistrationError", + "ServiceRegistryUnavailableError", "ServiceResolutionError", ] diff --git a/src/omnibase_infra/event_bus/kafka_event_bus.py b/src/omnibase_infra/event_bus/kafka_event_bus.py index a2407b4679..a768892ca5 100644 --- a/src/omnibase_infra/event_bus/kafka_event_bus.py +++ b/src/omnibase_infra/event_bus/kafka_event_bus.py @@ -133,13 +133,18 @@ Usage: ```python from omnibase_infra.event_bus.kafka_event_bus import KafkaEventBus + from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig # Option 1: Use defaults with environment variable overrides bus = KafkaEventBus.default() await bus.start() - # Option 2: Explicit configuration - bus = KafkaEventBus(bootstrap_servers="kafka:9092", environment="dev") + # Option 2: Explicit configuration via config model + config = ModelKafkaEventBusConfig( + bootstrap_servers="kafka:9092", + environment="dev", + ) + bus = KafkaEventBus(config=config) await bus.start() # Subscribe to a topic @@ -243,7 +248,11 @@ class KafkaEventBus(MixinAsyncCircuitBreaker): Example: ```python - bus = KafkaEventBus(bootstrap_servers="kafka:9092", environment="dev") + config = ModelKafkaEventBusConfig( + bootstrap_servers="kafka:9092", + environment="dev", + ) + bus = KafkaEventBus(config=config) await bus.start() # Subscribe @@ -263,45 +272,12 @@ async def handler(msg): def __init__( self, config: ModelKafkaEventBusConfig | None = None, - # Backwards compatibility parameters (override config if provided) - bootstrap_servers: str | None = None, - environment: str | None = None, - group: str | None = None, - timeout_seconds: int | None = None, - max_retry_attempts: int | None = None, - retry_backoff_base: float | None = None, - circuit_breaker_threshold: int | None = None, - circuit_breaker_reset_timeout: float | None = None, ) -> None: """Initialize the Kafka event bus. - Design Note: - This __init__ method intentionally accepts 10 parameters (1 config + 8 overrides + self) - to maintain backwards compatibility while transitioning to config-driven initialization. - The parameters follow the ONEX config pattern: - - Primary: `config` parameter (ModelKafkaEventBusConfig) for modern usage - - Overrides: 8 optional parameters for backwards compatibility - - This approach allows gradual migration from direct parameters to config objects - without breaking existing code. Recommended usage is factory methods: - - KafkaEventBus.default() for defaults with env var overrides - - KafkaEventBus.from_config(config) for config-driven initialization - - KafkaEventBus.from_yaml(path) for YAML-based configuration - - The parameter count is acceptable for infrastructure components supporting - multiple initialization patterns during deprecation periods. - Args: config: Configuration model containing all settings. If not provided, defaults are used with environment variable overrides. - bootstrap_servers: Override bootstrap servers from config - environment: Override environment identifier from config - group: Override consumer group identifier from config - timeout_seconds: Override timeout from config - max_retry_attempts: Override max retry attempts from config - retry_backoff_base: Override retry backoff base from config - circuit_breaker_threshold: Override circuit breaker threshold from config - circuit_breaker_reset_timeout: Override circuit breaker reset timeout from config Raises: ProtocolConfigurationError: If circuit_breaker_threshold is not a positive integer @@ -318,9 +294,6 @@ def __init__( # Using factory methods bus = KafkaEventBus.default() bus = KafkaEventBus.from_yaml(Path("kafka.yaml")) - - # Backwards compatible direct parameters - bus = KafkaEventBus(bootstrap_servers="kafka:9092", environment="dev") ``` """ # Use provided config or create default with environment overrides @@ -330,37 +303,16 @@ def __init__( # Store config reference self._config = config - # Apply parameter overrides for backwards compatibility - self._bootstrap_servers = ( - bootstrap_servers - if bootstrap_servers is not None - else config.bootstrap_servers - ) - self._environment = ( - environment if environment is not None else config.environment - ) - self._group = group if group is not None else config.group - self._timeout_seconds = ( - timeout_seconds if timeout_seconds is not None else config.timeout_seconds - ) - self._max_retry_attempts = ( - max_retry_attempts - if max_retry_attempts is not None - else config.max_retry_attempts - ) - self._retry_backoff_base = ( - retry_backoff_base - if retry_backoff_base is not None - else config.retry_backoff_base - ) + # Apply config values + self._bootstrap_servers = config.bootstrap_servers + self._environment = config.environment + self._group = config.group + self._timeout_seconds = config.timeout_seconds + self._max_retry_attempts = config.max_retry_attempts + self._retry_backoff_base = config.retry_backoff_base - # Circuit breaker configuration with override support - effective_threshold = ( - circuit_breaker_threshold - if circuit_breaker_threshold is not None - else config.circuit_breaker_threshold - ) - if effective_threshold < 1: + # Circuit breaker configuration + if config.circuit_breaker_threshold < 1: context = ModelInfraErrorContext( transport_type=EnumInfraTransportType.KAFKA, operation="init", @@ -368,21 +320,16 @@ def __init__( correlation_id=uuid4(), ) raise ProtocolConfigurationError( - f"circuit_breaker_threshold must be a positive integer, got {effective_threshold}", + f"circuit_breaker_threshold must be a positive integer, got {config.circuit_breaker_threshold}", context=context, parameter="circuit_breaker_threshold", - value=effective_threshold, + value=config.circuit_breaker_threshold, ) - effective_reset_timeout = ( - circuit_breaker_reset_timeout - if circuit_breaker_reset_timeout is not None - else config.circuit_breaker_reset_timeout - ) # Initialize circuit breaker mixin self._init_circuit_breaker( - threshold=effective_threshold, - reset_timeout=effective_reset_timeout, + threshold=config.circuit_breaker_threshold, + reset_timeout=config.circuit_breaker_reset_timeout, service_name=f"kafka.{self._environment}", transport_type=EnumInfraTransportType.KAFKA, ) diff --git a/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py b/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py index f502066d25..12e8dd5cd8 100644 --- a/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py +++ b/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py @@ -189,7 +189,7 @@ class ModelKafkaEventBusConfig(BaseModel): retry_backoff_base: float = Field( default=1.0, description="Base delay in seconds for exponential backoff", - ge=0.1, + ge=0.001, # Allow very short backoffs for testing (minimum 1ms) le=60.0, ) @@ -203,7 +203,7 @@ class ModelKafkaEventBusConfig(BaseModel): circuit_breaker_reset_timeout: float = Field( default=30.0, description="Seconds before circuit breaker resets to half-open state", - ge=1.0, + ge=0.01, # Allow short timeouts for testing (minimum 10ms) le=3600.0, ) diff --git a/src/omnibase_infra/idempotency/models/__init__.py b/src/omnibase_infra/idempotency/models/__init__.py index f6004a9040..a6ea523f1f 100644 --- a/src/omnibase_infra/idempotency/models/__init__.py +++ b/src/omnibase_infra/idempotency/models/__init__.py @@ -7,17 +7,11 @@ Exports: ModelIdempotencyStoreHealthCheckResult: Result of an idempotency store health check - ModelHealthCheckResult: Backward compatibility alias for ModelIdempotencyStoreHealthCheckResult ModelIdempotencyRecord: Represents a stored idempotency record ModelIdempotencyCheckResult: Result of an idempotency check operation ModelIdempotencyGuardConfig: Configuration for idempotency guard middleware ModelIdempotencyStoreMetrics: Metrics for store observability ModelPostgresIdempotencyStoreConfig: Configuration for PostgreSQL store - -.. versionchanged:: 0.7.0 - Renamed ModelHealthCheckResult to ModelIdempotencyStoreHealthCheckResult - to avoid naming conflicts with runtime.models.ModelHealthCheckResult. - The old name is preserved as a backward compatibility alias. """ from omnibase_infra.idempotency.models.model_idempotency_check_result import ( @@ -30,7 +24,6 @@ ModelIdempotencyRecord, ) from omnibase_infra.idempotency.models.model_idempotency_store_health_check_result import ( - ModelHealthCheckResult, # Backward compatibility alias ModelIdempotencyStoreHealthCheckResult, ) from omnibase_infra.idempotency.models.model_idempotency_store_metrics import ( @@ -41,7 +34,6 @@ ) __all__ = [ - "ModelHealthCheckResult", # Backward compatibility alias "ModelIdempotencyCheckResult", "ModelIdempotencyGuardConfig", "ModelIdempotencyRecord", diff --git a/src/omnibase_infra/idempotency/models/model_health_check_result.py b/src/omnibase_infra/idempotency/models/model_health_check_result.py deleted file mode 100644 index ebc4d43fe7..0000000000 --- a/src/omnibase_infra/idempotency/models/model_health_check_result.py +++ /dev/null @@ -1,34 +0,0 @@ -# SPDX-License-Identifier: MIT -# Copyright (c) 2025 OmniNode Team -"""Deprecated: Health Check Result Model. - -.. deprecated:: 0.7.0 - This module is deprecated. Use - ``omnibase_infra.idempotency.models.model_idempotency_store_health_check_result`` - instead. This module is preserved for backward compatibility only. - -The model has been renamed to ``ModelIdempotencyStoreHealthCheckResult`` to avoid -naming conflicts with ``omnibase_infra.runtime.models.ModelHealthCheckResult``. -""" - -from __future__ import annotations - -import warnings - -from omnibase_infra.idempotency.models.model_idempotency_store_health_check_result import ( - ModelIdempotencyStoreHealthCheckResult, -) - -# Issue deprecation warning on module import -warnings.warn( - "omnibase_infra.idempotency.models.model_health_check_result is deprecated. " - "Use omnibase_infra.idempotency.models.model_idempotency_store_health_check_result " - "or import ModelIdempotencyStoreHealthCheckResult from omnibase_infra.idempotency.models.", - DeprecationWarning, - stacklevel=2, -) - -# Backward compatibility alias -ModelHealthCheckResult = ModelIdempotencyStoreHealthCheckResult - -__all__: list[str] = ["ModelHealthCheckResult"] diff --git a/src/omnibase_infra/idempotency/models/model_idempotency_store_health_check_result.py b/src/omnibase_infra/idempotency/models/model_idempotency_store_health_check_result.py index 08ea73b673..f34d92aa81 100644 --- a/src/omnibase_infra/idempotency/models/model_idempotency_store_health_check_result.py +++ b/src/omnibase_infra/idempotency/models/model_idempotency_store_health_check_result.py @@ -11,10 +11,6 @@ type for the `reason` field to constrain valid status values. For general-purpose handler health checks, see ``omnibase_infra.runtime.models.ModelHealthCheckResult``. - -.. versionadded:: 0.7.0 - Renamed from ModelHealthCheckResult to avoid naming conflicts - with runtime.models.ModelHealthCheckResult (PR #91 feedback). """ from __future__ import annotations @@ -80,10 +76,6 @@ class ModelIdempotencyStoreHealthCheckResult(BaseModel): ) -# Backward compatibility alias -ModelHealthCheckResult = ModelIdempotencyStoreHealthCheckResult - __all__: list[str] = [ - "ModelHealthCheckResult", # Backward compatibility "ModelIdempotencyStoreHealthCheckResult", ] diff --git a/src/omnibase_infra/mixins/mixin_node_introspection.py b/src/omnibase_infra/mixins/mixin_node_introspection.py index 7c5466f41c..7501f79700 100644 --- a/src/omnibase_infra/mixins/mixin_node_introspection.py +++ b/src/omnibase_infra/mixins/mixin_node_introspection.py @@ -231,20 +231,12 @@ async def shutdown(self): HEARTBEAT_TOPIC = "node.heartbeat" REQUEST_INTROSPECTION_TOPIC = "node.request_introspection" -# Backward-compatible alias for CapabilitiesTypedDict -# The canonical definition is in model_node_introspection_event.py -# This provides a shorter name for internal use and maintains backward compatibility -CapabilitiesDict = CapabilitiesTypedDict - # Performance threshold constants (in milliseconds) PERF_THRESHOLD_GET_CAPABILITIES_MS = 50.0 PERF_THRESHOLD_DISCOVER_CAPABILITIES_MS = 30.0 PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS = 50.0 PERF_THRESHOLD_CACHE_HIT_MS = 1.0 -# Backwards compatibility alias (ModelIntrospectionPerformanceMetrics imported from models.discovery) -IntrospectionPerformanceMetrics = ModelIntrospectionPerformanceMetrics - class PerformanceMetricsCacheDict(TypedDict, total=False): """TypedDict for JSON-serialized ModelIntrospectionPerformanceMetrics. @@ -451,7 +443,7 @@ def _internal_helper(self, data: dict) -> dict: # Performance metrics tracking (instance-level) # Stores the most recent performance metrics from introspection operations - _introspection_last_metrics: IntrospectionPerformanceMetrics | None + _introspection_last_metrics: ModelIntrospectionPerformanceMetrics | None # Active operations tracking (instance-level) # Thread-safe counter for tracking concurrent operations @@ -672,24 +664,6 @@ def __init__(self, node_config): }, ) - def initialize_introspection_from_config( - self, - config: ModelIntrospectionConfig, - ) -> None: - """Alias for initialize_introspection for backward compatibility. - - This method is provided for backward compatibility with code that - uses the more explicit name. New code should use - ``initialize_introspection()`` directly. - - Args: - config: Configuration model containing all introspection settings. - - See Also: - initialize_introspection: The canonical initialization method. - """ - self.initialize_introspection(config) - def _ensure_initialized(self) -> None: """Ensure introspection has been initialized. @@ -968,7 +942,7 @@ async def _get_state_from_method(self) -> str | None: ) return None - async def get_capabilities(self) -> CapabilitiesDict: + async def get_capabilities(self) -> CapabilitiesTypedDict: """Extract node capabilities via reflection. Uses the inspect module to discover: @@ -1061,7 +1035,7 @@ async def get_capabilities(self) -> CapabilitiesDict: operations.append(name) # Build capabilities dict - capabilities: CapabilitiesDict = { + capabilities: CapabilitiesTypedDict = { "operations": operations, "protocols": self._discover_protocols(), "has_fsm": self._has_fsm_state(), @@ -1236,7 +1210,7 @@ async def get_introspection_data(self) -> ModelNodeIntrospectionEvent: slow_operations.append("cache_hit") # Create frozen metrics object with final values - metrics = IntrospectionPerformanceMetrics( + metrics = ModelIntrospectionPerformanceMetrics( total_introspection_ms=elapsed_ms, cache_hit=True, threshold_exceeded=threshold_exceeded, @@ -1318,7 +1292,7 @@ async def get_introspection_data(self) -> ModelNodeIntrospectionEvent: slow_operations.append("total_introspection") # Create frozen metrics object with final values - metrics = IntrospectionPerformanceMetrics( + metrics = ModelIntrospectionPerformanceMetrics( get_capabilities_ms=get_capabilities_ms, discover_capabilities_ms=discover_capabilities_ms, get_endpoints_ms=get_endpoints_ms, @@ -2216,7 +2190,7 @@ def invalidate_introspection_cache(self) -> None: extra={"node_id": self._introspection_node_id}, ) - def get_performance_metrics(self) -> IntrospectionPerformanceMetrics | None: + def get_performance_metrics(self) -> ModelIntrospectionPerformanceMetrics | None: """Get the most recent performance metrics from introspection operations. Returns the performance metrics captured during the last call to @@ -2224,7 +2198,7 @@ def get_performance_metrics(self) -> IntrospectionPerformanceMetrics | None: performance and detect when operations exceed the <50ms threshold. Returns: - IntrospectionPerformanceMetrics if introspection has been called, + ModelIntrospectionPerformanceMetrics if introspection has been called, None if no introspection has been performed yet. Example: @@ -2417,10 +2391,9 @@ async def get_active_operations_count(self) -> int: "PERF_THRESHOLD_GET_CAPABILITIES_MS", "PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS", "REQUEST_INTROSPECTION_TOPIC", - "CapabilitiesDict", # Backward-compatible alias for CapabilitiesTypedDict "CapabilitiesTypedDict", # Re-export from model for convenience "IntrospectionCacheDict", - "IntrospectionPerformanceMetrics", "MixinNodeIntrospection", + "ModelIntrospectionPerformanceMetrics", "PerformanceMetricsCacheDict", # TypedDict for cached performance metrics ] diff --git a/src/omnibase_infra/models/lifecycle/__init__.py b/src/omnibase_infra/models/lifecycle/__init__.py index 8f7e2bd126..c930e1a188 100644 --- a/src/omnibase_infra/models/lifecycle/__init__.py +++ b/src/omnibase_infra/models/lifecycle/__init__.py @@ -32,12 +32,8 @@ ) from omnibase_infra.runtime.models.model_lifecycle_result import ModelLifecycleResult -# Alias for backward compatibility with code importing ModelHandlerHealthCheckResult -ModelHandlerHealthCheckResult = ModelHealthCheckResult - __all__ = [ "ModelBatchLifecycleResult", - "ModelHandlerHealthCheckResult", # Backward compatibility alias "ModelHealthCheckResult", "ModelLifecycleResult", ] diff --git a/src/omnibase_infra/nodes/reducers/registration_reducer.py b/src/omnibase_infra/nodes/reducers/registration_reducer.py index 25926af27c..a15d6b5686 100644 --- a/src/omnibase_infra/nodes/reducers/registration_reducer.py +++ b/src/omnibase_infra/nodes/reducers/registration_reducer.py @@ -500,10 +500,6 @@ def failure( ) -# Backwards compatibility alias -ValidationResult = ModelValidationResult - - # TODO(OMN-889): Complete pure reducer implementation - add reduce_confirmation() method class RegistrationReducer: """Pure reducer for node registration workflow. @@ -723,7 +719,9 @@ def _is_valid(self, event: ModelNodeIntrospectionEvent) -> bool: """ return self._validate_event(event).is_valid - def _validate_event(self, event: ModelNodeIntrospectionEvent) -> ValidationResult: + def _validate_event( + self, event: ModelNodeIntrospectionEvent + ) -> ModelValidationResult: """Validate introspection event with detailed error information. Validates that required fields are present for registration workflow. @@ -753,7 +751,7 @@ def _validate_event(self, event: ModelNodeIntrospectionEvent) -> ValidationResul """ # Validate node_id: required for registration identity if event.node_id is None: - return ValidationResult.failure( + return ModelValidationResult.failure( error_code="missing_node_id", field_name="node_id", error_message="node_id is required for registration identity", @@ -761,7 +759,7 @@ def _validate_event(self, event: ModelNodeIntrospectionEvent) -> ValidationResul # Validate node_type: must be present if not hasattr(event, "node_type") or event.node_type is None: - return ValidationResult.failure( + return ModelValidationResult.failure( error_code="missing_node_type", field_name="node_type", error_message="node_type is required for service categorization", @@ -776,7 +774,7 @@ def _validate_event(self, event: ModelNodeIntrospectionEvent) -> ValidationResul EnumNodeKind.ORCHESTRATOR.value, } if event.node_type not in valid_node_types: - return ValidationResult.failure( + return ModelValidationResult.failure( error_code="invalid_node_type", field_name="node_type", error_message=( @@ -784,7 +782,7 @@ def _validate_event(self, event: ModelNodeIntrospectionEvent) -> ValidationResul ), ) - return ValidationResult.success() + return ModelValidationResult.success() def _derive_deterministic_event_id( self, event: ModelNodeIntrospectionEvent @@ -1180,8 +1178,7 @@ def _build_output( # Performance threshold constants (for tests and monitoring) "PERF_THRESHOLD_REDUCE_MS", # Validation types (for tests and custom validators) - "ModelValidationResult", # Canonical ONEX name (Model* convention) + "ModelValidationResult", "RegistrationReducer", "ValidationErrorCode", - "ValidationResult", # Backwards compatibility alias ] diff --git a/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py b/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py index b6042252bc..e588b60a0a 100644 --- a/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py +++ b/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py @@ -12,9 +12,9 @@ from omnibase_infra.plugins.plugin_compute_base import PluginComputeBase from omnibase_infra.protocols.protocol_plugin_compute import ( - PluginContext, - PluginInputData, - PluginOutputData, + ModelPluginContext, + ModelPluginInputData, + ModelPluginOutputData, ) @@ -51,8 +51,8 @@ class PluginJsonNormalizer(PluginComputeBase): MAX_RECURSION_DEPTH: int = 100 def execute( - self, input_data: PluginInputData, context: PluginContext - ) -> PluginOutputData: + self, input_data: ModelPluginInputData, context: ModelPluginContext + ) -> ModelPluginOutputData: """Execute JSON normalization with type-safe inputs and outputs. Args: @@ -82,7 +82,7 @@ def execute( json_data, _max_depth=effective_max_depth ) output: JsonNormalizerOutput = {"normalized": normalized} - return cast(PluginOutputData, output) + return cast(ModelPluginOutputData, output) except RecursionError as e: raise OnexError( @@ -177,7 +177,7 @@ def _sort_keys_recursively( for item in obj ] - def validate_input(self, input_data: PluginInputData) -> None: + def validate_input(self, input_data: ModelPluginInputData) -> None: """Validate input with runtime type checking and type guards. This method validates input structure before execution. By protocol design, diff --git a/src/omnibase_infra/plugins/examples/plugin_json_normalizer_error_handling.py b/src/omnibase_infra/plugins/examples/plugin_json_normalizer_error_handling.py index 18c65f608a..840e7aa7ec 100644 --- a/src/omnibase_infra/plugins/examples/plugin_json_normalizer_error_handling.py +++ b/src/omnibase_infra/plugins/examples/plugin_json_normalizer_error_handling.py @@ -25,9 +25,9 @@ from omnibase_infra.plugins.plugin_compute_base import PluginComputeBase from omnibase_infra.protocols.protocol_plugin_compute import ( - PluginContext, - PluginInputData, - PluginOutputData, + ModelPluginContext, + ModelPluginInputData, + ModelPluginOutputData, ) @@ -41,8 +41,8 @@ class PluginJsonNormalizerErrorHandling(PluginComputeBase): __slots__ = () # Enforce statelessness - no instance attributes def execute( - self, input_data: PluginInputData, context: PluginContext - ) -> PluginOutputData: + self, input_data: ModelPluginInputData, context: ModelPluginContext + ) -> ModelPluginOutputData: """Execute JSON normalization with comprehensive error handling. Args: @@ -89,7 +89,7 @@ def execute( # Return result with correlation_id for tracing return cast( - PluginOutputData, + ModelPluginOutputData, { "normalized": normalized, "correlation_id": correlation_id, @@ -162,7 +162,7 @@ def _sort_keys_recursively(self, obj: JsonType) -> JsonType: # Primitive values (str, int, float, bool, None) returned unchanged return obj - def validate_input(self, input_data: PluginInputData) -> None: + def validate_input(self, input_data: ModelPluginInputData) -> None: """Validate that input contains JSON-compatible data. Args: diff --git a/src/omnibase_infra/plugins/plugin_compute_base.py b/src/omnibase_infra/plugins/plugin_compute_base.py index c9a60ab188..6bffefbcd3 100644 --- a/src/omnibase_infra/plugins/plugin_compute_base.py +++ b/src/omnibase_infra/plugins/plugin_compute_base.py @@ -82,16 +82,16 @@ ```python from omnibase_infra.plugins import PluginComputeBase from omnibase_infra.protocols.protocol_plugin_compute import ( - PluginInputData, - PluginContext, - PluginOutputData, + ModelPluginInputData, + ModelPluginContext, + ModelPluginOutputData, ) # Step 1: Implement plugin (pure computation) class DataValidatorPlugin(PluginComputeBase): def execute( - self, input_data: PluginInputData, context: PluginContext - ) -> PluginOutputData: + self, input_data: ModelPluginInputData, context: ModelPluginContext + ) -> ModelPluginOutputData: # Pure validation logic (no I/O) is_valid = self._validate_schema(input_data) return { @@ -99,7 +99,7 @@ def execute( "errors": [] if is_valid else self._get_errors(input_data), } - def validate_input(self, input_data: PluginInputData) -> None: + def validate_input(self, input_data: ModelPluginInputData) -> None: if "schema" not in input_data: raise ValueError("schema field required") @@ -165,11 +165,6 @@ async def execute(self, input_model: ModelInput) -> ModelOutput: from omnibase_infra.plugins.models.model_plugin_input_data import ModelPluginInputData from omnibase_infra.plugins.models.model_plugin_output_data import ModelPluginOutputData -# Type aliases for backwards compatibility with protocol imports -PluginContext = ModelPluginContext -PluginInputData = ModelPluginInputData -PluginOutputData = ModelPluginOutputData - class PluginComputeBase(ABC): """Abstract base class for compute plugins. @@ -226,15 +221,15 @@ class PluginComputeBase(ABC): Example: ```python from omnibase_infra.protocols.protocol_plugin_compute import ( - PluginInputData, - PluginContext, - PluginOutputData, + ModelPluginInputData, + ModelPluginContext, + ModelPluginOutputData, ) class MyComputePlugin(PluginComputeBase): def execute( - self, input_data: PluginInputData, context: PluginContext - ) -> PluginOutputData: + self, input_data: ModelPluginInputData, context: ModelPluginContext + ) -> ModelPluginOutputData: # Handle edge cases if not input_data: return {"result": None, "warning": "Empty input"} @@ -252,7 +247,7 @@ def execute( result = self._process(input_data) return {"result": result} - def validate_input(self, input_data: PluginInputData) -> None: + def validate_input(self, input_data: ModelPluginInputData) -> None: # Optional: Validate required fields upfront if "required_field" not in input_data: raise ValueError("Missing required_field") @@ -267,8 +262,8 @@ def validate_input(self, input_data: PluginInputData) -> None: @abstractmethod def execute( - self, input_data: PluginInputData, context: PluginContext - ) -> PluginOutputData: + self, input_data: ModelPluginInputData, context: ModelPluginContext + ) -> ModelPluginOutputData: """Execute computation. MUST be deterministic. Given the same input_data and context, this method MUST return @@ -368,14 +363,14 @@ def execute( Example - Handling Edge Cases: ```python from omnibase_infra.protocols.protocol_plugin_compute import ( - PluginInputData, - PluginContext, - PluginOutputData, + ModelPluginInputData, + ModelPluginContext, + ModelPluginOutputData, ) def execute( - self, input_data: PluginInputData, context: PluginContext - ) -> PluginOutputData: + self, input_data: ModelPluginInputData, context: ModelPluginContext + ) -> ModelPluginOutputData: # Edge Case 1 & 2: Handle None/empty inputs if not input_data: return {"result": None, "warning": "Empty input"} @@ -411,7 +406,7 @@ def execute( """ ... - def validate_input(self, input_data: PluginInputData) -> None: + def validate_input(self, input_data: ModelPluginInputData) -> None: """Optional input validation hook. Override this method to validate input_data before execution. @@ -425,7 +420,7 @@ def validate_input(self, input_data: PluginInputData) -> None: """ return # Default: no validation - def validate_output(self, output: PluginOutputData) -> None: + def validate_output(self, output: ModelPluginOutputData) -> None: """Optional output validation hook. Override this method to validate computation results after execution. diff --git a/src/omnibase_infra/protocols/protocol_plugin_compute.py b/src/omnibase_infra/protocols/protocol_plugin_compute.py index 3d30231ff4..197d3a760a 100644 --- a/src/omnibase_infra/protocols/protocol_plugin_compute.py +++ b/src/omnibase_infra/protocols/protocol_plugin_compute.py @@ -52,13 +52,13 @@ @runtime_checkable class ProtocolPluginCompute(Protocol): - def execute(self, input_data: PluginInputData, context: PluginContext) -> PluginOutputData: + def execute(self, input_data: ModelPluginInputData, context: ModelPluginContext) -> ModelPluginOutputData: '''Execute deterministic computation.''' ... # Example plugin implementation class JsonSchemaValidator: - def execute(self, input_data: PluginInputData, context: PluginContext) -> PluginOutputData: + def execute(self, input_data: ModelPluginInputData, context: ModelPluginContext) -> ModelPluginOutputData: '''Validate JSON data against schema.''' schema = context.get("schema", {}) data = input_data.get("data", {}) @@ -116,38 +116,9 @@ def _get_errors(self, data: dict[str, object], schema: dict[str, object]) -> lis "ModelPluginContext", "ModelPluginInputData", "ModelPluginOutputData", - "PluginContext", - "PluginInputData", - "PluginOutputData", "ProtocolPluginCompute", ] -# Type aliases for backwards compatibility -# These map the legacy TypedDict names to their Pydantic model replacements -PluginInputData = ModelPluginInputData -"""Type alias for backwards compatibility. - -Deprecated: Use ModelPluginInputData directly for new code. -This alias maps to ModelPluginInputData Pydantic model which replaces -the former TypedDict definition. -""" - -PluginContext = ModelPluginContext -"""Type alias for backwards compatibility. - -Deprecated: Use ModelPluginContext directly for new code. -This alias maps to ModelPluginContext Pydantic model which replaces -the former TypedDict definition. -""" - -PluginOutputData = ModelPluginOutputData -"""Type alias for backwards compatibility. - -Deprecated: Use ModelPluginOutputData directly for new code. -This alias maps to ModelPluginOutputData Pydantic model which replaces -the former TypedDict definition. -""" - @runtime_checkable class ProtocolPluginCompute(Protocol): @@ -171,7 +142,7 @@ class ProtocolPluginCompute(Protocol): Example: ```python class DataNormalizer: - def execute(self, input_data: PluginInputData, context: PluginContext) -> PluginOutputData: + def execute(self, input_data: ModelPluginInputData, context: ModelPluginContext) -> ModelPluginOutputData: '''Normalize numeric data to [0, 1] range.''' values: list[float] = input_data.get("values", []) min_val: float = context.get("min_value", 0.0) @@ -192,8 +163,8 @@ def execute(self, input_data: PluginInputData, context: PluginContext) -> Plugin """ def execute( - self, input_data: PluginInputData, context: PluginContext - ) -> PluginOutputData: + self, input_data: ModelPluginInputData, context: ModelPluginContext + ) -> ModelPluginOutputData: """Execute deterministic computation on input data. This method must be deterministic: given the same input_data and context, @@ -231,7 +202,7 @@ def execute( Example - Input Validation Error: ```python - def execute(self, input_data: PluginInputData, context: PluginContext) -> PluginOutputData: + def execute(self, input_data: ModelPluginInputData, context: ModelPluginContext) -> ModelPluginOutputData: from omnibase_core.errors import OnexError from omnibase_core.enums import CoreErrorCode @@ -257,7 +228,7 @@ def execute(self, input_data: PluginInputData, context: PluginContext) -> Plugin Example - Computation Error with Context: ```python - def execute(self, input_data: PluginInputData, context: PluginContext) -> PluginOutputData: + def execute(self, input_data: ModelPluginInputData, context: ModelPluginContext) -> ModelPluginOutputData: from omnibase_core.errors import OnexError from omnibase_core.enums import CoreErrorCode @@ -295,7 +266,7 @@ def execute(self, input_data: PluginInputData, context: PluginContext) -> Plugin Example - Type Validation Error: ```python - def execute(self, input_data: PluginInputData, context: PluginContext) -> PluginOutputData: + def execute(self, input_data: ModelPluginInputData, context: ModelPluginContext) -> ModelPluginOutputData: from omnibase_core.errors import OnexError from omnibase_core.enums import CoreErrorCode @@ -327,7 +298,7 @@ def execute(self, input_data: PluginInputData, context: PluginContext) -> Plugin Example - Fallback Values: ```python - def execute(self, input_data: PluginInputData, context: PluginContext) -> PluginOutputData: + def execute(self, input_data: ModelPluginInputData, context: ModelPluginContext) -> ModelPluginOutputData: from omnibase_core.errors import OnexError from omnibase_core.enums import CoreErrorCode diff --git a/src/omnibase_infra/runtime/__init__.py b/src/omnibase_infra/runtime/__init__.py index 3bacb37a4d..10c3a07ee3 100644 --- a/src/omnibase_infra/runtime/__init__.py +++ b/src/omnibase_infra/runtime/__init__.py @@ -2,16 +2,6 @@ # Copyright (c) 2025 OmniNode Team """Runtime module for omnibase_infra. -Migration Note (OMN-934): - This module renames "Handler" terminology to "Dispatcher" for message routing. - Legacy aliases are provided for backwards compatibility in MessageDispatchEngine: - - - register_handler() -> register_dispatcher() (method alias provided) - - handler_count -> dispatcher_count (property alias provided) - - get_handler_metrics() -> get_dispatcher_metrics() (method alias provided) - - Note: DispatcherRegistry is a NEW class (no HandlerRegistry predecessor existed). - This module provides the runtime infrastructure for the ONEX infrastructure layer, including three SINGLE SOURCE OF TRUTH registries and the runtime execution host. diff --git a/src/omnibase_infra/runtime/container_wiring.py b/src/omnibase_infra/runtime/container_wiring.py index f2f38b3970..54c6082c62 100644 --- a/src/omnibase_infra/runtime/container_wiring.py +++ b/src/omnibase_infra/runtime/container_wiring.py @@ -79,44 +79,53 @@ ) from omnibase_infra.runtime.message_dispatch_engine import MessageDispatchEngine -logger = logging.getLogger(__name__) - +from omnibase_infra.errors import ServiceRegistryUnavailableError -class ServiceRegistryUnavailableError(RuntimeError): - """Raised when container.service_registry is None due to circular import.""" +logger = logging.getLogger(__name__) -def _check_service_registry_available( +def _validate_service_registry( container: ModelONEXContainer, operation: str, ) -> None: - """Check if container.service_registry is available and raise clear error if not. + """Validate that container.service_registry is not None. - This function provides an early, clear error message when the service_registry - is None, which can happen due to circular import issues in omnibase_core. + This validation should be called before any operation that uses + container.service_registry to provide clear error messages when + the service registry is unavailable. Args: - container: The ONEX container to check. - operation: Name of the operation being attempted (for error message). + container: The ONEX container to validate. + operation: Description of the operation being attempted. Raises: - ServiceRegistryUnavailableError: If service_registry is None, with a clear - message explaining the circular import issue and remediation steps. + ServiceRegistryUnavailableError: If service_registry is None. + + Example: + >>> _validate_service_registry(container, "register PolicyRegistry") + >>> # Proceed with registration... """ + if not hasattr(container, "service_registry"): + raise ServiceRegistryUnavailableError( + "Container missing 'service_registry' attribute", + operation=operation, + hint=( + "Expected ModelONEXContainer from omnibase_core. " + "Check that omnibase_core is properly installed." + ), + ) + if container.service_registry is None: raise ServiceRegistryUnavailableError( - f"Cannot {operation}: container.service_registry is None.\n\n" - f"Root Cause:\n" - f" This is likely due to a circular import bug in omnibase_core.\n" - f" The ModelONEXContainer was instantiated before its ServiceRegistry\n" - f" dependency was fully initialized.\n\n" - f"How to Fix:\n" - f" 1. Upgrade omnibase_core to a version >= 0.6.3 that fixes circular imports\n" - f" 2. File a bug at https://github.com/omninode/omnibase_core/issues if not already reported\n\n" - f"Workaround for Tests:\n" - f" - Use a mock container with a properly initialized service_registry\n" - f" - Or skip container wiring for unit tests that don't require service resolution\n" - f" - Example: container.service_registry = MockServiceRegistry()" + "Container service_registry is None", + operation=operation, + hint=( + "ModelONEXContainer.service_registry returns None when:\n" + " 1. enable_service_registry=False was passed to constructor\n" + " 2. ServiceRegistry module is not available/installed\n" + " 3. Container initialization encountered an import error\n" + "Check container logs for 'ServiceRegistry not available' warnings." + ), ) @@ -126,6 +135,10 @@ def _analyze_attribute_error(error_str: str) -> tuple[str, str]: Extracts the missing attribute name from the error string and provides a user-friendly hint for common container API issues. + Note: service_registry missing/None cases are handled by _validate_service_registry() + which is called before operations. This function handles other AttributeErrors + (e.g., missing register_instance method). + Args: error_str: The string representation of the AttributeError. @@ -134,12 +147,7 @@ def _analyze_attribute_error(error_str: str) -> tuple[str, str]: """ missing_attr = error_str.split("'")[-2] if "'" in error_str else "unknown" - if "service_registry" in error_str: - hint = ( - "Container missing 'service_registry' attribute. " - "Expected ModelONEXContainer from omnibase_core." - ) - elif "register_instance" in error_str: + if "register_instance" in error_str: hint = ( "Container.service_registry missing 'register_instance' method. " "Check omnibase_core version compatibility (requires v0.5.6 or later)." @@ -224,8 +232,8 @@ async def wire_infrastructure_services( >>> hasattr(compute_reg, 'register_plugin') and callable(compute_reg.register_plugin) True """ - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "wire infrastructure services") + # Validate service_registry is available before attempting registration + _validate_service_registry(container, "wire_infrastructure_services") services_registered: list[str] = [] @@ -373,8 +381,8 @@ async def get_policy_registry_from_container( wire_infrastructure_services(). If not, it will raise RuntimeError. For auto-registration, use get_or_create_policy_registry() instead. """ - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "resolve PolicyRegistry") + # Validate service_registry is available + _validate_service_registry(container, "resolve PolicyRegistry") try: registry: PolicyRegistry = await container.service_registry.resolve_service( @@ -382,13 +390,10 @@ async def get_policy_registry_from_container( ) return registry except AttributeError as e: + # Note: service_registry case is now handled by _validate_service_registry + # This block handles other AttributeErrors like missing resolve_service error_str = str(e) - if "service_registry" in error_str: - hint = ( - "Container missing 'service_registry' attribute. " - "Expected ModelONEXContainer from omnibase_core." - ) - elif "resolve_service" in error_str: + if "resolve_service" in error_str: hint = ( "Container.service_registry missing 'resolve_service' method. " "Check omnibase_core version compatibility (requires v0.5.6 or later)." @@ -465,8 +470,8 @@ async def get_or_create_policy_registry( wire_infrastructure_services() for production code to ensure proper initialization order and error handling. """ - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "get or create PolicyRegistry") + # Validate service_registry is available + _validate_service_registry(container, "get_or_create PolicyRegistry") try: # Try to resolve existing PolicyRegistry @@ -536,8 +541,8 @@ async def get_handler_registry_from_container( This function assumes ProtocolBindingRegistry was registered via wire_infrastructure_services(). If not, it will raise RuntimeError. """ - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "resolve ProtocolBindingRegistry") + # Validate service_registry is available + _validate_service_registry(container, "resolve ProtocolBindingRegistry") try: registry: ProtocolBindingRegistry = ( @@ -545,13 +550,10 @@ async def get_handler_registry_from_container( ) return registry except AttributeError as e: + # Note: service_registry case is now handled by _validate_service_registry + # This block handles other AttributeErrors like missing resolve_service error_str = str(e) - if "service_registry" in error_str: - hint = ( - "Container missing 'service_registry' attribute. " - "Expected ModelONEXContainer from omnibase_core." - ) - elif "resolve_service" in error_str: + if "resolve_service" in error_str: hint = ( "Container.service_registry missing 'resolve_service' method. " "Check omnibase_core version compatibility (requires v0.5.6 or later)." @@ -624,8 +626,8 @@ async def get_compute_registry_from_container( wire_infrastructure_services(). If not, it will raise RuntimeError. For auto-registration, use get_or_create_compute_registry() instead. """ - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "resolve RegistryCompute") + # Validate service_registry is available + _validate_service_registry(container, "resolve RegistryCompute") try: registry: RegistryCompute = await container.service_registry.resolve_service( @@ -633,13 +635,10 @@ async def get_compute_registry_from_container( ) return registry except AttributeError as e: + # Note: service_registry case is now handled by _validate_service_registry + # This block handles other AttributeErrors like missing resolve_service error_str = str(e) - if "service_registry" in error_str: - hint = ( - "Container missing 'service_registry' attribute. " - "Expected ModelONEXContainer from omnibase_core." - ) - elif "resolve_service" in error_str: + if "resolve_service" in error_str: hint = ( "Container.service_registry missing 'resolve_service' method. " "Check omnibase_core version compatibility (requires v0.5.6 or later)." @@ -716,8 +715,8 @@ async def get_or_create_compute_registry( wire_infrastructure_services() for production code to ensure proper initialization order and error handling. """ - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "get or create RegistryCompute") + # Validate service_registry is available + _validate_service_registry(container, "get_or_create RegistryCompute") try: # Try to resolve existing RegistryCompute @@ -822,14 +821,14 @@ async def wire_registration_handlers( ProjectorRegistration, ) + # Validate service_registry is available before attempting registration + _validate_service_registry(container, "wire_registration_handlers") + # Resolve the actual liveness interval (from param, env var, or default) resolved_liveness_interval = get_liveness_interval_seconds( liveness_interval_seconds ) - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "wire registration handlers") - services_registered: list[str] = [] try: @@ -926,13 +925,10 @@ async def wire_registration_handlers( ) except AttributeError as e: + # Note: service_registry case is now handled by _validate_service_registry + # This block handles other AttributeErrors like missing register_instance error_str = str(e) - if "service_registry" in error_str: - hint = ( - "Container missing 'service_registry' attribute. " - "Expected ModelONEXContainer from omnibase_core." - ) - elif "register_instance" in error_str: + if "register_instance" in error_str: hint = ( "Container.service_registry missing 'register_instance' method. " "Check omnibase_core version compatibility (requires v0.5.6 or later)." @@ -994,8 +990,8 @@ async def get_projection_reader_from_container( """ from omnibase_infra.projectors import ProjectionReaderRegistration - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "resolve ProjectionReaderRegistration") + # Validate service_registry is available + _validate_service_registry(container, "resolve ProjectionReaderRegistration") try: reader: ProjectionReaderRegistration = ( @@ -1038,8 +1034,8 @@ async def get_handler_node_introspected_from_container( HandlerNodeIntrospected, ) - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "resolve HandlerNodeIntrospected") + # Validate service_registry is available + _validate_service_registry(container, "resolve HandlerNodeIntrospected") try: handler: HandlerNodeIntrospected = ( @@ -1080,8 +1076,8 @@ async def get_handler_runtime_tick_from_container( HandlerRuntimeTick, ) - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "resolve HandlerRuntimeTick") + # Validate service_registry is available + _validate_service_registry(container, "resolve HandlerRuntimeTick") try: handler: HandlerRuntimeTick = await container.service_registry.resolve_service( @@ -1122,8 +1118,8 @@ async def get_handler_node_registration_acked_from_container( HandlerNodeRegistrationAcked, ) - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "resolve HandlerNodeRegistrationAcked") + # Validate service_registry is available + _validate_service_registry(container, "resolve HandlerNodeRegistrationAcked") try: handler: HandlerNodeRegistrationAcked = ( @@ -1217,8 +1213,8 @@ async def wire_registration_dispatchers( DispatcherRuntimeTick, ) - # Early check for None service_registry (circular import bug in omnibase_core) - _check_service_registry_available(container, "wire registration dispatchers") + # Validate service_registry is available + _validate_service_registry(container, "wire_registration_dispatchers") dispatchers_registered: list[str] = [] routes_registered: list[str] = [] @@ -1338,8 +1334,6 @@ async def wire_registration_dispatchers( __all__: list[str] = [ - # Exception for circular import issues - "ServiceRegistryUnavailableError", # Container wiring functions "get_compute_registry_from_container", "get_handler_node_introspected_from_container", diff --git a/src/omnibase_infra/runtime/kernel.py b/src/omnibase_infra/runtime/kernel.py index 389f5bd7ff..f8d1d04f42 100644 --- a/src/omnibase_infra/runtime/kernel.py +++ b/src/omnibase_infra/runtime/kernel.py @@ -35,7 +35,6 @@ Environment Variables: ONEX_CONTRACTS_DIR: Path to contracts directory (default: ./contracts) - Falls back to legacy CONTRACTS_DIR if ONEX_CONTRACTS_DIR is not set. ONEX_HTTP_PORT: Port for health check HTTP server (default: 8085) ONEX_LOG_LEVEL: Logging level (default: INFO) ONEX_ENVIRONMENT: Runtime environment name (default: local) @@ -77,6 +76,7 @@ ) from omnibase_infra.event_bus.inmemory_event_bus import InMemoryEventBus from omnibase_infra.event_bus.kafka_event_bus import KafkaEventBus +from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig from omnibase_infra.event_bus.models.model_event_message import ModelEventMessage from omnibase_infra.models.registration.model_node_introspection_event import ( ModelNodeIntrospectionEvent, @@ -115,9 +115,7 @@ DEFAULT_RUNTIME_CONFIG = "runtime/runtime_config.yaml" # Environment variable names for contracts directory -# Prefer ONEX_ prefix, fall back to legacy name for backward compatibility ENV_CONTRACTS_DIR = "ONEX_CONTRACTS_DIR" -ENV_CONTRACTS_DIR_LEGACY = "CONTRACTS_DIR" DEFAULT_INPUT_TOPIC = "requests" DEFAULT_OUTPUT_TOPIC = "responses" DEFAULT_GROUP_ID = "onex-runtime" @@ -128,31 +126,18 @@ def _get_contracts_dir() -> Path: - """Get contracts directory from environment with legacy fallback. + """Get contracts directory from environment. - Checks for the ONEX_CONTRACTS_DIR environment variable first. If not set, - falls back to the legacy CONTRACTS_DIR for backward compatibility. - Logs an info message when legacy variable is used. + Reads the ONEX_CONTRACTS_DIR environment variable. If not set, + returns the default contracts directory. Returns: Path to the contracts directory. """ - # Prefer ONEX_ prefix onex_value = os.environ.get(ENV_CONTRACTS_DIR) if onex_value: return Path(onex_value) - # Fall back to legacy name for backward compatibility - legacy_value = os.environ.get(ENV_CONTRACTS_DIR_LEGACY) - if legacy_value: - logger.info( - "Using legacy %s; consider migrating to %s", - ENV_CONTRACTS_DIR_LEGACY, - ENV_CONTRACTS_DIR, - ) - return Path(legacy_value) - - # Default return Path(DEFAULT_CONTRACTS_DIR) @@ -328,7 +313,6 @@ async def bootstrap() -> int: Bootstrap Sequence: 1. Determine contracts directory from ONEX_CONTRACTS_DIR environment variable - (falls back to legacy CONTRACTS_DIR if ONEX_CONTRACTS_DIR is not set) 2. Load and validate runtime configuration from contracts or environment 3. Create and initialize event bus (InMemoryEventBus or KafkaEventBus based on config) 4. Create ModelONEXContainer and wire infrastructure services (async) @@ -359,7 +343,6 @@ async def bootstrap() -> int: Environment Variables: ONEX_CONTRACTS_DIR: Path to contracts directory (default: ./contracts) - Falls back to legacy CONTRACTS_DIR if ONEX_CONTRACTS_DIR is not set. ONEX_HTTP_PORT: Port for health check server (default: 8085) ONEX_LOG_LEVEL: Logging level (default: INFO) ONEX_ENVIRONMENT: Environment name (default: local) @@ -394,7 +377,7 @@ async def bootstrap() -> int: bootstrap_start_time = time.time() try: - # 1. Determine contracts directory (with legacy fallback) + # 1. Determine contracts directory contracts_dir = _get_contracts_dir() logger.info( "ONEX Kernel starting with contracts_dir=%s (correlation_id=%s)", @@ -432,13 +415,13 @@ async def bootstrap() -> int: if use_kafka: # Use KafkaEventBus for production/integration testing - # KafkaEventBus.default() picks up KAFKA_BOOTSTRAP_SERVERS from environment - event_bus = KafkaEventBus( + kafka_config = ModelKafkaEventBusConfig( bootstrap_servers=kafka_bootstrap_servers or "localhost:9092", environment=environment, group=config.consumer_group, circuit_breaker_threshold=config.event_bus.circuit_breaker_threshold, ) + event_bus = KafkaEventBus(config=kafka_config) event_bus_type = "kafka" # Start KafkaEventBus to connect to Kafka/Redpanda and enable consumers @@ -1198,7 +1181,6 @@ def main() -> None: __all__: list[str] = [ "ENV_CONTRACTS_DIR", - "ENV_CONTRACTS_DIR_LEGACY", "bootstrap", "load_runtime_config", "main", diff --git a/src/omnibase_infra/runtime/message_dispatch_engine.py b/src/omnibase_infra/runtime/message_dispatch_engine.py index 5e32f5f95a..61fa6b3599 100644 --- a/src/omnibase_infra/runtime/message_dispatch_engine.py +++ b/src/omnibase_infra/runtime/message_dispatch_engine.py @@ -92,9 +92,8 @@ The lock is NEVER held during I/O operations (dispatcher execution), ensuring that slow dispatchers do not block metrics updates in other threads. - Legacy dict-based metrics (``_metrics``) use simple atomic increments and - may be approximate under very high concurrency. For production monitoring, - prefer ``get_structured_metrics()`` which returns a consistent snapshot. + For production monitoring, use ``get_structured_metrics()`` which returns + a consistent snapshot. Related: - OMN-934: Message dispatch engine implementation @@ -171,7 +170,7 @@ class ModelLogContextKwargs(TypedDict, total=False): All fields are optional (total=False) since callers pass only the relevant subset. ModelDispatchLogContext validators handle None-to-sentinel - conversion for backwards compatibility. + conversion. .. versionadded:: 0.6.3 Created as part of Union Reduction Phase 2 (OMN-1002) to eliminate @@ -189,13 +188,7 @@ class ModelLogContextKwargs(TypedDict, total=False): error_code: EnumCoreErrorCode | None -# Type alias for dispatcher output topics (LEGACY - for backwards compatibility) -# -# .. deprecated:: 0.6.0 -# External dispatchers should continue using this type for their return values, -# but new internal code should prefer ModelDispatchOutcome which eliminates -# the 3-way union pattern. The dispatch engine normalizes legacy outputs to -# ModelDispatchOutcome internally. +# Type alias for dispatcher output topics # # Dispatchers can return: # - str: A single output topic @@ -348,15 +341,14 @@ class MessageDispatchEngine: pure and fast (~microseconds), so lock contention is minimal. - Structured metrics: Use ``_metrics_lock`` for atomic updates - - Legacy dict metrics: Simple increments, may be approximate under high load - - Prefer ``get_structured_metrics()`` for production monitoring + - Use ``get_structured_metrics()`` for production monitoring **METRICS CAVEAT**: While metrics updates are protected by a lock, - get_metrics() and get_structured_metrics() provide point-in-time - snapshots. Under high concurrent load, metrics may be approximate - between snapshot reads. For production monitoring, consider exporting - metrics to a dedicated metrics backend (Prometheus, StatsD, etc.) for - accurate aggregation across time windows. + get_structured_metrics() provides point-in-time snapshots. Under high + concurrent load, metrics may be approximate between snapshot reads. + For production monitoring, consider exporting metrics to a dedicated + metrics backend (Prometheus, StatsD, etc.) for accurate aggregation + across time windows. Logging Levels: - **INFO**: Dispatch start/complete with topic, category, dispatcher count @@ -461,32 +453,6 @@ def __init__( # Delegates time injection rule enforcement to a single source of truth. self._context_enforcer: DispatchContextEnforcer = DispatchContextEnforcer() - # Legacy metrics dict (for backwards compatibility) - # - # DEPRECATION NOTICE: - # This dict-based metrics format is deprecated in favor of the structured - # ModelDispatchMetrics accessible via get_structured_metrics(). The legacy - # format is retained only for backwards compatibility with existing consumers. - # - # Why not convert to a Pydantic model? - # - ModelDispatchMetrics already exists and provides full typed metrics - # - This dict duplicates that functionality for legacy API support - # - Converting would add a third format without benefit - # - The proper migration path is: consumers should switch to get_structured_metrics() - # - # Type: int | float covers all metric values (counts are int, latency is float) - self._metrics: dict[str, int | float] = { - "dispatch_count": 0, - "dispatch_success_count": 0, - "dispatch_error_count": 0, - "total_latency_ms": 0.0, - "dispatcher_execution_count": 0, - "dispatcher_error_count": 0, - "routes_matched_count": 0, - "no_dispatcher_count": 0, - "category_mismatch_count": 0, - } - def register_route(self, route: ModelDispatchRoute) -> None: """ Register a routing rule. @@ -611,7 +577,7 @@ def register_dispatcher( with appropriate time injection based on ONEX rules: - REDUCER/COMPUTE: now=None (deterministic execution) - ORCHESTRATOR/EFFECT/RUNTIME_HOST: now=datetime.now(UTC) - When None, dispatcher is called without context (backwards compatible). + When None, dispatcher is called without context. Raises: ModelOnexError: If engine is frozen (INVALID_STATE) @@ -907,10 +873,6 @@ async def dispatch( correlation_id = envelope.correlation_id or uuid4() trace_id = envelope.trace_id - # Update dispatch count (protected by lock for thread safety) - with self._metrics_lock: - self._metrics["dispatch_count"] += 1 - # Step 1: Parse topic to get category topic_category = EnumMessageCategory.from_topic(topic) if topic_category is None: @@ -920,8 +882,6 @@ async def dispatch( # Update metrics (protected by lock for thread safety) with self._metrics_lock: - self._metrics["dispatch_error_count"] += 1 - self._metrics["total_latency_ms"] += duration_ms self._structured_metrics = self._structured_metrics.record_dispatch( duration_ms=duration_ms, success=False, @@ -978,8 +938,7 @@ async def dispatch( # The code below is disabled until infer_category() is available: # envelope_category = envelope.infer_category() # if envelope_category != topic_category: - # self._metrics["category_mismatch_count"] += 1 - # ... (category mismatch handling) + # ... (category mismatch handling with structured metrics) # Step 3: Get message type from payload message_type = type(envelope.payload).__name__ @@ -1013,9 +972,6 @@ async def dispatch( # Update metrics (protected by lock for thread safety) with self._metrics_lock: - self._metrics["no_dispatcher_count"] += 1 - self._metrics["dispatch_error_count"] += 1 - self._metrics["total_latency_ms"] += duration_ms self._structured_metrics = self._structured_metrics.record_dispatch( duration_ms=duration_ms, success=False, @@ -1063,9 +1019,6 @@ async def dispatch( executed_dispatcher_ids: list[str] = [] for dispatcher_entry in matching_dispatchers: - # Update execution count (protected by lock) - with self._metrics_lock: - self._metrics["dispatcher_execution_count"] += 1 dispatcher_start_time = time.perf_counter() # Log dispatcher execution at DEBUG level @@ -1180,7 +1133,6 @@ async def dispatch( # These operations are pure (no I/O) and fast (~microseconds), # so holding the lock during computation is acceptable. with self._metrics_lock: - self._metrics["dispatcher_error_count"] += 1 existing_dispatcher_metrics = ( self._structured_metrics.dispatcher_metrics.get( dispatcher_entry.dispatcher_id @@ -1250,22 +1202,16 @@ async def dispatch( # Update all metrics atomically (protected by lock) with self._metrics_lock: - self._metrics["total_latency_ms"] += duration_ms - self._metrics["routes_matched_count"] += len(matching_dispatchers) - if dispatcher_errors: - self._metrics["dispatch_error_count"] += 1 - else: - self._metrics["dispatch_success_count"] += 1 - # NOTE: dispatcher_id is NOT passed here because per-dispatcher metrics - # (including dispatcher_execution_count) are already updated in the - # dispatcher loop above. Passing dispatcher_id here would cause - # double-counting of dispatcher_execution_count. + # NOTE: dispatcher_id and handler_error are NOT passed here because + # per-dispatcher metrics (including dispatcher_execution_count and + # dispatcher_error_count) are already updated in the dispatcher loop + # above. Passing them here would cause double-counting. self._structured_metrics = self._structured_metrics.record_dispatch( duration_ms=duration_ms, success=status == EnumDispatchStatus.SUCCESS, category=topic_category, dispatcher_id=None, # Already tracked in dispatcher loop - handler_error=len(dispatcher_errors) > 0, + handler_error=False, # Already tracked in dispatcher loop routes_matched=len(matching_dispatchers), topic=topic, error_message=dispatcher_errors[0] if dispatcher_errors else None, @@ -1690,7 +1636,7 @@ def _dispatcher_accepts_context( # signature mismatches where a 2+ parameter dispatcher might not # actually expect a ModelDispatchContext. # - # This is NON-BLOCKING - we still return True for backwards compatibility. + # This is NON-BLOCKING - we still return True. # The warning is informational to help improve code quality. second_param = params[1] second_name = second_param.name.lower() @@ -1717,45 +1663,6 @@ def _dispatcher_accepts_context( ) return False - def get_metrics(self) -> dict[str, int | float]: - """ - Get dispatch metrics for observability (legacy format). - - .. deprecated:: 0.4.0 - Use :meth:`get_structured_metrics` instead. This method returns - a simple dict which may have approximate values under very high - concurrency. The structured metrics API provides consistent snapshots - and richer observability data. - - Returns a snapshot of current metrics including: - - dispatch_count: Total number of dispatch calls - - dispatch_success_count: Successful dispatches - - dispatch_error_count: Failed dispatches - - total_latency_ms: Cumulative latency in milliseconds - - dispatcher_execution_count: Total dispatcher executions - - dispatcher_error_count: Dispatcher execution failures - - routes_matched_count: Total route matches - - no_dispatcher_count: Dispatches with no matching dispatcher - - category_mismatch_count: Category validation failures - - Returns: - Dictionary with metrics (copy of internal state) - - Example: - >>> metrics = engine.get_metrics() - >>> print(f"Success rate: {metrics['dispatch_success_count'] / metrics['dispatch_count']:.1%}") - - Note: - Returns a copy to prevent external modification. - For high-frequency monitoring, consider caching the result. - For structured metrics, use get_structured_metrics() instead. - - .. versionadded:: 0.4.0 - """ - # Return a copy under lock to ensure consistent snapshot - with self._metrics_lock: - return dict(self._metrics) - def get_structured_metrics(self) -> ModelDispatchMetrics: """ Get structured dispatch metrics using Pydantic model. @@ -1794,7 +1701,6 @@ def reset_metrics(self) -> None: Reset all metrics to initial state. Useful for testing or when starting a new monitoring period. - Resets both legacy dict-based metrics and structured metrics. Thread Safety: This method acquires ``_metrics_lock`` to ensure atomic reset @@ -1803,23 +1709,11 @@ def reset_metrics(self) -> None: Example: >>> engine.reset_metrics() - >>> assert engine.get_metrics()["dispatch_count"] == 0 >>> assert engine.get_structured_metrics().total_dispatches == 0 .. versionadded:: 0.4.0 """ with self._metrics_lock: - self._metrics = { - "dispatch_count": 0, - "dispatch_success_count": 0, - "dispatch_error_count": 0, - "total_latency_ms": 0.0, - "dispatcher_execution_count": 0, - "dispatcher_error_count": 0, - "routes_matched_count": 0, - "no_dispatcher_count": 0, - "category_mismatch_count": 0, - } self._structured_metrics = ModelDispatchMetrics() self._logger.debug("Metrics reset to initial state") @@ -1861,12 +1755,6 @@ def dispatcher_count(self) -> int: """Get the number of registered dispatchers.""" return len(self._dispatchers) - # Legacy property for backward compatibility - @property - def handler_count(self) -> int: - """Get the number of registered dispatchers (legacy alias).""" - return len(self._dispatchers) - def __str__(self) -> str: """Human-readable string representation.""" return ( @@ -1896,25 +1784,3 @@ def __repr__(self) -> str: f"dispatchers={dispatcher_repr}, " f"frozen={self._frozen})" ) - - # Legacy method aliases for backward compatibility - def register_handler( - self, - handler_id: str, - handler: DispatcherFunc, - category: EnumMessageCategory, - message_types: set[str] | None = None, - node_kind: EnumNodeKind | None = None, - ) -> None: - """Register a message handler (legacy alias for register_dispatcher).""" - return self.register_dispatcher( - dispatcher_id=handler_id, - dispatcher=handler, - category=category, - message_types=message_types, - node_kind=node_kind, - ) - - def get_handler_metrics(self, handler_id: str) -> ModelDispatcherMetrics | None: - """Get metrics for a specific handler (legacy alias for get_dispatcher_metrics).""" - return self.get_dispatcher_metrics(handler_id) diff --git a/src/omnibase_infra/runtime/registry_compute.py b/src/omnibase_infra/runtime/registry_compute.py index d925fcd81f..7bddedfeda 100644 --- a/src/omnibase_infra/runtime/registry_compute.py +++ b/src/omnibase_infra/runtime/registry_compute.py @@ -150,20 +150,17 @@ def __exit__(self, *args: object) -> None: # Environment variable for configuring the semver LRU cache size # Set ONEX_COMPUTE_REGISTRY_CACHE_SIZE to tune cache size for large deployments -# Falls back to legacy COMPUTE_REGISTRY_CACHE_SIZE for backward compatibility ENV_COMPUTE_REGISTRY_CACHE_SIZE = "ONEX_COMPUTE_REGISTRY_CACHE_SIZE" -ENV_COMPUTE_REGISTRY_CACHE_SIZE_LEGACY = "COMPUTE_REGISTRY_CACHE_SIZE" # Default cache size when environment variable is not set _DEFAULT_SEMVER_CACHE_SIZE = 128 def _get_compute_registry_cache_size() -> int: - """Get compute registry cache size from environment with legacy fallback. + """Get compute registry cache size from environment. - Checks for the ONEX_COMPUTE_REGISTRY_CACHE_SIZE environment variable first. - If not set, falls back to the legacy COMPUTE_REGISTRY_CACHE_SIZE for - backward compatibility. Logs an info message when legacy variable is used. + Reads the ONEX_COMPUTE_REGISTRY_CACHE_SIZE environment variable. + If not set, returns the default cache size. Range validation: Cache size must be between 1 and 10000. - Below 1: Uses default (logged as warning) @@ -179,7 +176,6 @@ def _get_compute_registry_cache_size() -> int: from omnibase_infra.enums import EnumInfraTransportType from omnibase_infra.utils.util_env_parsing import parse_env_int - # Prefer ONEX_ prefix if os.environ.get(ENV_COMPUTE_REGISTRY_CACHE_SIZE) is not None: return parse_env_int( ENV_COMPUTE_REGISTRY_CACHE_SIZE, @@ -190,23 +186,6 @@ def _get_compute_registry_cache_size() -> int: service_name="compute_registry", ) - # Fall back to legacy name for backward compatibility - if os.environ.get(ENV_COMPUTE_REGISTRY_CACHE_SIZE_LEGACY) is not None: - logger.info( - "Using legacy %s; consider migrating to %s", - ENV_COMPUTE_REGISTRY_CACHE_SIZE_LEGACY, - ENV_COMPUTE_REGISTRY_CACHE_SIZE, - ) - return parse_env_int( - ENV_COMPUTE_REGISTRY_CACHE_SIZE_LEGACY, - _DEFAULT_SEMVER_CACHE_SIZE, - min_value=1, - max_value=10000, - transport_type=EnumInfraTransportType.HTTP, - service_name="compute_registry", - ) - - # Default return _DEFAULT_SEMVER_CACHE_SIZE @@ -299,7 +278,6 @@ class RegistryCompute: ONEX_COMPUTE_REGISTRY_CACHE_SIZE: Configure the semver LRU cache size for large deployments. Set to a higher value (e.g., 256, 512) if you have many unique version strings. Default: 128. - Falls back to legacy COMPUTE_REGISTRY_CACHE_SIZE if not set. Attributes: _registry: Internal dictionary mapping ModelComputeKey instances to plugin classes @@ -327,7 +305,6 @@ class RegistryCompute: # ========================================================================== # Semver cache size - configurable via ONEX_COMPUTE_REGISTRY_CACHE_SIZE env var - # Falls back to legacy COMPUTE_REGISTRY_CACHE_SIZE for backward compatibility # Read at class definition time; can be overridden via class attribute before first parse SEMVER_CACHE_SIZE: int = _get_compute_registry_cache_size() @@ -1091,7 +1068,6 @@ def _reset_semver_cache(cls) -> None: __all__: list[str] = [ # Environment variable constants "ENV_COMPUTE_REGISTRY_CACHE_SIZE", - "ENV_COMPUTE_REGISTRY_CACHE_SIZE_LEGACY", # Re-export models for convenience "ModelComputeKey", "ModelComputeRegistration", diff --git a/src/omnibase_infra/utils/util_semver.py b/src/omnibase_infra/utils/util_semver.py index 7496087d5d..0e2e6b8d2f 100644 --- a/src/omnibase_infra/utils/util_semver.py +++ b/src/omnibase_infra/utils/util_semver.py @@ -25,7 +25,6 @@ from __future__ import annotations import re -import warnings # Semantic versioning pattern: MAJOR.MINOR.PATCH[-prerelease][+build] # See: https://semver.org/ @@ -66,8 +65,6 @@ def validate_version_lenient(v: str) -> str: """Validate version format with lenient parsing. Accepts flexible version formats including partial versions. - Used by compute registry models that need backward compatibility - with abbreviated version strings. Accepted formats: - "1" (major only) @@ -202,26 +199,8 @@ def normalize_version_cached(version: str) -> str: ) -def clear_normalize_version_cache() -> None: - """REMOVED: Cache no longer exists since normalize_version_cached was removed. - - This function is kept for API compatibility but does nothing. - - .. deprecated:: - This function is deprecated and will be removed in a future version. - Caching is now handled internally by ModelSemVer. - """ - warnings.warn( - "clear_normalize_version_cache() is deprecated and does nothing. " - "Caching is handled internally by ModelSemVer.", - DeprecationWarning, - stacklevel=2, - ) - - __all__: list[str] = [ "SEMVER_PATTERN", - "clear_normalize_version_cache", "normalize_version", "normalize_version_cached", "validate_semver", diff --git a/src/omnibase_infra/validation/__init__.py b/src/omnibase_infra/validation/__init__.py index 6e81bd0083..98e94de298 100644 --- a/src/omnibase_infra/validation/__init__.py +++ b/src/omnibase_infra/validation/__init__.py @@ -130,7 +130,7 @@ from omnibase_infra.validation.execution_shape_validator import ( EXECUTION_SHAPE_RULES, ExecutionShapeValidator, - HandlerInfo, + ModelDetectedNodeInfo, ModelExecutionShapeValidationResult, get_execution_shape_rules, validate_execution_shapes, @@ -219,7 +219,7 @@ # AST-based execution shape validation (OMN-958) "ExecutionShapeValidator", "ExecutionShapeViolationError", - "HandlerInfo", + "ModelDetectedNodeInfo", "ModelContractLintResult", "ModelContractViolation", "ModelExecutionShapeValidationResult", diff --git a/src/omnibase_infra/validation/chain_propagation_validator.py b/src/omnibase_infra/validation/chain_propagation_validator.py index 57c3cab813..5ef6a06051 100644 --- a/src/omnibase_infra/validation/chain_propagation_validator.py +++ b/src/omnibase_infra/validation/chain_propagation_validator.py @@ -35,17 +35,15 @@ - causation_id: Optional field referencing parent's envelope_id For envelopes without an explicit causation_id field, the validator - checks the metadata for a 'causation_id' key or 'parent_message_id' key. + checks the metadata for a 'causation_id' key. Causation ID Semantics: **Canonical Location**: When producing child messages, set causation_id in - ``metadata.tags["causation_id"]`` as a string UUID. This is the preferred - location for new ONEX implementations. + ``metadata.tags["causation_id"]`` as a string UUID. For HTTP transports, + use ``metadata.headers["x-causation-id"]``. **Why metadata.tags?** The ModelEventEnvelope's metadata.tags dict provides - a flexible, schema-stable location for tracing metadata. Unlike first-class - envelope fields (which require schema changes), tags can be extended without - breaking compatibility. + a flexible, schema-stable location for tracing metadata. **Producer Responsibility**: When creating a child envelope from a parent: @@ -61,19 +59,6 @@ ), ) - **Fallback Locations**: The validator also checks these locations for - backwards compatibility with legacy systems and external integrations: - - - ``envelope.causation_id`` - Direct attribute (if envelope model has it) - - ``metadata.tags["parent_message_id"]`` - Legacy terminology alias - - ``metadata.headers["x-causation-id"]`` - HTTP transport convention - - ``metadata.headers["causation-id"]`` - Alternate HTTP header name - - ``metadata.headers["x-parent-message-id"]`` - Legacy HTTP header - - These fallbacks exist to support interoperability with older ONEX versions, - HTTP-based message transports, and external event producers that may use - different naming conventions. - Thread Safety: The ChainPropagationValidator is stateless and thread-safe. All validation methods are pure functions that produce fresh result objects. @@ -156,26 +141,18 @@ # Causation ID Lookup Keys # ============================================================================== # These constants define the keys checked when resolving causation_id from -# envelope metadata. The canonical location is metadata.tags["causation_id"], -# but other keys are checked for backwards compatibility. +# envelope metadata. -CAUSATION_ID_TAG_KEYS: tuple[str, ...] = ("causation_id", "parent_message_id") -"""Keys checked in metadata.tags for causation_id (in priority order). +CAUSATION_ID_TAG_KEYS: tuple[str, ...] = ("causation_id",) +"""Keys checked in metadata.tags for causation_id. -The first key ("causation_id") is the canonical location for new implementations. -The second key ("parent_message_id") is a legacy alias maintained for backwards -compatibility with older ONEX versions. +The canonical location for causation_id is metadata.tags["causation_id"]. """ -CAUSATION_ID_HEADER_KEYS: tuple[str, ...] = ( - "x-causation-id", - "causation-id", - "x-parent-message-id", -) -"""Keys checked in metadata.headers for causation_id (in priority order). +CAUSATION_ID_HEADER_KEYS: tuple[str, ...] = ("x-causation-id",) +"""Keys checked in metadata.headers for causation_id. -These HTTP-style header keys support interoperability with HTTP-based message -transports and external event producers that may use different naming conventions. +The canonical HTTP header for causation_id is "x-causation-id". """ # ============================================================================== @@ -219,11 +196,13 @@ def get_correlation_id(envelope: ModelEventEnvelope[object]) -> UUID | None: def get_causation_id(envelope: ModelEventEnvelope[object]) -> UUID | None: """Get the causation_id from an envelope. - Canonical Location: + Canonical Locations: The **canonical location** for causation_id is ``metadata.tags["causation_id"]`` - stored as a string UUID. When creating child envelopes, producers SHOULD set + stored as a string UUID. When creating child envelopes, producers MUST set causation_id in this location for consistency across the ONEX ecosystem. + For HTTP transports, use ``metadata.headers["x-causation-id"]``. + Example of canonical usage when producing a child message:: child_envelope = ModelEventEnvelope( @@ -235,29 +214,15 @@ def get_causation_id(envelope: ModelEventEnvelope[object]) -> UUID | None: ), ) - Fallback Locations (Backwards Compatibility): - This function checks multiple locations for interoperability with legacy - systems and external message producers that may use different conventions: - - 1. **Direct attribute** ``envelope.causation_id`` (UUID) - Some envelope - implementations may expose causation_id as a first-class attribute. + Lookup Order: + 1. **Direct attribute** ``envelope.causation_id`` (UUID) - If envelope + exposes causation_id as a first-class attribute. 2. **Metadata tags** ``metadata.tags["causation_id"]`` (string -> UUID) - - **Canonical location**. Preferred for new implementations. - - 3. **Metadata tags** ``metadata.tags["parent_message_id"]`` (string -> UUID) - - Legacy alias. Some systems use "parent_message_id" terminology. - - 4. **Metadata headers** with keys ``x-causation-id``, ``causation-id``, - ``x-parent-message-id`` (string -> UUID) - HTTP-style headers for - interoperability with HTTP-based message transports. + Canonical location. - Best Practices: - - **Producers**: Always set ``metadata.tags["causation_id"]`` when creating - child envelopes. This ensures consistent behavior across validators. - - **Consumers**: Use this function for extraction - it handles all formats. - - **Migrations**: When migrating from legacy formats, populate both the - canonical location and the legacy location during the transition period. + 3. **Metadata headers** ``metadata.headers["x-causation-id"]`` (string -> UUID) - + Canonical HTTP header. Args: envelope: The event envelope. @@ -275,13 +240,11 @@ def get_causation_id(envelope: ModelEventEnvelope[object]) -> UUID | None: if isinstance(causation_id, UUID): return causation_id - # Check metadata for causation_id or parent_message_id - # Note: metadata.headers or metadata.tags might contain this info + # Check metadata for causation_id if hasattr(envelope, "metadata") and envelope.metadata is not None: metadata = envelope.metadata - # Check if metadata has a tags dict with causation info - # Uses CAUSATION_ID_TAG_KEYS constant for lookup priority + # Check metadata.tags for causation_id (canonical location) if hasattr(metadata, "tags") and metadata.tags: tags = metadata.tags for key in CAUSATION_ID_TAG_KEYS: @@ -299,8 +262,7 @@ def get_causation_id(envelope: ModelEventEnvelope[object]) -> UUID | None: value, ) - # Check headers dict as well - # Uses CAUSATION_ID_HEADER_KEYS constant for lookup priority + # Check metadata.headers for x-causation-id (HTTP transport) if hasattr(metadata, "headers") and metadata.headers: headers = metadata.headers for key in CAUSATION_ID_HEADER_KEYS: diff --git a/src/omnibase_infra/validation/execution_shape_validator.py b/src/omnibase_infra/validation/execution_shape_validator.py index c13d4aeb47..8062735327 100644 --- a/src/omnibase_infra/validation/execution_shape_validator.py +++ b/src/omnibase_infra/validation/execution_shape_validator.py @@ -55,7 +55,7 @@ - The validator is stateless after initialization (only stores immutable rules) - AST parsing happens per-file with no shared mutable state - - Each validation creates fresh HandlerInfo and violation objects + - Each validation creates fresh ModelDetectedNodeInfo and violation objects For custom validation rules, create a new `ExecutionShapeValidator` instance. For repeated validation of the same files, the singleton pattern provides @@ -277,10 +277,6 @@ class ModelDetectedNodeInfo(BaseModel): file_path: str -# Backwards compatibility alias -HandlerInfo = ModelDetectedNodeInfo - - class ExecutionShapeValidator: """AST-based validator for ONEX handler execution shapes. @@ -429,7 +425,9 @@ def validate_directory( return violations - def _find_handlers(self, tree: ast.AST, file_path: str) -> list[HandlerInfo]: + def _find_handlers( + self, tree: ast.AST, file_path: str + ) -> list[ModelDetectedNodeInfo]: """Find all handler classes and functions in an AST. Detection methods: @@ -444,14 +442,14 @@ def _find_handlers(self, tree: ast.AST, file_path: str) -> list[HandlerInfo]: Returns: List of detected handlers with their archetype information. """ - handlers: list[HandlerInfo] = [] + handlers: list[ModelDetectedNodeInfo] = [] for node in ast.walk(tree): if isinstance(node, ast.ClassDef): archetype = self._detect_node_archetype_from_class(node) if archetype is not None: handlers.append( - HandlerInfo( + ModelDetectedNodeInfo( name=node.name, node_archetype=archetype, node=node, @@ -463,7 +461,7 @@ def _find_handlers(self, tree: ast.AST, file_path: str) -> list[HandlerInfo]: archetype = self._detect_node_archetype_from_decorator(node) if archetype is not None: handlers.append( - HandlerInfo( + ModelDetectedNodeInfo( name=node.name, node_archetype=archetype, node=node, @@ -515,15 +513,9 @@ def _detect_node_archetype_from_decorator( Looks for patterns like: @node_archetype(EnumNodeArchetype.EFFECT) - @handler_type(EnumNodeArchetype.EFFECT) # Backward compatible alias @effect_handler @reducer_handler - Note: - Both ``@node_archetype`` and ``@handler_type`` decorator names are - supported for backward compatibility during the migration from - handler-type to node-archetype terminology. - Args: node: The AST node with decorators. @@ -632,7 +624,7 @@ def _get_name_from_expr(self, expr: ast.expr) -> str | None: def _validate_handler( self, - handler: HandlerInfo, + handler: ModelDetectedNodeInfo, ) -> list[ModelExecutionShapeViolationResult]: """Validate a single handler for execution shape violations. @@ -679,7 +671,7 @@ def _validate_handler( def _check_return_type_violations( self, method: ast.FunctionDef | ast.AsyncFunctionDef, - handler: HandlerInfo, + handler: ModelDetectedNodeInfo, rule: ModelExecutionShapeRule, ) -> list[ModelExecutionShapeViolationResult]: """Check for forbidden return type violations. @@ -766,7 +758,7 @@ def _is_return_type_allowed( def _analyze_return_value( self, return_node: ast.Return, - handler: HandlerInfo, + handler: ModelDetectedNodeInfo, rule: ModelExecutionShapeRule, ) -> list[ModelExecutionShapeViolationResult]: """Analyze a return statement for forbidden types. @@ -917,7 +909,7 @@ def _detect_message_category( def _create_return_type_violation( self, - handler: HandlerInfo, + handler: ModelDetectedNodeInfo, line_number: int, category: EnumMessageCategory | EnumNodeOutputType, type_name: str, @@ -1017,7 +1009,7 @@ def is_projection(cat: EnumMessageCategory | EnumNodeOutputType) -> bool: def _check_direct_publish_violations( self, method: ast.FunctionDef | ast.AsyncFunctionDef, - handler: HandlerInfo, + handler: ModelDetectedNodeInfo, ) -> list[ModelExecutionShapeViolationResult]: """Check for direct publish method calls. @@ -1066,7 +1058,7 @@ def _check_direct_publish_violations( def _check_system_time_violations( self, method: ast.FunctionDef | ast.AsyncFunctionDef, - handler: HandlerInfo, + handler: ModelDetectedNodeInfo, ) -> list[ModelExecutionShapeViolationResult]: """Check for system time access in reducers. @@ -1234,7 +1226,7 @@ def get_execution_shape_rules() -> dict[EnumNodeArchetype, ModelExecutionShapeRu # - The validator's rules dictionary is immutable after initialization # - No per-validation state is stored in the validator instance # - AST parsing happens per-file (no shared mutable state) -# - The HandlerInfo and violations are created fresh for each file +# - The ModelDetectedNodeInfo and violations are created fresh for each file # # Thread Safety: # - The singleton is created at module import time (before any threads) @@ -1267,7 +1259,7 @@ def get_execution_shape_rules() -> dict[EnumNodeArchetype, ModelExecutionShapeRu __all__ = [ "EXECUTION_SHAPE_RULES", "ExecutionShapeValidator", - "HandlerInfo", + "ModelDetectedNodeInfo", "ModelExecutionShapeValidationResult", "get_execution_shape_rules", "validate_execution_shapes", diff --git a/src/omnibase_infra/validation/validation_exemptions.yaml b/src/omnibase_infra/validation/validation_exemptions.yaml index b0265c6a34..25cf3497c3 100644 --- a/src/omnibase_infra/validation/validation_exemptions.yaml +++ b/src/omnibase_infra/validation/validation_exemptions.yaml @@ -237,10 +237,10 @@ pattern_exemptions: - CLAUDE.md (ONEX Architecture - Handler Classification) ticket: OMN-1092 - file_pattern: 'execution_shape_validator\.py' - class_pattern: "Class name 'HandlerInfo'" - violation_pattern: "contains anti-pattern 'Handler'" + class_pattern: "Class name 'ModelDetectedNodeInfo'" + violation_pattern: "contains anti-pattern 'Model'" reason: > - HandlerInfo is a validation data class for describing handler information during AST analysis - it describes handlers, not implements handler behavior. + ModelDetectedNodeInfo is a validation data class for describing detected node information during AST analysis - follows Model* naming convention. documentation: - docs/validation/validator_reference.md diff --git a/tests/conftest.py b/tests/conftest.py index 1b308c9dae..327ce26bbe 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -402,15 +402,18 @@ async def container_with_registries() -> AsyncGenerator[ModelONEXContainer, None Note: This fixture is async because wire_infrastructure_services() is async. - Known Issue (omnibase_core 0.6.2): - Due to a circular import bug in omnibase_core 0.6.2, container.service_registry - may be None when ModelONEXContainer is instantiated. When this occurs, this - fixture will skip the test with a clear message. Upgrade to omnibase_core >= 0.6.3 - once available to resolve this issue. + Important (OMN-1257): + In omnibase_core 0.6.2+, container.service_registry may return None if: + - enable_service_registry=False was passed to constructor + - The ServiceRegistry module is not installed/available + This fixture explicitly enables service_registry and validates it. Yields: ModelONEXContainer instance with infrastructure services wired. + Raises: + pytest.skip: If service_registry is None (ServiceRegistry module unavailable). + Example: >>> async def test_with_real_container(container_with_registries): ... from omnibase_infra.runtime.policy_registry import PolicyRegistry @@ -427,22 +430,28 @@ async def container_with_registries() -> AsyncGenerator[ModelONEXContainer, None """ from omnibase_core.container import ModelONEXContainer - from omnibase_infra.runtime.container_wiring import wire_infrastructure_services + from omnibase_infra.runtime.container_wiring import ( + ServiceRegistryUnavailableError, + wire_infrastructure_services, + ) - # Create real container - container = ModelONEXContainer() + # Create real container with service_registry explicitly enabled + # In omnibase_core 0.6.2+, this may still return None if module unavailable + container = ModelONEXContainer(enable_service_registry=True) - # Guard: Check for circular import bug in omnibase_core 0.6.2 - # When service_registry is None, container wiring will fail. - # Skip the test gracefully with a clear message rather than failing. + # Check if service_registry is available (OMN-1257) if container.service_registry is None: pytest.skip( - "Skipping: container.service_registry is None due to circular import bug " - "in omnibase_core 0.6.2. Upgrade to omnibase_core >= 0.6.3 to fix." + "ServiceRegistry not available in omnibase_core. " + "Tests requiring container.service_registry will be skipped. " + "Check omnibase_core logs for 'ServiceRegistry not available' warnings." ) - # Wire infrastructure services (async operation) - await wire_infrastructure_services(container) + try: + # Wire infrastructure services (async operation) + await wire_infrastructure_services(container) + except ServiceRegistryUnavailableError as e: + pytest.skip(f"ServiceRegistry unavailable: {e}") # Yield container for proper fixture teardown semantics. # ModelONEXContainer doesn't have explicit cleanup methods currently, @@ -463,11 +472,6 @@ async def container_with_handler_registry( Note: This fixture is async because resolve_service() is async. - Known Issue (omnibase_core 0.6.2): - Due to a circular import bug in omnibase_core 0.6.2, container.service_registry - may be None. The parent fixture (container_with_registries) should skip in this - case, but we add a guard here as well for safety. - Args: container_with_registries: Container fixture (automatically injected). @@ -482,14 +486,6 @@ async def container_with_handler_registry( """ from omnibase_infra.runtime.handler_registry import ProtocolBindingRegistry - # Guard: Double-check service_registry availability (parent fixture should skip, - # but add defensive check in case fixture ordering changes) - if container_with_registries.service_registry is None: - pytest.skip( - "Skipping: container.service_registry is None due to circular import bug " - "in omnibase_core 0.6.2. Upgrade to omnibase_core >= 0.6.3 to fix." - ) - registry: ProtocolBindingRegistry = ( await container_with_registries.service_registry.resolve_service( ProtocolBindingRegistry diff --git a/tests/integration/dlq/conftest.py b/tests/integration/dlq/conftest.py index 80dd8d4d62..c3dfe1c618 100644 --- a/tests/integration/dlq/conftest.py +++ b/tests/integration/dlq/conftest.py @@ -42,8 +42,8 @@ logger = logging.getLogger(__name__) from omnibase_infra.dlq import ( - DLQTrackingService, ModelDlqTrackingConfig, + ServiceDlqTracking, ) # ============================================================================= @@ -130,7 +130,7 @@ def dlq_tracking_config() -> ModelDlqTrackingConfig: @pytest_asyncio.fixture async def dlq_tracking_service( dlq_tracking_config: ModelDlqTrackingConfig, -) -> AsyncGenerator[DLQTrackingService, None]: +) -> AsyncGenerator[ServiceDlqTracking, None]: """Create and initialize DLQ tracking service for tests. This fixture handles the complete lifecycle of the DLQ tracking service: @@ -148,7 +148,7 @@ async def dlq_tracking_service( dlq_tracking_config: Test configuration with unique table name. Yields: - Initialized DLQTrackingService ready for testing. + Initialized ServiceDlqTracking ready for testing. Example: >>> async def test_record_replay(dlq_tracking_service): @@ -156,7 +156,7 @@ async def dlq_tracking_service( ... await dlq_tracking_service.record_replay_attempt(record) ... # Table is automatically cleaned up after test """ - service = DLQTrackingService(dlq_tracking_config) + service = ServiceDlqTracking(dlq_tracking_config) await service.initialize() yield service @@ -182,7 +182,7 @@ async def dlq_tracking_service( await service.shutdown() except Exception as e: logger.warning( - "Cleanup failed for DLQTrackingService shutdown: %s", + "Cleanup failed for ServiceDlqTracking shutdown: %s", e, exc_info=True, ) diff --git a/tests/integration/dlq/test_dlq_tracking_integration.py b/tests/integration/dlq/test_dlq_tracking_integration.py index ea1fbee672..3320d53965 100644 --- a/tests/integration/dlq/test_dlq_tracking_integration.py +++ b/tests/integration/dlq/test_dlq_tracking_integration.py @@ -2,7 +2,7 @@ # Copyright (c) 2025 OmniNode Team """Integration tests for DLQ PostgreSQL tracking service. -These tests validate DLQTrackingService behavior against actual PostgreSQL +These tests validate ServiceDlqTracking behavior against actual PostgreSQL infrastructure running on the remote infrastructure server. They require proper database credentials and will be skipped gracefully if the database is not available. @@ -51,10 +51,10 @@ import pytest from omnibase_infra.dlq import ( - DLQTrackingService, EnumReplayStatus, ModelDlqReplayRecord, ModelDlqTrackingConfig, + ServiceDlqTracking, ) from .conftest import POSTGRES_AVAILABLE @@ -77,13 +77,13 @@ # ============================================================================= -class TestDLQTrackingServiceInitialization: - """Tests for DLQTrackingService initialization and lifecycle.""" +class TestServiceDlqTrackingInitialization: + """Tests for ServiceDlqTracking initialization and lifecycle.""" @pytest.mark.asyncio async def test_initialize_creates_table( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, dlq_tracking_config: ModelDlqTrackingConfig, ) -> None: """Test that initialize creates the tracking table. @@ -109,7 +109,7 @@ async def test_initialize_creates_table( @pytest.mark.asyncio async def test_initialize_creates_indexes( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, dlq_tracking_config: ModelDlqTrackingConfig, ) -> None: """Test that initialize creates required indexes. @@ -144,7 +144,7 @@ async def test_initialize_creates_indexes( @pytest.mark.asyncio async def test_initialize_idempotent( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test that calling initialize multiple times is safe. @@ -168,13 +168,13 @@ async def test_initialize_idempotent( # ============================================================================= -class TestDLQTrackingServiceRecord: +class TestServiceDlqTrackingRecord: """Tests for recording replay attempts.""" @pytest.mark.asyncio async def test_record_replay_attempt_success( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test recording a successful replay attempt. @@ -216,7 +216,7 @@ async def test_record_replay_attempt_success( @pytest.mark.asyncio async def test_record_replay_attempt_failure( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test recording a failed replay attempt. @@ -254,7 +254,7 @@ async def test_record_replay_attempt_failure( @pytest.mark.asyncio async def test_record_replay_attempt_skipped( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test recording a skipped replay attempt. @@ -289,7 +289,7 @@ async def test_record_replay_attempt_skipped( @pytest.mark.asyncio async def test_record_replay_attempt_pending( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test recording a pending replay attempt. @@ -322,7 +322,7 @@ async def test_record_replay_attempt_pending( @pytest.mark.asyncio async def test_record_replay_different_topics( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test recording a replay with different original and target topics. @@ -359,13 +359,13 @@ async def test_record_replay_different_topics( # ============================================================================= -class TestDLQTrackingServiceQuery: +class TestServiceDlqTrackingQuery: """Tests for querying replay history.""" @pytest.mark.asyncio async def test_get_replay_history_multiple_attempts( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test getting history with multiple replay attempts. @@ -413,7 +413,7 @@ async def test_get_replay_history_multiple_attempts( @pytest.mark.asyncio async def test_get_replay_history_empty( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test getting history for non-existent message returns empty list. @@ -427,7 +427,7 @@ async def test_get_replay_history_empty( @pytest.mark.asyncio async def test_get_replay_history_isolation( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test that history queries are properly isolated by message ID. @@ -471,7 +471,7 @@ async def test_get_replay_history_isolation( @pytest.mark.asyncio async def test_get_replay_history_preserves_uuids( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test that UUID fields are correctly preserved through storage. @@ -517,13 +517,13 @@ async def test_get_replay_history_preserves_uuids( # ============================================================================= -class TestDLQTrackingServiceHealth: +class TestServiceDlqTrackingHealth: """Tests for health check functionality.""" @pytest.mark.asyncio async def test_health_check_success( self, - dlq_tracking_service: DLQTrackingService, + dlq_tracking_service: ServiceDlqTracking, ) -> None: """Test health check returns True when service is healthy. @@ -545,7 +545,7 @@ async def test_health_check_not_initialized( A service that has not been initialized should report unhealthy since it cannot perform any operations. """ - service = DLQTrackingService(dlq_tracking_config) + service = ServiceDlqTracking(dlq_tracking_config) # Not initialized result = await service.health_check() @@ -561,7 +561,7 @@ async def test_health_check_after_shutdown( After shutdown, the service should report unhealthy since the connection pool is closed. """ - service = DLQTrackingService(dlq_tracking_config) + service = ServiceDlqTracking(dlq_tracking_config) await service.initialize() # Verify healthy before shutdown @@ -580,7 +580,7 @@ async def test_health_check_after_shutdown( # ============================================================================= -class TestDLQTrackingServiceLifecycle: +class TestServiceDlqTrackingLifecycle: """Tests for service lifecycle management.""" @pytest.mark.asyncio @@ -593,7 +593,7 @@ async def test_shutdown_idempotent( The service should handle repeated shutdown calls gracefully without raising errors. """ - service = DLQTrackingService(dlq_tracking_config) + service = ServiceDlqTracking(dlq_tracking_config) await service.initialize() # First shutdown @@ -614,7 +614,7 @@ async def test_shutdown_closes_pool( dlq_tracking_config: ModelDlqTrackingConfig, ) -> None: """Test that shutdown properly closes the connection pool.""" - service = DLQTrackingService(dlq_tracking_config) + service = ServiceDlqTracking(dlq_tracking_config) await service.initialize() # Verify pool exists diff --git a/tests/unit/event_bus/test_kafka_event_bus.py b/tests/unit/event_bus/test_kafka_event_bus.py index b5de11e982..3edc499200 100644 --- a/tests/unit/event_bus/test_kafka_event_bus.py +++ b/tests/unit/event_bus/test_kafka_event_bus.py @@ -54,11 +54,12 @@ async def kafka_event_bus(self, mock_producer: AsyncMock) -> KafkaEventBus: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, group=TEST_GROUP, ) + bus = KafkaEventBus(config=config) yield bus # Cleanup: Ensure resources are freed even if test fails try: @@ -180,8 +181,8 @@ def test_default_properties(self) -> None: assert event_bus.adapter is event_bus def test_custom_properties(self) -> None: - """Test custom property values.""" - event_bus = KafkaEventBus( + """Test custom property values via config.""" + config = ModelKafkaEventBusConfig( bootstrap_servers="kafka.staging:9092", environment="staging", group="worker-group", @@ -189,6 +190,7 @@ def test_custom_properties(self) -> None: max_retry_attempts=5, retry_backoff_base=2.0, ) + event_bus = KafkaEventBus(config=config) assert event_bus.environment == "staging" assert event_bus.group == "worker-group" assert event_bus.adapter is event_bus @@ -230,12 +232,13 @@ async def kafka_event_bus(self, mock_producer: AsyncMock) -> KafkaEventBus: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, group=TEST_GROUP, max_retry_attempts=0, # Disable retries for faster tests ) + bus = KafkaEventBus(config=config) yield bus # Cleanup: Ensure resources are freed even if test fails try: @@ -327,12 +330,13 @@ async def test_publish_circuit_breaker_open(self, mock_producer: AsyncMock) -> N "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, group=TEST_GROUP, circuit_breaker_threshold=1, # Open after 1 failure ) + event_bus = KafkaEventBus(config=config) await event_bus.start() # Record a failure to open the circuit @@ -368,11 +372,12 @@ async def kafka_event_bus(self, mock_producer: AsyncMock) -> KafkaEventBus: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, group=TEST_GROUP, ) + bus = KafkaEventBus(config=config) yield bus # Cleanup: Ensure resources are freed even if test fails try: @@ -470,11 +475,12 @@ async def kafka_event_bus(self, mock_producer: AsyncMock) -> KafkaEventBus: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, group=TEST_GROUP, ) + bus = KafkaEventBus(config=config) yield bus # Cleanup: Ensure resources are freed even if test fails try: @@ -554,14 +560,18 @@ def mock_producer(self) -> AsyncMock: return producer def test_circuit_breaker_threshold_validation(self) -> None: - """Test that invalid circuit_breaker_threshold raises ProtocolConfigurationError.""" - from omnibase_infra.errors import ProtocolConfigurationError + """Test that invalid circuit_breaker_threshold raises validation error. + + Since KafkaEventBus now only accepts config-driven initialization, + we test the config model validation instead. + """ + from pydantic import ValidationError - with pytest.raises(ProtocolConfigurationError, match="positive integer"): - KafkaEventBus(circuit_breaker_threshold=0) + with pytest.raises(ValidationError, match="greater than or equal to 1"): + ModelKafkaEventBusConfig(circuit_breaker_threshold=0) - with pytest.raises(ProtocolConfigurationError, match="positive integer"): - KafkaEventBus(circuit_breaker_threshold=-1) + with pytest.raises(ValidationError, match="greater than or equal to 1"): + ModelKafkaEventBusConfig(circuit_breaker_threshold=-1) @pytest.mark.asyncio async def test_circuit_breaker_opens_after_failures( @@ -572,10 +582,11 @@ async def test_circuit_breaker_opens_after_failures( "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, circuit_breaker_threshold=3, ) + event_bus = KafkaEventBus(config=config) # Record failures async with event_bus._circuit_breaker_lock: @@ -601,10 +612,11 @@ async def test_circuit_breaker_resets_on_success( "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, circuit_breaker_threshold=5, ) + event_bus = KafkaEventBus(config=config) # Record some failures async with event_bus._circuit_breaker_lock: @@ -635,11 +647,12 @@ async def test_circuit_breaker_half_open_state( "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, circuit_breaker_threshold=1, circuit_breaker_reset_timeout=0.1, # Very short for testing ) + event_bus = KafkaEventBus(config=config) # Step 1: Verify initial state is CLOSED async with event_bus._circuit_breaker_lock: @@ -687,11 +700,12 @@ async def test_circuit_breaker_blocks_when_open( "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, circuit_breaker_threshold=1, circuit_breaker_reset_timeout=60, # Long timeout ) + event_bus = KafkaEventBus(config=config) # Open the circuit async with event_bus._circuit_breaker_lock: @@ -729,7 +743,8 @@ async def test_connection_error_type(self, mock_producer: AsyncMock) -> None: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + config = ModelKafkaEventBusConfig(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + event_bus = KafkaEventBus(config=config) with pytest.raises(InfraConnectionError) as exc_info: await event_bus.start() @@ -739,7 +754,8 @@ async def test_connection_error_type(self, mock_producer: AsyncMock) -> None: @pytest.mark.asyncio async def test_unavailable_error_when_not_started(self) -> None: """Test InfraUnavailableError raised when bus not started.""" - event_bus = KafkaEventBus(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + config = ModelKafkaEventBusConfig(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + event_bus = KafkaEventBus(config=config) with pytest.raises(InfraUnavailableError) as exc_info: await event_bus.publish("test-topic", None, b"test") @@ -755,10 +771,11 @@ async def test_timeout_error_handling(self, mock_producer: AsyncMock) -> None: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, timeout_seconds=5, ) + event_bus = KafkaEventBus(config=config) with pytest.raises(InfraTimeoutError) as exc_info: await event_bus.start() @@ -804,11 +821,12 @@ async def mock_send(*args: object, **kwargs: object) -> asyncio.Future[object]: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, max_retry_attempts=3, retry_backoff_base=0.01, # Fast retries for testing ) + event_bus = KafkaEventBus(config=config) await event_bus.start() # This should succeed after retries @@ -834,11 +852,12 @@ async def mock_send(*args: object, **kwargs: object) -> None: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, max_retry_attempts=2, retry_backoff_base=0.01, # Fast retries for testing ) + event_bus = KafkaEventBus(config=config) await event_bus.start() with pytest.raises(InfraConnectionError) as exc_info: @@ -886,10 +905,11 @@ class TestEnvelope(BaseModel): "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, max_retry_attempts=0, ) + event_bus = KafkaEventBus(config=config) await event_bus.start() envelope = TestEnvelope(message="hello", count=42) @@ -911,10 +931,11 @@ async def test_publish_envelope_with_dict(self, mock_producer: AsyncMock) -> Non "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, max_retry_attempts=0, ) + event_bus = KafkaEventBus(config=config) await event_bus.start() envelope = {"message": "hello", "count": 42} @@ -960,11 +981,12 @@ async def test_broadcast_to_environment(self, mock_producer: AsyncMock) -> None: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, max_retry_attempts=0, ) + event_bus = KafkaEventBus(config=config) await event_bus.start() await event_bus.broadcast_to_environment("test_cmd", {"key": "value"}) @@ -990,11 +1012,12 @@ async def test_broadcast_to_specific_environment( "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, max_retry_attempts=0, ) + event_bus = KafkaEventBus(config=config) await event_bus.start() await event_bus.broadcast_to_environment( @@ -1014,11 +1037,12 @@ async def test_send_to_group(self, mock_producer: AsyncMock) -> None: "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, max_retry_attempts=0, ) + event_bus = KafkaEventBus(config=config) await event_bus.start() await event_bus.send_to_group("test_cmd", {"key": "value"}, "target-group") @@ -1232,7 +1256,8 @@ async def test_consumer_started_for_subscription( return_value=mock_consumer, ), ): - event_bus = KafkaEventBus(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + config = ModelKafkaEventBusConfig(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + event_bus = KafkaEventBus(config=config) await event_bus.start() async def handler(msg: ModelEventMessage) -> None: @@ -1260,7 +1285,8 @@ async def test_close_stops_all_consumers( return_value=mock_consumer, ), ): - event_bus = KafkaEventBus(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + config = ModelKafkaEventBusConfig(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + event_bus = KafkaEventBus(config=config) await event_bus.start() async def handler(msg: ModelEventMessage) -> None: @@ -1292,7 +1318,8 @@ async def test_start_consuming_auto_starts(self, mock_producer: AsyncMock) -> No "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + config = ModelKafkaEventBusConfig(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + event_bus = KafkaEventBus(config=config) # Create a task that starts consuming async def consume_briefly() -> None: @@ -1316,7 +1343,8 @@ async def test_start_consuming_exits_on_shutdown( "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer", return_value=mock_producer, ): - event_bus = KafkaEventBus(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + config = ModelKafkaEventBusConfig(bootstrap_servers=TEST_BOOTSTRAP_SERVERS) + event_bus = KafkaEventBus(config=config) await event_bus.start() consuming_started = asyncio.Event() @@ -1370,31 +1398,18 @@ def test_config_property_returns_model(self) -> None: assert bus.config == config assert isinstance(bus.config, ModelKafkaEventBusConfig) - def test_backwards_compatibility_with_direct_params(self) -> None: - """Test that direct parameters still work for backwards compatibility.""" - bus = KafkaEventBus( + def test_config_only_initialization(self) -> None: + """Test that KafkaEventBus only accepts config-driven initialization.""" + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, group=TEST_GROUP, ) + bus = KafkaEventBus(config=config) assert bus.environment == "test" assert bus.group == "test-group" - - def test_parameter_overrides_config(self) -> None: - """Test that explicit parameters override config values.""" - config = ModelKafkaEventBusConfig( - bootstrap_servers="config:9092", - environment="config-env", - group="config-group", - ) - bus = KafkaEventBus( - config=config, - environment="override-env", # This should override - ) - - assert bus.environment == "override-env" - assert bus.group == "config-group" # This should come from config + assert bus.config.bootstrap_servers == TEST_BOOTSTRAP_SERVERS def test_from_yaml_creates_bus(self, tmp_path: Path) -> None: """Test from_yaml() factory method with a temporary config file.""" @@ -1982,10 +1997,11 @@ class TestKafkaEventBusTopicValidation: @pytest.fixture def event_bus(self) -> KafkaEventBus: """Create KafkaEventBus for validation testing.""" - return KafkaEventBus( + config = ModelKafkaEventBusConfig( bootstrap_servers=TEST_BOOTSTRAP_SERVERS, environment=TEST_ENVIRONMENT, ) + return KafkaEventBus(config=config) @pytest.fixture def correlation_id(self) -> UUID: diff --git a/tests/unit/event_bus/test_kafka_threading_safety.py b/tests/unit/event_bus/test_kafka_threading_safety.py index 70caf94a55..915310249f 100644 --- a/tests/unit/event_bus/test_kafka_threading_safety.py +++ b/tests/unit/event_bus/test_kafka_threading_safety.py @@ -37,6 +37,7 @@ import pytest from omnibase_infra.event_bus.kafka_event_bus import KafkaEventBus +from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig class SimulatedProducerError(Exception): @@ -240,11 +241,11 @@ async def test_circuit_breaker_concurrent_access_thread_safe(self) -> None: This test verifies thread-safe circuit breaker behavior under concurrent failure conditions. All operations properly catch and suppress exceptions. """ - bus = KafkaEventBus( - config=None, + config = ModelKafkaEventBusConfig( bootstrap_servers="localhost:9092", circuit_breaker_threshold=3, ) + bus = KafkaEventBus(config=config) with patch( "omnibase_infra.event_bus.kafka_event_bus.AIOKafkaProducer" diff --git a/tests/unit/mixins/test_mixin_node_introspection.py b/tests/unit/mixins/test_mixin_node_introspection.py index b38ca7af3e..72b80938e2 100644 --- a/tests/unit/mixins/test_mixin_node_introspection.py +++ b/tests/unit/mixins/test_mixin_node_introspection.py @@ -53,11 +53,11 @@ PERF_THRESHOLD_CACHE_HIT_MS, PERF_THRESHOLD_GET_CAPABILITIES_MS, PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS, - IntrospectionPerformanceMetrics, MixinNodeIntrospection, ) from omnibase_infra.models.discovery import ( ModelIntrospectionConfig, + ModelIntrospectionPerformanceMetrics, ModelNodeIntrospectionEvent, ) from omnibase_infra.models.registration import ModelNodeHeartbeatEvent @@ -1858,7 +1858,7 @@ async def test_default_keywords_not_mutated(self) -> None: @pytest.mark.unit @pytest.mark.asyncio -class TestMixinNodeIntrospectionPerformanceMetrics: +class TestMixinNodeModelIntrospectionPerformanceMetrics: """Tests for performance metrics tracking and retrieval.""" async def test_get_performance_metrics_returns_none_before_introspection( @@ -1892,7 +1892,7 @@ async def test_get_performance_metrics_returns_metrics_after_introspection( metrics = node.get_performance_metrics() assert metrics is not None - assert isinstance(metrics, IntrospectionPerformanceMetrics) + assert isinstance(metrics, ModelIntrospectionPerformanceMetrics) async def test_performance_metrics_contains_expected_fields(self) -> None: """Test that performance metrics contain all expected fields.""" @@ -2691,7 +2691,7 @@ async def test_benchmark_50ms_target_verification(self) -> None: async def test_benchmark_performance_metrics_validation(self) -> None: """Verify that get_performance_metrics() returns valid timing data. - Validates that IntrospectionPerformanceMetrics captures accurate + Validates that ModelIntrospectionPerformanceMetrics captures accurate timing information that matches actual measured times. """ node = MockNode() @@ -2717,7 +2717,7 @@ async def test_benchmark_performance_metrics_validation(self) -> None: # Validate metrics structure assert metrics is not None, "get_performance_metrics() should return metrics" - assert isinstance(metrics, IntrospectionPerformanceMetrics) + assert isinstance(metrics, ModelIntrospectionPerformanceMetrics) # Validate timing fields are populated assert metrics.total_introspection_ms > 0, ( diff --git a/tests/unit/runtime/test_container_wiring.py b/tests/unit/runtime/test_container_wiring.py index 1894033c03..4f526ef507 100644 --- a/tests/unit/runtime/test_container_wiring.py +++ b/tests/unit/runtime/test_container_wiring.py @@ -213,14 +213,23 @@ class TestAnalyzeAttributeError: """Test _analyze_attribute_error() helper function.""" def test_service_registry_missing(self) -> None: - """Test detection of missing service_registry attribute.""" + """Test detection of missing service_registry attribute. + + Note: After OMN-1257 refactoring, service_registry missing/None cases + are handled by _validate_service_registry() which raises + ServiceRegistryUnavailableError before _analyze_attribute_error is called. + This test verifies the fallback behavior for service_registry when it + somehow reaches _analyze_attribute_error (returns generic hint). + """ from omnibase_infra.runtime.container_wiring import _analyze_attribute_error error_str = "'MockContainer' object has no attribute 'service_registry'" - _missing_attr, hint = _analyze_attribute_error(error_str) + missing_attr, hint = _analyze_attribute_error(error_str) + # After refactoring, service_registry case returns generic hint + # (actual service_registry validation is in _validate_service_registry) + assert missing_attr == "service_registry" assert "service_registry" in hint - assert "ModelONEXContainer" in hint def test_register_instance_missing(self) -> None: """Test detection of missing register_instance method.""" diff --git a/tests/unit/runtime/test_container_wiring_registration.py b/tests/unit/runtime/test_container_wiring_registration.py index 1093145b51..5e2c5994c6 100644 --- a/tests/unit/runtime/test_container_wiring_registration.py +++ b/tests/unit/runtime/test_container_wiring_registration.py @@ -132,23 +132,42 @@ async def test_raises_runtime_error_on_registration_failure(self) -> None: await wire_registration_handlers(mock_container, mock_pool) @pytest.mark.asyncio - async def test_raises_runtime_error_on_missing_service_registry(self) -> None: - """Test that ServiceRegistryUnavailableError is raised if service_registry is None. + async def test_raises_error_on_missing_service_registry(self) -> None: + """Test that ServiceRegistryUnavailableError is raised if container missing service_registry. - Note: This tests the new behavior in omnibase_core 0.6.2+ where service_registry - may be None due to circular import issues. + OMN-1257: Changed from RuntimeError to ServiceRegistryUnavailableError + for clearer error messages when service_registry is missing or None. """ - from omnibase_infra.runtime.container_wiring import ( + from omnibase_infra.errors import ServiceRegistryUnavailableError + + mock_container = MagicMock(spec=[]) # No service_registry attribute + del mock_container.service_registry + + mock_pool = MagicMock() + + with pytest.raises( ServiceRegistryUnavailableError, - ) + match="Container missing 'service_registry' attribute", + ): + await wire_registration_handlers(mock_container, mock_pool) + + @pytest.mark.asyncio + async def test_raises_error_on_none_service_registry(self) -> None: + """Test that ServiceRegistryUnavailableError is raised if service_registry is None. + + OMN-1257: Tests the second validation branch where service_registry + attribute exists but is set to None (e.g., when enable_service_registry=False). + """ + from omnibase_infra.errors import ServiceRegistryUnavailableError mock_container = MagicMock() - mock_container.service_registry = None # Simulate circular import issue + mock_container.service_registry = None # Exists but is None mock_pool = MagicMock() with pytest.raises( - ServiceRegistryUnavailableError, match="service_registry is None" + ServiceRegistryUnavailableError, + match="Container service_registry is None", ): await wire_registration_handlers(mock_container, mock_pool) @@ -305,3 +324,217 @@ async def test_raises_runtime_error_if_not_registered(self) -> None: RuntimeError, match="HandlerNodeRegistrationAcked not registered" ): await get_handler_node_registration_acked_from_container(mock_container) + + +class TestWireInfrastructureServicesValidation: + """Tests for wire_infrastructure_services service_registry validation.""" + + @pytest.mark.asyncio + async def test_raises_error_on_missing_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry attribute is missing.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import wire_infrastructure_services + + mock_container = MagicMock(spec=[]) # No service_registry attribute + del mock_container.service_registry + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container missing 'service_registry' attribute", + ): + await wire_infrastructure_services(mock_container) + + @pytest.mark.asyncio + async def test_raises_error_on_none_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry is None.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import wire_infrastructure_services + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container service_registry is None", + ): + await wire_infrastructure_services(mock_container) + + @pytest.mark.asyncio + async def test_error_contains_operation_name(self) -> None: + """Test that error message includes operation context.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import wire_infrastructure_services + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises(ServiceRegistryUnavailableError) as exc_info: + await wire_infrastructure_services(mock_container) + + # Verify operation name is in the error message + assert "wire_infrastructure_services" in str(exc_info.value) + + +class TestGetPolicyRegistryFromContainerValidation: + """Tests for get_policy_registry_from_container service_registry validation.""" + + @pytest.mark.asyncio + async def test_raises_error_on_missing_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry attribute is missing.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_policy_registry_from_container, + ) + + mock_container = MagicMock(spec=[]) + del mock_container.service_registry + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container missing 'service_registry' attribute", + ): + await get_policy_registry_from_container(mock_container) + + @pytest.mark.asyncio + async def test_raises_error_on_none_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry is None.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_policy_registry_from_container, + ) + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container service_registry is None", + ): + await get_policy_registry_from_container(mock_container) + + @pytest.mark.asyncio + async def test_error_contains_operation_name(self) -> None: + """Test that error message includes operation context.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_policy_registry_from_container, + ) + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises(ServiceRegistryUnavailableError) as exc_info: + await get_policy_registry_from_container(mock_container) + + # Verify operation name is in the error message + assert "resolve PolicyRegistry" in str(exc_info.value) + + +class TestGetComputeRegistryFromContainerValidation: + """Tests for get_compute_registry_from_container service_registry validation.""" + + @pytest.mark.asyncio + async def test_raises_error_on_missing_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry attribute is missing.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_compute_registry_from_container, + ) + + mock_container = MagicMock(spec=[]) + del mock_container.service_registry + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container missing 'service_registry' attribute", + ): + await get_compute_registry_from_container(mock_container) + + @pytest.mark.asyncio + async def test_raises_error_on_none_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry is None.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_compute_registry_from_container, + ) + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container service_registry is None", + ): + await get_compute_registry_from_container(mock_container) + + @pytest.mark.asyncio + async def test_error_contains_operation_name(self) -> None: + """Test that error message includes operation context.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_compute_registry_from_container, + ) + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises(ServiceRegistryUnavailableError) as exc_info: + await get_compute_registry_from_container(mock_container) + + # Verify operation name is in the error message + assert "resolve RegistryCompute" in str(exc_info.value) + + +class TestGetHandlerRegistryFromContainerValidation: + """Tests for get_handler_registry_from_container service_registry validation.""" + + @pytest.mark.asyncio + async def test_raises_error_on_missing_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry attribute is missing.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_handler_registry_from_container, + ) + + mock_container = MagicMock(spec=[]) + del mock_container.service_registry + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container missing 'service_registry' attribute", + ): + await get_handler_registry_from_container(mock_container) + + @pytest.mark.asyncio + async def test_raises_error_on_none_service_registry(self) -> None: + """Test ServiceRegistryUnavailableError when service_registry is None.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_handler_registry_from_container, + ) + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises( + ServiceRegistryUnavailableError, + match="Container service_registry is None", + ): + await get_handler_registry_from_container(mock_container) + + @pytest.mark.asyncio + async def test_error_contains_operation_name(self) -> None: + """Test that error message includes operation context.""" + from omnibase_infra.errors import ServiceRegistryUnavailableError + from omnibase_infra.runtime.container_wiring import ( + get_handler_registry_from_container, + ) + + mock_container = MagicMock() + mock_container.service_registry = None + + with pytest.raises(ServiceRegistryUnavailableError) as exc_info: + await get_handler_registry_from_container(mock_container) + + # Verify operation name is in the error message + assert "resolve ProtocolBindingRegistry" in str(exc_info.value) diff --git a/tests/unit/runtime/test_kernel.py b/tests/unit/runtime/test_kernel.py index 602b956774..1b622edcbc 100644 --- a/tests/unit/runtime/test_kernel.py +++ b/tests/unit/runtime/test_kernel.py @@ -423,10 +423,11 @@ async def test_bootstrap_creates_kafka_event_bus_when_configured( mock_kafka_bus.assert_called_once() mock_inmemory_bus.assert_not_called() - # Verify correct parameters were passed + # Verify config was passed with correct parameters call_kwargs = mock_kafka_bus.call_args[1] - assert call_kwargs["bootstrap_servers"] == "kafka:9092" - assert call_kwargs["environment"] == "test-env" + config = call_kwargs["config"] + assert config.bootstrap_servers == "kafka:9092" + assert config.environment == "test-env" async def test_bootstrap_uses_contracts_dir_from_env( self, @@ -436,11 +437,11 @@ async def test_bootstrap_uses_contracts_dir_from_env( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: - """Test that bootstrap uses CONTRACTS_DIR from environment. + """Test that bootstrap uses ONEX_CONTRACTS_DIR from environment. Uses pytest's tmp_path fixture for automatic temporary directory cleanup. """ - monkeypatch.setenv("CONTRACTS_DIR", str(tmp_path)) + monkeypatch.setenv("ONEX_CONTRACTS_DIR", str(tmp_path)) with patch("omnibase_infra.runtime.kernel.asyncio.Event") as mock_event: event_instance = MagicMock() event_instance.wait = AsyncMock(return_value=None) @@ -663,7 +664,7 @@ async def test_full_bootstrap_with_real_event_bus( This test uses real components except for the shutdown wait and health server. Health server is mocked to avoid port conflicts in parallel tests. """ - monkeypatch.setenv("CONTRACTS_DIR", str(tmp_path)) + monkeypatch.setenv("ONEX_CONTRACTS_DIR", str(tmp_path)) with patch("omnibase_infra.runtime.kernel.HealthServer") as mock_health: mock_health_instance = MagicMock() diff --git a/tests/unit/runtime/test_message_dispatch_engine.py b/tests/unit/runtime/test_message_dispatch_engine.py index 183c53515a..71a747ae50 100644 --- a/tests/unit/runtime/test_message_dispatch_engine.py +++ b/tests/unit/runtime/test_message_dispatch_engine.py @@ -254,7 +254,7 @@ def sync_handler(envelope: ModelEventEnvelope[object]) -> str: category=EnumMessageCategory.EVENT, ) - assert dispatch_engine.handler_count == 1 + assert dispatch_engine.dispatcher_count == 1 def test_register_handler_valid_async( self, dispatch_engine: MessageDispatchEngine @@ -270,7 +270,7 @@ async def async_handler(envelope: ModelEventEnvelope[object]) -> str: category=EnumMessageCategory.EVENT, ) - assert dispatch_engine.handler_count == 1 + assert dispatch_engine.dispatcher_count == 1 def test_register_handler_with_message_types( self, dispatch_engine: MessageDispatchEngine @@ -287,7 +287,7 @@ def handler(envelope: ModelEventEnvelope[object]) -> str: message_types={"UserCreatedEvent", "UserUpdatedEvent"}, ) - assert dispatch_engine.handler_count == 1 + assert dispatch_engine.dispatcher_count == 1 def test_register_handler_multiple_categories( self, dispatch_engine: MessageDispatchEngine @@ -319,7 +319,7 @@ def intent_handler(envelope: ModelEventEnvelope[object]) -> str: category=EnumMessageCategory.INTENT, ) - assert dispatch_engine.handler_count == 3 + assert dispatch_engine.dispatcher_count == 3 def test_register_handler_duplicate_raises_error( self, dispatch_engine: MessageDispatchEngine @@ -525,7 +525,7 @@ def test_freeze_empty_engine(self, dispatch_engine: MessageDispatchEngine) -> No assert dispatch_engine.is_frozen assert dispatch_engine.route_count == 0 - assert dispatch_engine.handler_count == 0 + assert dispatch_engine.dispatcher_count == 0 # ============================================================================ @@ -1265,17 +1265,17 @@ class TestMetrics: def test_initial_metrics(self, dispatch_engine: MessageDispatchEngine) -> None: """Test initial metrics values.""" - metrics = dispatch_engine.get_metrics() - - assert metrics["dispatch_count"] == 0 - assert metrics["dispatch_success_count"] == 0 - assert metrics["dispatch_error_count"] == 0 - assert metrics["total_latency_ms"] == 0.0 - assert metrics["dispatcher_execution_count"] == 0 - assert metrics["dispatcher_error_count"] == 0 - assert metrics["routes_matched_count"] == 0 - assert metrics["no_dispatcher_count"] == 0 - assert metrics["category_mismatch_count"] == 0 + metrics = dispatch_engine.get_structured_metrics() + + assert metrics.total_dispatches == 0 + assert metrics.successful_dispatches == 0 + assert metrics.failed_dispatches == 0 + assert metrics.total_latency_ms == 0.0 + assert metrics.dispatcher_execution_count == 0 + assert metrics.dispatcher_error_count == 0 + assert metrics.routes_matched_count == 0 + assert metrics.no_dispatcher_count == 0 + assert metrics.category_mismatch_count == 0 @pytest.mark.asyncio async def test_metrics_updated_on_success( @@ -1305,13 +1305,13 @@ async def handler(envelope: ModelEventEnvelope[object]) -> str: await dispatch_engine.dispatch("dev.user.events.v1", event_envelope) - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == 1 - assert metrics["dispatch_success_count"] == 1 - assert metrics["dispatch_error_count"] == 0 - assert metrics["dispatcher_execution_count"] == 1 - assert metrics["total_latency_ms"] > 0 - assert metrics["routes_matched_count"] == 1 + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == 1 + assert metrics.successful_dispatches == 1 + assert metrics.failed_dispatches == 0 + assert metrics.dispatcher_execution_count == 1 + assert metrics.total_latency_ms > 0 + assert metrics.routes_matched_count == 1 @pytest.mark.asyncio async def test_metrics_updated_on_handler_error( @@ -1341,11 +1341,11 @@ async def failing_handler(envelope: ModelEventEnvelope[object]) -> None: await dispatch_engine.dispatch("dev.user.events.v1", event_envelope) - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == 1 - assert metrics["dispatch_error_count"] == 1 - assert metrics["dispatcher_execution_count"] == 1 - assert metrics["dispatcher_error_count"] == 1 + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == 1 + assert metrics.failed_dispatches == 1 + assert metrics.dispatcher_execution_count == 1 + assert metrics.dispatcher_error_count == 1 @pytest.mark.asyncio async def test_metrics_updated_on_no_dispatcher( @@ -1358,10 +1358,10 @@ async def test_metrics_updated_on_no_dispatcher( await dispatch_engine.dispatch("dev.user.events.v1", event_envelope) - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == 1 - assert metrics["dispatch_error_count"] == 1 - assert metrics["no_dispatcher_count"] == 1 + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == 1 + assert metrics.failed_dispatches == 1 + assert metrics.no_dispatcher_count == 1 @pytest.mark.skip( reason="TODO(OMN-934): Re-enable when ModelEventEnvelope.infer_category() is implemented in omnibase_core" @@ -1378,10 +1378,10 @@ async def test_metrics_updated_on_category_mismatch( # Sending COMMAND envelope to events topic await dispatch_engine.dispatch("dev.user.events.v1", command_envelope) - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == 1 - assert metrics["dispatch_error_count"] == 1 - assert metrics["category_mismatch_count"] == 1 + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == 1 + assert metrics.failed_dispatches == 1 + assert metrics.category_mismatch_count == 1 @pytest.mark.asyncio async def test_metrics_accumulate_across_dispatches( @@ -1413,10 +1413,10 @@ async def handler(envelope: ModelEventEnvelope[object]) -> str: for _ in range(5): await dispatch_engine.dispatch("dev.user.events.v1", event_envelope) - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == 5 - assert metrics["dispatch_success_count"] == 5 - assert metrics["dispatcher_execution_count"] == 5 + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == 5 + assert metrics.successful_dispatches == 5 + assert metrics.dispatcher_execution_count == 5 # ============================================================================ @@ -1715,27 +1715,27 @@ def test_route_count(self, dispatch_engine: MessageDispatchEngine) -> None: ) assert dispatch_engine.route_count == 2 - def test_handler_count(self, dispatch_engine: MessageDispatchEngine) -> None: - """Test handler_count property.""" + def test_dispatcher_count(self, dispatch_engine: MessageDispatchEngine) -> None: + """Test dispatcher_count property.""" def handler(envelope: ModelEventEnvelope[object]) -> None: pass - assert dispatch_engine.handler_count == 0 + assert dispatch_engine.dispatcher_count == 0 dispatch_engine.register_dispatcher( dispatcher_id="handler-1", dispatcher=handler, category=EnumMessageCategory.EVENT, ) - assert dispatch_engine.handler_count == 1 + assert dispatch_engine.dispatcher_count == 1 dispatch_engine.register_dispatcher( dispatcher_id="handler-2", dispatcher=handler, category=EnumMessageCategory.COMMAND, ) - assert dispatch_engine.handler_count == 2 + assert dispatch_engine.dispatcher_count == 2 # ============================================================================ @@ -1997,11 +1997,11 @@ async def handler(envelope: ModelEventEnvelope[object]) -> str: assert len(results) == dispatch_count # Verify metrics match dispatch count - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == dispatch_count - assert metrics["dispatch_success_count"] == dispatch_count - assert metrics["dispatch_error_count"] == 0 - assert metrics["dispatcher_execution_count"] == dispatch_count + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == dispatch_count + assert metrics.successful_dispatches == dispatch_count + assert metrics.failed_dispatches == 0 + assert metrics.dispatcher_execution_count == dispatch_count @pytest.mark.asyncio async def test_concurrent_dispatch_with_multiple_handlers(self) -> None: @@ -2093,11 +2093,11 @@ async def handler2(envelope: ModelEventEnvelope[object]) -> str: assert len(handler2_results) == dispatch_count # Verify metrics - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == dispatch_count - assert metrics["dispatch_success_count"] == dispatch_count + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == dispatch_count + assert metrics.successful_dispatches == dispatch_count # Each dispatch invokes 2 handlers (fan-out) - assert metrics["dispatcher_execution_count"] == dispatch_count * 2 + assert metrics.dispatcher_execution_count == dispatch_count * 2 @pytest.mark.asyncio async def test_concurrent_dispatch_with_failures(self) -> None: @@ -2196,14 +2196,14 @@ async def failing_handler(envelope: ModelEventEnvelope[object]) -> str: assert len(failure_results) == dispatch_count # Verify metrics track both successes and failures - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == dispatch_count + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == dispatch_count # All dispatches are marked as errors (due to partial failure) - assert metrics["dispatch_error_count"] == dispatch_count + assert metrics.failed_dispatches == dispatch_count # Each dispatch executes 2 handlers (1 success + 1 failure) - assert metrics["dispatcher_execution_count"] == dispatch_count * 2 + assert metrics.dispatcher_execution_count == dispatch_count * 2 # One handler fails per dispatch - assert metrics["dispatcher_error_count"] == dispatch_count + assert metrics.dispatcher_error_count == dispatch_count @pytest.mark.asyncio async def test_concurrent_dispatch_metrics_accuracy(self) -> None: @@ -2258,20 +2258,20 @@ async def handler(envelope: ModelEventEnvelope[object]) -> str: concurrent.futures.wait(futures) # Verify metrics accuracy - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == dispatch_count, ( - f"Expected {dispatch_count} dispatches, got {metrics['dispatch_count']}" + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == dispatch_count, ( + f"Expected {dispatch_count} dispatches, got {metrics.total_dispatches}" ) - assert metrics["dispatch_success_count"] == dispatch_count, ( - f"Expected {dispatch_count} successes, got {metrics['dispatch_success_count']}" + assert metrics.successful_dispatches == dispatch_count, ( + f"Expected {dispatch_count} successes, got {metrics.successful_dispatches}" ) - assert metrics["dispatcher_execution_count"] == dispatch_count, ( + assert metrics.dispatcher_execution_count == dispatch_count, ( f"Expected {dispatch_count} dispatcher executions, " - f"got {metrics['dispatcher_execution_count']}" + f"got {metrics.dispatcher_execution_count}" ) - assert metrics["routes_matched_count"] == dispatch_count, ( + assert metrics.routes_matched_count == dispatch_count, ( f"Expected {dispatch_count} route matches, " - f"got {metrics['routes_matched_count']}" + f"got {metrics.routes_matched_count}" ) @@ -2367,9 +2367,9 @@ async def variable_delay_handler(envelope: ModelEventEnvelope[object]) -> str: assert len(execution_order) == dispatch_count # Verify metrics - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == dispatch_count - assert metrics["dispatch_success_count"] == dispatch_count + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == dispatch_count + assert metrics.successful_dispatches == dispatch_count @pytest.mark.asyncio async def test_concurrent_dispatch_metrics_lock_stress(self) -> None: @@ -2428,25 +2428,13 @@ async def fast_handler(envelope: ModelEventEnvelope[object]) -> str: concurrent.futures.wait(futures) # Verify metrics accuracy after high-concurrency stress - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == dispatch_count, ( - f"Expected {dispatch_count}, got {metrics['dispatch_count']}" - ) - assert metrics["dispatch_success_count"] == dispatch_count - assert metrics["dispatcher_execution_count"] == dispatch_count - - # Verify structured metrics consistency - # Note: Structured metrics track dispatcher executions in two places: - # 1. During individual dispatcher execution (in the loop) - # 2. During final record_dispatch call - # This results in dispatcher_execution_count being higher than legacy metrics - # The legacy metrics are the source of truth for simple counts - structured_metrics = dispatch_engine.get_structured_metrics() - assert structured_metrics.total_dispatches == dispatch_count - assert structured_metrics.successful_dispatches == dispatch_count + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == dispatch_count, ( + f"Expected {dispatch_count}, got {metrics.total_dispatches}" + ) + assert metrics.successful_dispatches == dispatch_count # dispatcher_execution_count is tracked per dispatcher call in the loop - # plus additional increments during per-dispatcher metrics updates - assert structured_metrics.dispatcher_execution_count >= dispatch_count + assert metrics.dispatcher_execution_count == dispatch_count @pytest.mark.asyncio async def test_concurrent_dispatch_with_message_type_filtering(self) -> None: @@ -2629,9 +2617,9 @@ async def stable_handler(envelope: ModelEventEnvelope[object]) -> str: assert results_count == dispatch_count # Verify metrics after extended run - metrics = dispatch_engine.get_metrics() - assert metrics["dispatch_count"] == dispatch_count - assert metrics["dispatch_success_count"] == dispatch_count + metrics = dispatch_engine.get_structured_metrics() + assert metrics.total_dispatches == dispatch_count + assert metrics.successful_dispatches == dispatch_count @pytest.mark.asyncio async def test_concurrent_dispatch_with_sync_handlers(self) -> None: @@ -2717,8 +2705,8 @@ async def async_handler(envelope: ModelEventEnvelope[object]) -> str: assert len(async_results) == dispatch_count # Verify metrics - metrics = dispatch_engine.get_metrics() - assert metrics["dispatcher_execution_count"] == dispatch_count * 2 + metrics = dispatch_engine.get_structured_metrics() + assert metrics.dispatcher_execution_count == dispatch_count * 2 @pytest.mark.asyncio async def test_concurrent_dispatch_correlation_id_preservation(self) -> None: diff --git a/tests/unit/runtime/test_policy_registry.py b/tests/unit/runtime/test_policy_registry.py index e0d3111a8b..4ed86aee18 100644 --- a/tests/unit/runtime/test_policy_registry.py +++ b/tests/unit/runtime/test_policy_registry.py @@ -2356,6 +2356,7 @@ class TestPolicyRegistryVersionNormalizationIntegration: # Partial Version Normalization Tests # ========================================================================= + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_partial_version_major_only_registers_and_retrieves( self, policy_registry: PolicyRegistry ) -> None: @@ -2365,6 +2366,9 @@ def test_partial_version_major_only_registers_and_retrieves( - Registration with "1" succeeds - Lookup with "1" finds the policy - Lookup with "1.0.0" also finds the policy (normalized match) + + Note: This test intentionally uses non-normalized versions ('1') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="partial-major", @@ -2387,6 +2391,7 @@ def test_partial_version_major_only_registers_and_retrieves( policy_cls_normalized = policy_registry.get("partial-major", version="1.0.0") assert policy_cls_normalized is MockSyncPolicy + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_partial_version_major_minor_registers_and_retrieves( self, policy_registry: PolicyRegistry ) -> None: @@ -2396,6 +2401,9 @@ def test_partial_version_major_minor_registers_and_retrieves( - Registration with "1.2" succeeds - Lookup with "1.2" finds the policy - Lookup with "1.2.0" also finds the policy (normalized match) + + Note: This test intentionally uses non-normalized versions ('1.2') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="partial-major-minor", @@ -2419,12 +2427,16 @@ def test_partial_version_major_minor_registers_and_retrieves( ) assert policy_cls_normalized is MockSyncPolicy + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_partial_version_zero_normalizes_correctly( self, policy_registry: PolicyRegistry ) -> None: """Test edge case: version '0' normalizes to '0.0.0'. This is an important edge case as zero versions are valid in semver. + + Note: This test intentionally uses non-normalized versions ('0') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="zero-version", @@ -2443,10 +2455,15 @@ def test_partial_version_zero_normalizes_correctly( policy_cls_normalized = policy_registry.get("zero-version", version="0.0.0") assert policy_cls_normalized is MockSyncPolicy + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_partial_version_zero_major_minor_normalizes_correctly( self, policy_registry: PolicyRegistry ) -> None: - """Test edge case: version '0.1' normalizes to '0.1.0'.""" + """Test edge case: version '0.1' normalizes to '0.1.0'. + + Note: This test intentionally uses non-normalized versions ('0.1') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. + """ policy_registry.register_policy( policy_id="zero-minor-version", policy_class=MockSyncPolicy, # type: ignore[arg-type] @@ -2464,12 +2481,16 @@ def test_partial_version_zero_major_minor_normalizes_correctly( # Whitespace Trimming Integration Tests # ========================================================================= + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_whitespace_trimming_space_padded_version( self, policy_registry: PolicyRegistry ) -> None: """Test that ' 1.0.0 ' is trimmed and matches '1.0.0'. Verifies full integration: register with whitespace, lookup without. + + Note: This test intentionally uses non-normalized versions (' 1.0.0 ') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="space-padded", @@ -2488,12 +2509,16 @@ def test_whitespace_trimming_space_padded_version( policy_cls_with_spaces = policy_registry.get("space-padded", version=" 1.0.0 ") assert policy_cls_with_spaces is MockSyncPolicy + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_whitespace_trimming_tab_newline_version( self, policy_registry: PolicyRegistry ) -> None: """Test that tab and newline characters are trimmed from versions. Verifies: '\\t1.0.0\\n' normalizes to '1.0.0'. + + Note: This test intentionally uses non-normalized versions with whitespace to + verify normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="tab-newline", @@ -2510,12 +2535,16 @@ def test_whitespace_trimming_tab_newline_version( # Lookup with different whitespace patterns assert policy_registry.is_registered("tab-newline", version=" 1.0.0 ") + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_whitespace_with_partial_version_combined( self, policy_registry: PolicyRegistry ) -> None: """Test combined whitespace trimming and partial version normalization. Verifies: ' 2 ' normalizes to '2.0.0'. + + Note: This test intentionally uses non-normalized versions (' 2 ') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="whitespace-partial", @@ -2537,6 +2566,7 @@ def test_whitespace_with_partial_version_combined( # ModelPolicyKey Lookup Equivalence Tests # ========================================================================= + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_model_policy_key_equivalence_partial_versions( self, policy_registry: PolicyRegistry ) -> None: @@ -2544,6 +2574,9 @@ def test_model_policy_key_equivalence_partial_versions( This tests the core of ModelPolicyKey normalization - that registering with '1' and looking up with '1.0.0' uses the same key internally. + + Note: This test intentionally uses non-normalized versions ('1') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. """ # Register with partial version policy_registry.register_policy( @@ -2563,12 +2596,16 @@ def test_model_policy_key_equivalence_partial_versions( policy_cls_full = policy_registry.get("key-equiv-partial", version="1.0.0") assert policy_cls_partial is policy_cls_full is MockPolicyV1 + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_model_policy_key_equivalence_whitespace_versions( self, policy_registry: PolicyRegistry ) -> None: """Test that ModelPolicyKey treats whitespace-padded versions equivalently. Registering with ' 1.0.0 ' and looking up with '1.0.0' should match. + + Note: This test intentionally uses non-normalized versions (' 1.0.0 ') to + verify normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="key-equiv-whitespace", @@ -2583,12 +2620,16 @@ def test_model_policy_key_equivalence_whitespace_versions( policy_cls = policy_registry.get("key-equiv-whitespace", version=version) assert policy_cls is MockPolicyV1, f"Failed for version: {version!r}" + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_multiple_versions_with_normalization( self, policy_registry: PolicyRegistry ) -> None: """Test that multiple normalized versions don't collide unexpectedly. Registering '1' and '1.0.0' should be treated as the same version. + + Note: This test intentionally uses non-normalized versions ('1') to verify + normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="multi-norm", @@ -2613,12 +2654,16 @@ def test_multiple_versions_with_normalization( policy_cls = policy_registry.get("multi-norm") assert policy_cls is MockPolicyV2 + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_latest_version_selection_with_partial_versions( self, policy_registry: PolicyRegistry ) -> None: """Test that get() returns latest version with mixed partial/full formats. Register '1', '2.0', '3.0.0' - get() should return the policy for '3.0.0'. + + Note: This test intentionally uses non-normalized versions ('1', '2.0') to + verify normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="latest-partial", @@ -2690,12 +2735,16 @@ def test_version_zero_zero_zero_explicit( # v-prefix Normalization Tests # ========================================================================= + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_v_prefix_stripped_during_normalization( self, policy_registry: PolicyRegistry ) -> None: """Test that 'v1.0.0' is normalized to '1.0.0'. Common in git tags but not strictly semver. + + Note: This test intentionally uses non-normalized versions ('v1.0.0') to + verify normalization behavior. The DeprecationWarning is expected and suppressed. """ policy_registry.register_policy( policy_id="v-prefix", @@ -2712,10 +2761,15 @@ def test_v_prefix_stripped_during_normalization( # Should also work with v prefix in lookup (both normalized) assert policy_registry.is_registered("v-prefix", version="v1.0.0") + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_v_prefix_with_partial_version( self, policy_registry: PolicyRegistry ) -> None: - """Test that 'v2' normalizes to '2.0.0'.""" + """Test that 'v2' normalizes to '2.0.0'. + + Note: This test intentionally uses non-normalized versions ('v2') to + verify normalization behavior. The DeprecationWarning is expected and suppressed. + """ policy_registry.register_policy( policy_id="v-partial", policy_class=MockSyncPolicy, # type: ignore[arg-type] diff --git a/tests/unit/utils/test_util_semver.py b/tests/unit/utils/test_util_semver.py index cbb02efc4a..f8ce2cb97f 100644 --- a/tests/unit/utils/test_util_semver.py +++ b/tests/unit/utils/test_util_semver.py @@ -12,7 +12,6 @@ from omnibase_infra.utils.util_semver import ( SEMVER_PATTERN, - clear_normalize_version_cache, normalize_version, normalize_version_cached, validate_semver, @@ -60,10 +59,6 @@ def test_normalize_version_internal_flag_still_raises(self) -> None: class TestNormalizeVersionCachedRaisesTypeError: """Tests that normalize_version_cached() raises TypeError for string input.""" - def setup_method(self) -> None: - """Clear the cache before each test.""" - clear_normalize_version_cache() - def test_normalize_version_cached_raises_type_error(self) -> None: """Test that normalize_version_cached raises TypeError for any string input.""" with pytest.raises(TypeError, match="String version input is not allowed"): @@ -79,15 +74,6 @@ def test_normalize_version_cached_error_mentions_model_semver(self) -> None: assert "parse()" in error_message -class TestClearNormalizeVersionCache: - """Tests for clear_normalize_version_cache function.""" - - def test_clear_cache_is_noop(self) -> None: - """Test that clear_normalize_version_cache is a no-op (cache no longer exists).""" - # Should not raise any errors - clear_normalize_version_cache() - - class TestValidateSemver: """Tests for validate_semver function."""