diff --git a/scripts/validation/validate_naming.py b/scripts/validation/validate_naming.py index c1b5c844f6..6e51825fe1 100755 --- a/scripts/validation/validate_naming.py +++ b/scripts/validation/validate_naming.py @@ -219,19 +219,27 @@ class NamingConventionValidator: "merge/": [ "ContractMergeEngine", # Merge engine for typed contract merging (OMN-1127) ], - # REPLAY INFRASTRUCTURE: Effect recording and replay for deterministic testing - # Location: pipeline/replay/ - Replay/recording infrastructure for effects - # Rationale: RecorderEffect records effects for replay, not a Node class. - # The heuristic flags "effect" as a Node indicator, but RecorderEffect - # is a recorder that captures effects, not a node that processes them. - # Similarly, other replay infrastructure classes use descriptive names - # appropriate to their function (ExecutorReplay, InjectorRng, etc.) + # REPLAY INFRASTRUCTURE: Deterministic replay utilities for testing and debugging + # Location: services/replay/ - Replay infrastructure services (injectors/recorders) + # Rationale: These classes provide deterministic replay capabilities (OMN-1116, OMN-1205). + # RecorderEffect, InjectorTime, InjectorRNG are passive observers and injection + # mechanisms - NOT nodes. The heuristic flags "effect" as a Node indicator, but + # RecorderEffect RECORDS effects, it doesn't PERFORM them as a node would. + # Nodes are addressable graph vertices with contracts and dispatch semantics. + # Recorders and injectors are lifecycle-bound utilities for side-effect capture. + "services/replay/": [ + "RecorderEffect", # Effect recording for deterministic replay (OMN-1116) + "InjectorTime", # Time injection for deterministic replay (OMN-1116) + "InjectorRNG", # RNG injection for deterministic replay (OMN-1116) + ], + # REPLAY INFRASTRUCTURE: Effect replay execution for deterministic testing + # Location: pipeline/replay/ - Replay execution and session management + # Rationale: ExecutorReplay and SessionReplay handle replay execution and session state. + # The heuristic flags "effect" as a Node indicator, but these are replay + # infrastructure classes that EXECUTE replays, not node implementations. "pipeline/replay/": [ - "RecorderEffect", # Effect recorder for deterministic replay (OMN-1116) - "ExecutorReplay", # Replay executor for effect playback - "InjectorRng", # RNG injector for deterministic randomness - "InjectorTime", # Time injector for deterministic timestamps - "SessionReplay", # Replay session manager + "ExecutorReplay", # Replay executor for effect playback (OMN-1116) + "SessionReplay", # Replay session manager (OMN-1116) ], } diff --git a/src/omnibase_core/enums/replay/__init__.py b/src/omnibase_core/enums/replay/__init__.py index b4726b5ea6..fccb2dd6cd 100644 --- a/src/omnibase_core/enums/replay/__init__.py +++ b/src/omnibase_core/enums/replay/__init__.py @@ -3,15 +3,23 @@ # SPDX-License-Identifier: Apache-2.0 """Replay infrastructure enums. -This module provides enums for the deterministic replay infrastructure. +This module provides enums for the deterministic replay infrastructure: + +- **EnumRecorderMode**: Recording/replaying/pass-through modes for effect recorder +- **EnumReplayMode**: Production/recording/replaying execution modes +- **EnumOverrideInjectionPoint**: Injection points for config overrides (OMN-1205) .. versionadded:: 0.4.0 """ +from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, +) from omnibase_core.enums.replay.enum_recorder_mode import EnumRecorderMode from omnibase_core.enums.replay.enum_replay_mode import EnumReplayMode __all__ = [ + "EnumOverrideInjectionPoint", "EnumRecorderMode", "EnumReplayMode", ] diff --git a/src/omnibase_core/enums/replay/enum_override_injection_point.py b/src/omnibase_core/enums/replay/enum_override_injection_point.py new file mode 100644 index 0000000000..aa398d6553 --- /dev/null +++ b/src/omnibase_core/enums/replay/enum_override_injection_point.py @@ -0,0 +1,55 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +"""EnumOverrideInjectionPoint - Injection points for config overrides. + +Defines where configuration overrides can be applied during replay execution. +This is a CLOSED vocabulary - new injection points require explicit addition. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from enum import Enum + + +class EnumOverrideInjectionPoint(str, Enum): + """Injection points for configuration overrides during replay. + + Defines WHERE configuration overrides can be applied during replay + execution. This is a closed vocabulary - new injection points require + explicit addition. + + - **HANDLER_CONFIG**: Patch handler configuration model (pre-execution). + Applied before handler execution begins. + + - **ENVIRONMENT**: Environment variable overlay (NOT os.environ mutation). + Provides isolated environment context without affecting global state. + + - **CONTEXT**: Patch replay context fields (copy only, not original). + Modifies a copy of the context to avoid side effects. + + Thread Safety: + Enum values are immutable and thread-safe. + + Example: + >>> point = EnumOverrideInjectionPoint.HANDLER_CONFIG + >>> point == "handler_config" + True + >>> EnumOverrideInjectionPoint("environment") + + + .. versionadded:: 0.4.0 + """ + + HANDLER_CONFIG = "handler_config" + """Patch handler configuration model (pre-execution).""" + + ENVIRONMENT = "environment" + """Environment variable overlay (NOT os.environ mutation).""" + + CONTEXT = "context" + """Patch replay context fields (copy only, not original).""" + + +__all__ = ["EnumOverrideInjectionPoint"] diff --git a/src/omnibase_core/models/replay/__init__.py b/src/omnibase_core/models/replay/__init__.py index 1ded25e393..44d15d795c 100644 --- a/src/omnibase_core/models/replay/__init__.py +++ b/src/omnibase_core/models/replay/__init__.py @@ -6,8 +6,15 @@ This module provides model definitions for deterministic replay infrastructure: +- **ModelConfigOverride**: Single configuration override with key path and value +- **ModelConfigOverrideSet**: Collection of configuration overrides for replay scenarios +- **ModelConfigOverrideFieldPreview**: Preview of a single field change from override +- **ModelConfigOverridePreview**: Complete preview of all changes from applying overrides +- **ModelConfigOverrideResult**: Result of applying configuration overrides +- **ModelConfigOverrideValidation**: Validation result for configuration overrides - **ModelEffectRecord**: Captured effect intent and result pair for replay - **ModelReplayContext**: Determinism context bundling time, RNG seed, and effect records +- **ModelReplayInput**: Input configuration for replay execution - **ModelExecutionCorpus**: Collection of execution manifests for replay testing - **ModelCorpusStatistics**: Computed statistics for an execution corpus - **ModelCorpusTimeRange**: Time range for corpus executions @@ -48,6 +55,9 @@ .. versionadded:: 0.4.0 Added Replay Infrastructure (OMN-1116) +.. versionadded:: 0.4.0 + Added Configuration Override Models (OMN-1205) + .. versionadded:: 0.4.0 Added Execution Corpus Model (OMN-1202) """ @@ -55,6 +65,20 @@ from omnibase_core.models.manifest.model_execution_manifest import ( ModelExecutionManifest, ) +from omnibase_core.models.replay.model_config_override import ModelConfigOverride +from omnibase_core.models.replay.model_config_override_field_preview import ( + ModelConfigOverrideFieldPreview, +) +from omnibase_core.models.replay.model_config_override_preview import ( + ModelConfigOverridePreview, +) +from omnibase_core.models.replay.model_config_override_result import ( + ModelConfigOverrideResult, +) +from omnibase_core.models.replay.model_config_override_set import ModelConfigOverrideSet +from omnibase_core.models.replay.model_config_override_validation import ( + ModelConfigOverrideValidation, +) from omnibase_core.models.replay.model_corpus_capture_window import ( ModelCorpusCaptureWindow, ) @@ -63,15 +87,25 @@ from omnibase_core.models.replay.model_effect_record import ModelEffectRecord from omnibase_core.models.replay.model_execution_corpus import ModelExecutionCorpus from omnibase_core.models.replay.model_replay_context import ModelReplayContext +from omnibase_core.models.replay.model_replay_input import ModelReplayInput __all__ = [ - # Replay models + # Configuration override models + "ModelConfigOverride", + "ModelConfigOverrideFieldPreview", + "ModelConfigOverridePreview", + "ModelConfigOverrideResult", + "ModelConfigOverrideSet", + "ModelConfigOverrideValidation", + # Corpus models "ModelCorpusCaptureWindow", "ModelCorpusStatistics", "ModelCorpusTimeRange", + # Replay models "ModelEffectRecord", "ModelExecutionCorpus", "ModelReplayContext", + "ModelReplayInput", # Related models re-exported for convenience "ModelExecutionManifest", ] diff --git a/src/omnibase_core/models/replay/model_config_override.py b/src/omnibase_core/models/replay/model_config_override.py new file mode 100644 index 0000000000..d009aae114 --- /dev/null +++ b/src/omnibase_core/models/replay/model_config_override.py @@ -0,0 +1,60 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Single configuration override specification for replay injection. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, +) + +__all__ = ["ModelConfigOverride"] + + +class ModelConfigOverride(BaseModel): + """ + A single configuration override specification. + + Attributes: + path: Dot-separated path to the config field (e.g., "llm.openai.temperature"). + value: The new value to set at this path. + injection_point: Where to apply this override (handler_config, environment, context). + + Thread Safety: + Immutable (frozen=True) after creation - thread-safe for concurrent reads. + + Example: + >>> override = ModelConfigOverride( + ... path="llm.temperature", + ... value=0.7, + ... injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ... ) + """ + + # from_attributes=True: Enables construction from ORM/dataclass instances + # and ensures pytest-xdist compatibility across worker processes where + # class identity may differ due to independent imports. + # See CLAUDE.md "Pydantic from_attributes=True for Value Objects". + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + path: str = Field( + ..., + min_length=1, + description="Dot-separated path to the config field", + ) + value: Any = Field( + ..., + description="The new value to set at this path", + ) + injection_point: EnumOverrideInjectionPoint = Field( + default=EnumOverrideInjectionPoint.HANDLER_CONFIG, + description="Where to apply this override", + ) diff --git a/src/omnibase_core/models/replay/model_config_override_field_preview.py b/src/omnibase_core/models/replay/model_config_override_field_preview.py new file mode 100644 index 0000000000..5ce28cbd2a --- /dev/null +++ b/src/omnibase_core/models/replay/model_config_override_field_preview.py @@ -0,0 +1,104 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Single field preview showing before/after state for one config override. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, +) + +__all__ = ["ModelConfigOverrideFieldPreview"] + + +class ModelConfigOverrideFieldPreview(BaseModel): + """ + Single field preview showing before/after for one override. + + Represents the state change for a single configuration field when an + override is applied. Used for dry-run preview and user confirmation. + + Attributes: + path: Dot-separated path to the field (e.g., "retry.max_attempts"). + injection_point: Where this override will be applied in the pipeline. + old_value: Value before override (or MISSING sentinel for new fields). + new_value: Value after override. + value_type: Type of the new value (for display purposes). + path_exists: Whether the path exists in the original configuration. + + Thread Safety: + Immutable (frozen=True) after creation - thread-safe for concurrent reads. + + Example: + >>> preview = ModelConfigOverrideFieldPreview( + ... path="timeout", + ... injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ... old_value=30, + ... new_value=60, + ... value_type="int", + ... path_exists=True, + ... ) + >>> preview.to_markdown_row() + '| timeout | handler_config | `30` | `60` | exists |' + + .. versionadded:: 0.4.0 + """ + + # from_attributes=True: Enables construction from ORM/dataclass instances + # and ensures pytest-xdist compatibility across worker processes where + # class identity may differ due to independent imports. + # See CLAUDE.md "Pydantic from_attributes=True for Value Objects". + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + path: str = Field(..., description="Dot-separated path to the field") + injection_point: EnumOverrideInjectionPoint = Field( + ..., description="Where this override will be applied" + ) + old_value: Any = Field( + ..., description="Value before override (or MISSING sentinel)" + ) + new_value: Any = Field(..., description="Value after override") + value_type: str = Field(default="unknown", description="Type of the new value") + path_exists: bool = Field( + default=True, description="Whether the path exists in original" + ) + + def to_markdown_row(self) -> str: + """Generate markdown table row for this preview. + + Creates a formatted markdown table row showing the field path, + injection point, before/after values, and existence status. + + Returns: + A markdown table row string with pipe delimiters. + """ + old_str = self._format_value(self.old_value) + new_str = self._format_value(self.new_value) + status = "exists" if self.path_exists else "NEW" + return f"| {self.path} | {self.injection_point.value} | {old_str} | {new_str} | {status} |" + + def _format_value(self, value: Any) -> str: + """Format value for markdown display. + + Wraps values in backticks for code formatting. Handles None + specially as `null` and strings with quotes. + + Args: + value: The value to format. + + Returns: + A markdown-formatted string representation of the value. + """ + if value is None: + return "`null`" + if isinstance(value, str): + return f'`"{value}"`' + return f"`{value}`" diff --git a/src/omnibase_core/models/replay/model_config_override_preview.py b/src/omnibase_core/models/replay/model_config_override_preview.py new file mode 100644 index 0000000000..ade44dac46 --- /dev/null +++ b/src/omnibase_core/models/replay/model_config_override_preview.py @@ -0,0 +1,104 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Complete preview of all config overrides with before/after state. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.models.replay.model_config_override_field_preview import ( + ModelConfigOverrideFieldPreview, +) + +__all__ = ["ModelConfigOverridePreview"] + + +class ModelConfigOverridePreview(BaseModel): + """ + Complete preview of all overrides with before/after state. + + Provides patch-like diff visualization for user confirmation before + applying configuration overrides. Aggregates individual field previews + and tracks any issues like missing paths or type mismatches. + + Attributes: + field_previews: Tuple of individual field previews. + paths_not_found: Paths that don't exist in original config (will be created). + type_mismatches: Paths where new value type differs from original. + + Thread Safety: + Immutable (frozen=True) after creation - thread-safe for concurrent reads. + + Example: + >>> field_preview = ModelConfigOverrideFieldPreview( + ... path="retry.max_attempts", + ... injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ... old_value=3, + ... new_value=5, + ... ) + >>> preview = ModelConfigOverridePreview( + ... field_previews=(field_preview,), + ... paths_not_found=("new_setting",), + ... type_mismatches=(), + ... ) + >>> print(preview.to_markdown()) + ## Configuration Override Preview + ... + + .. versionadded:: 0.4.0 + """ + + # from_attributes=True: Enables construction from ORM/dataclass instances + # and ensures pytest-xdist compatibility across worker processes where + # class identity may differ due to independent imports. + # See CLAUDE.md "Pydantic from_attributes=True for Value Objects". + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + field_previews: tuple[ModelConfigOverrideFieldPreview, ...] = Field( + default_factory=tuple, + description="Individual field previews", + ) + paths_not_found: tuple[str, ...] = Field( + default_factory=tuple, + description="Paths that don't exist in original config (will be created)", + ) + type_mismatches: tuple[str, ...] = Field( + default_factory=tuple, + description="Paths where new value type differs from original", + ) + + def to_markdown(self) -> str: + """Generate markdown diff table. + + Creates a complete markdown document with: + - A header section + - A table showing all field overrides with before/after values + - Optional sections for new paths and type mismatches + + Returns: + A complete markdown string suitable for display to users. + """ + lines = [ + "## Configuration Override Preview", + "", + "| Path | Injection Point | Before | After | Status |", + "|------|-----------------|--------|-------|--------|", + ] + for preview in self.field_previews: + lines.append(preview.to_markdown_row()) + + if self.paths_not_found: + lines.extend(["", "### New Paths (will be created):", ""]) + for path in self.paths_not_found: + lines.append(f"- `{path}`") + + if self.type_mismatches: + lines.extend(["", "### Type Mismatches (review required):", ""]) + for path in self.type_mismatches: + lines.append(f"- `{path}`") + + return "\n".join(lines) diff --git a/src/omnibase_core/models/replay/model_config_override_result.py b/src/omnibase_core/models/replay/model_config_override_result.py new file mode 100644 index 0000000000..4426a31000 --- /dev/null +++ b/src/omnibase_core/models/replay/model_config_override_result.py @@ -0,0 +1,49 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Result model for override application. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +__all__ = ["ModelConfigOverrideResult"] + + +class ModelConfigOverrideResult(BaseModel): + """ + Result of applying configuration overrides. + + Contains the new (patched) configuration and application metadata. + Original config is NEVER modified - this contains a new copy. + + Thread Safety: + Immutable - the patched_config is a deep copy, safe to use concurrently. + """ + + # from_attributes=True: Enables construction from ORM/dataclass instances + # and ensures pytest-xdist compatibility across worker processes where + # class identity may differ due to independent imports. + # See CLAUDE.md "Pydantic from_attributes=True for Value Objects". + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + success: bool = Field(..., description="Whether all overrides were applied") + patched_config: Any = Field( + ..., description="New config with overrides applied (deep copy)" + ) + overrides_applied: int = Field( + default=0, description="Number of overrides successfully applied" + ) + paths_created: tuple[str, ...] = Field( + default_factory=tuple, + description="Paths that were created (did not exist before)", + ) + errors: tuple[str, ...] = Field( + default_factory=tuple, + description="Any errors during override application", + ) diff --git a/src/omnibase_core/models/replay/model_config_override_set.py b/src/omnibase_core/models/replay/model_config_override_set.py new file mode 100644 index 0000000000..3c54bcfc2c --- /dev/null +++ b/src/omnibase_core/models/replay/model_config_override_set.py @@ -0,0 +1,98 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Collection of configuration overrides for replay injection. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, +) +from omnibase_core.models.replay.model_config_override import ModelConfigOverride + +__all__ = ["ModelConfigOverrideSet"] + + +class ModelConfigOverrideSet(BaseModel): + """ + A collection of configuration overrides for a replay session. + + Groups overrides by injection point for efficient batch application. + + Thread Safety: + Immutable (frozen=True) after creation - thread-safe for concurrent reads. + """ + + # from_attributes=True: Enables construction from ORM/dataclass instances + # and ensures pytest-xdist compatibility across worker processes where + # class identity may differ due to independent imports. + # See CLAUDE.md "Pydantic from_attributes=True for Value Objects". + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + overrides: tuple[ModelConfigOverride, ...] = Field( + default_factory=tuple, + description="Ordered list of overrides to apply", + ) + + @property + def by_injection_point( + self, + ) -> dict[EnumOverrideInjectionPoint, list[ModelConfigOverride]]: + """Group overrides by their injection point for efficient batch application. + + Creates a dictionary mapping each injection point to the list of overrides + that should be applied at that point. This enables the injection service + to process overrides in batches per injection point rather than iterating + through all overrides for each injection point. + + Returns: + Dict mapping EnumOverrideInjectionPoint to list of ModelConfigOverride. + Only includes injection points that have at least one override. + + Example: + >>> override_set = ModelConfigOverrideSet(overrides=( + ... ModelConfigOverride(path="a", value=1, injection_point=HANDLER_CONFIG), + ... ModelConfigOverride(path="b", value=2, injection_point=ENVIRONMENT), + ... ModelConfigOverride(path="c", value=3, injection_point=HANDLER_CONFIG), + ... )) + >>> grouped = override_set.by_injection_point + >>> len(grouped[HANDLER_CONFIG]) + 2 + >>> len(grouped[ENVIRONMENT]) + 1 + """ + result: dict[EnumOverrideInjectionPoint, list[ModelConfigOverride]] = {} + for override in self.overrides: + result.setdefault(override.injection_point, []).append(override) + return result + + def with_override(self, override: ModelConfigOverride) -> ModelConfigOverrideSet: + """Return a new override set with an additional override appended. + + Implements immutable update pattern - creates a new instance rather than + modifying the existing one. This is consistent with frozen=True and + enables safe concurrent usage. + + Args: + override: The new override to append to the set. + + Returns: + A new ModelConfigOverrideSet with all existing overrides plus the + new override at the end. + + Example: + >>> original = ModelConfigOverrideSet(overrides=(override1,)) + >>> updated = original.with_override(override2) + >>> len(original.overrides) # Original unchanged + 1 + >>> len(updated.overrides) # New set has both + 2 + """ + return ModelConfigOverrideSet(overrides=(*self.overrides, override)) diff --git a/src/omnibase_core/models/replay/model_config_override_validation.py b/src/omnibase_core/models/replay/model_config_override_validation.py new file mode 100644 index 0000000000..f7980d54cf --- /dev/null +++ b/src/omnibase_core/models/replay/model_config_override_validation.py @@ -0,0 +1,137 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Validation result for configuration overrides. + +This module provides the ModelConfigOverrideValidation model which represents +validation results for config overrides, checking path existence, type +compatibility, and override consistency. + +Design: + ModelConfigOverrideValidation follows the violations/warnings/suggestions + pattern used by ServiceContractValidator and other validation components + in the codebase. This provides a consistent interface for reporting + validation outcomes with graduated severity levels. + + - **Violations**: Critical errors that must be fixed + - **Warnings**: Non-critical issues that should be addressed + - **Suggestions**: Optional improvements for better practices + +Architecture: + The validation model is used by config override injection to validate + override specifications before applying them to execution contexts:: + + ConfigOverrideInjector + | + +-- validate_overrides() + -> ModelConfigOverrideValidation + is_valid: bool + violations: tuple[str, ...] + warnings: tuple[str, ...] + suggestions: tuple[str, ...] + +Thread Safety: + ModelConfigOverrideValidation is frozen (immutable) after creation, + making it safe to share across threads. All fields use immutable + tuple types for collections. + +Usage: + .. code-block:: python + + from omnibase_core.models.replay import ModelConfigOverrideValidation + + # Valid result + result = ModelConfigOverrideValidation( + is_valid=True, + paths_validated=5, + type_checks_passed=5, + ) + + # Invalid result with violations + result = ModelConfigOverrideValidation( + is_valid=False, + violations=( + "Path 'config.missing' does not exist", + "Type mismatch: expected int, got str", + ), + warnings=("Config value 'timeout' is unusually high",), + paths_validated=5, + type_checks_passed=3, + ) + +Related: + - OMN-1205: Configuration Override Injection + - ServiceContractValidator: Uses same violations/warnings/suggestions pattern + - ModelReplayContext: Companion model for replay infrastructure + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from pydantic import BaseModel, ConfigDict, Field + + +class ModelConfigOverrideValidation(BaseModel): + """ + Validation result for a set of config overrides. + + Checks path existence, type compatibility, and override consistency. + Follows the violations/warnings/suggestions pattern from ServiceContractValidator. + + Attributes: + is_valid: Whether all overrides are valid (no violations). + violations: Critical validation errors that must be fixed. + warnings: Non-critical issues that should be addressed. + suggestions: Improvement recommendations (optional to fix). + paths_validated: Number of configuration paths checked. + type_checks_passed: Number of type compatibility checks passed. + + Thread Safety: + Immutable (frozen=True) after creation - thread-safe for concurrent reads. + + Example: + Creating a valid result:: + + result = ModelConfigOverrideValidation( + is_valid=True, + paths_validated=10, + type_checks_passed=10, + ) + + Creating an invalid result:: + + result = ModelConfigOverrideValidation( + is_valid=False, + violations=("Path 'x.y' not found",), + warnings=("Value exceeds recommended range",), + paths_validated=10, + type_checks_passed=8, + ) + + .. versionadded:: 0.4.0 + """ + + # from_attributes=True allows Pydantic to accept objects with matching + # attributes even when class identity differs (e.g., in pytest-xdist + # parallel execution where model classes are imported in separate workers). + # See CLAUDE.md section "Pydantic from_attributes=True for Value Objects". + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + is_valid: bool = Field(..., description="Whether all overrides are valid") + violations: tuple[str, ...] = Field( + default_factory=tuple, description="Critical validation errors" + ) + warnings: tuple[str, ...] = Field( + default_factory=tuple, description="Non-critical issues" + ) + suggestions: tuple[str, ...] = Field( + default_factory=tuple, description="Improvement recommendations" + ) + paths_validated: int = Field(default=0, description="Number of paths checked") + type_checks_passed: int = Field( + default=0, description="Number of type checks passed" + ) + + +__all__ = ["ModelConfigOverrideValidation"] diff --git a/src/omnibase_core/models/replay/model_replay_input.py b/src/omnibase_core/models/replay/model_replay_input.py new file mode 100644 index 0000000000..cbacfd8195 --- /dev/null +++ b/src/omnibase_core/models/replay/model_replay_input.py @@ -0,0 +1,83 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Replay input model that bundles original inputs with configuration overrides. + +This is the integration point for config override injection - overrides attach +to the REPLAY INPUT, not the manifest or handler config directly. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +from typing import Generic, TypeVar +from uuid import UUID, uuid4 + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.models.replay.model_config_override_set import ModelConfigOverrideSet + +T_Data = TypeVar("T_Data") + + +class ModelReplayInput(BaseModel, Generic[T_Data]): + """ + Input model for replay execution with optional configuration overrides. + + Bundles the original input data with any config overrides to apply during + replay. The overrides are applied at the START of replay execution, creating + patched copies of configurations - NEVER mutating originals. + + Key Invariant: + The original data (in `data` field) is NEVER modified. All overrides + produce new copies via immutable update patterns. + + Thread Safety: + Immutable (frozen=True) - safe for concurrent access. + + Example: + >>> from omnibase_core.models.replay import ModelReplayInput, ModelConfigOverrideSet + >>> replay_input = ModelReplayInput( + ... data=original_envelope, + ... config_overrides=ModelConfigOverrideSet(overrides=[ + ... ModelConfigOverride(path="llm.temperature", value=0.5), + ... ]), + ... ) + """ + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + input_id: UUID = Field( + default_factory=uuid4, + description="Unique identifier for this replay input", + ) + data: T_Data = Field( + ..., + description="The original input data for replay", + ) + config_overrides: ModelConfigOverrideSet | None = Field( + default=None, + description="Optional configuration overrides to apply during replay", + ) + + @property + def has_overrides(self) -> bool: + """Check if this input has any config overrides.""" + return ( + self.config_overrides is not None + and len(self.config_overrides.overrides) > 0 + ) + + def with_overrides( + self, overrides: ModelConfigOverrideSet + ) -> "ModelReplayInput[T_Data]": + """Return new input with overrides (immutable update).""" + return ModelReplayInput( + input_id=self.input_id, + data=self.data, + config_overrides=overrides, + ) + + +__all__ = ["ModelReplayInput"] diff --git a/src/omnibase_core/pipeline/__init__.py b/src/omnibase_core/pipeline/__init__.py index 06de07766a..2605545177 100644 --- a/src/omnibase_core/pipeline/__init__.py +++ b/src/omnibase_core/pipeline/__init__.py @@ -86,12 +86,12 @@ from omnibase_core.pipeline.manifest_logger import ManifestLogger from omnibase_core.pipeline.manifest_observer import ManifestObserver from omnibase_core.pipeline.registry_hook import RegistryHook -from omnibase_core.pipeline.replay import InjectorRNG from omnibase_core.pipeline.runner_pipeline import ( CANONICAL_PHASE_ORDER, HookCallable, RunnerPipeline, ) +from omnibase_core.services.replay.injector_rng import InjectorRNG __all__ = [ # Exceptions diff --git a/src/omnibase_core/pipeline/replay/__init__.py b/src/omnibase_core/pipeline/replay/__init__.py index c479386b80..40f9114367 100644 --- a/src/omnibase_core/pipeline/replay/__init__.py +++ b/src/omnibase_core/pipeline/replay/__init__.py @@ -2,16 +2,25 @@ # # SPDX-License-Identifier: Apache-2.0 """ -Replay infrastructure injectors, recorders, and executor. +Replay infrastructure injectors and executor. This module provides implementations for deterministic replay: -- **InjectorRNG**: RNG injection for deterministic replay -- **InjectorTime**: Time injection for deterministic replay -- **RecorderEffect**: Effect recording and replay for determinism +- **InjectorRNG**: RNG injection for deterministic replay (from services.replay) +- **InjectorTime**: Time injection for deterministic replay (from services.replay) +- **RecorderEffect**: Effect recording and replay for determinism (from services.replay) - **ExecutorReplay**: Replay executor orchestrating deterministic execution - **ReplaySession**: Active replay session with injected services +Note: + InjectorRNG, InjectorTime, and RecorderEffect are now located in + omnibase_core.services.replay and re-exported here for convenience. + For direct imports, use: + + from omnibase_core.services.replay.injector_rng import InjectorRNG + from omnibase_core.services.replay.injector_time import InjectorTime + from omnibase_core.services.replay.recorder_effect import RecorderEffect + Usage: >>> from omnibase_core.pipeline.replay import ( ... InjectorRNG, InjectorTime, RecorderEffect, ExecutorReplay, ReplaySession @@ -43,10 +52,10 @@ """ from omnibase_core.pipeline.replay.executor_replay import ExecutorReplay -from omnibase_core.pipeline.replay.injector_rng import InjectorRNG -from omnibase_core.pipeline.replay.injector_time import InjectorTime -from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect from omnibase_core.pipeline.replay.session_replay import ReplaySession +from omnibase_core.services.replay.injector_rng import InjectorRNG +from omnibase_core.services.replay.injector_time import InjectorTime +from omnibase_core.services.replay.recorder_effect import RecorderEffect __all__ = [ "ExecutorReplay", diff --git a/src/omnibase_core/pipeline/replay/executor_replay.py b/src/omnibase_core/pipeline/replay/executor_replay.py index 4d42e9908b..3707a4c723 100644 --- a/src/omnibase_core/pipeline/replay/executor_replay.py +++ b/src/omnibase_core/pipeline/replay/executor_replay.py @@ -91,10 +91,10 @@ from omnibase_core.enums.replay.enum_replay_mode import EnumReplayMode from omnibase_core.models.replay.model_effect_record import ModelEffectRecord from omnibase_core.models.replay.model_replay_context import ModelReplayContext -from omnibase_core.pipeline.replay.injector_rng import InjectorRNG -from omnibase_core.pipeline.replay.injector_time import InjectorTime -from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect from omnibase_core.pipeline.replay.session_replay import ReplaySession +from omnibase_core.services.replay.injector_rng import InjectorRNG +from omnibase_core.services.replay.injector_time import InjectorTime +from omnibase_core.services.replay.recorder_effect import RecorderEffect from omnibase_core.types.type_json import JsonType T = TypeVar("T") diff --git a/src/omnibase_core/pipeline/replay/session_replay.py b/src/omnibase_core/pipeline/replay/session_replay.py index 5c42ed0ca6..3ff94c6b96 100644 --- a/src/omnibase_core/pipeline/replay/session_replay.py +++ b/src/omnibase_core/pipeline/replay/session_replay.py @@ -63,9 +63,9 @@ from omnibase_core.models.replay.model_replay_context import ModelReplayContext if TYPE_CHECKING: - from omnibase_core.pipeline.replay.injector_rng import InjectorRNG - from omnibase_core.pipeline.replay.injector_time import InjectorTime - from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect + from omnibase_core.services.replay.injector_rng import InjectorRNG + from omnibase_core.services.replay.injector_time import InjectorTime + from omnibase_core.services.replay.recorder_effect import RecorderEffect @dataclass @@ -131,20 +131,20 @@ def mode(self) -> EnumReplayMode: def _default_time_service() -> InjectorTime: """Create default time service (deferred import to avoid circular deps).""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime return InjectorTime() def _default_rng_service() -> InjectorRNG: """Create default RNG service (deferred import to avoid circular deps).""" - from omnibase_core.pipeline.replay.injector_rng import InjectorRNG + from omnibase_core.services.replay.injector_rng import InjectorRNG return InjectorRNG() def _default_effect_recorder() -> RecorderEffect: """Create default effect recorder (deferred import to avoid circular deps).""" - from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect + from omnibase_core.services.replay.recorder_effect import RecorderEffect return RecorderEffect() diff --git a/src/omnibase_core/protocols/replay/protocol_effect_recorder.py b/src/omnibase_core/protocols/replay/protocol_effect_recorder.py index 7137f69a4d..0d65ed1378 100644 --- a/src/omnibase_core/protocols/replay/protocol_effect_recorder.py +++ b/src/omnibase_core/protocols/replay/protocol_effect_recorder.py @@ -21,7 +21,7 @@ .. code-block:: python from omnibase_core.protocols.replay import ProtocolEffectRecorder - from omnibase_core.pipeline.replay import RecorderEffect + from omnibase_core.services.replay.recorder_effect import RecorderEffect from omnibase_core.enums.replay import EnumRecorderMode # Production mode - pass-through (no recording) diff --git a/src/omnibase_core/protocols/replay/protocol_rng_service.py b/src/omnibase_core/protocols/replay/protocol_rng_service.py index 1178ae32b2..14dcf46319 100644 --- a/src/omnibase_core/protocols/replay/protocol_rng_service.py +++ b/src/omnibase_core/protocols/replay/protocol_rng_service.py @@ -90,7 +90,7 @@ class ProtocolRNGService(Protocol): .. code-block:: python from omnibase_core.protocols.replay import ProtocolRNGService - from omnibase_core.pipeline.replay import InjectorRNG + from omnibase_core.services.replay.injector_rng import InjectorRNG # Create with specific seed for replay rng: ProtocolRNGService = InjectorRNG(seed=42) diff --git a/src/omnibase_core/protocols/replay/protocol_time_service.py b/src/omnibase_core/protocols/replay/protocol_time_service.py index 67677a14cb..92f73d50aa 100644 --- a/src/omnibase_core/protocols/replay/protocol_time_service.py +++ b/src/omnibase_core/protocols/replay/protocol_time_service.py @@ -20,7 +20,7 @@ .. code-block:: python from omnibase_core.protocols.replay import ProtocolTimeService - from omnibase_core.pipeline.replay import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime from datetime import datetime, timezone # Production mode - returns current time diff --git a/src/omnibase_core/services/replay/__init__.py b/src/omnibase_core/services/replay/__init__.py new file mode 100644 index 0000000000..a72d11322a --- /dev/null +++ b/src/omnibase_core/services/replay/__init__.py @@ -0,0 +1,29 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Replay services module. + +This module contains services for the replay infrastructure including: + +- **InjectorRNG**: RNG injection for deterministic replay +- **InjectorTime**: Time injection for deterministic replay +- **RecorderEffect**: Effect recording and replay for determinism +- **ServiceConfigOverrideInjector**: Configuration override injection + +Note: Following OMN-1071 policy, services are NOT exported at package level. +Import directly from the specific service module: + + from omnibase_core.services.replay.injector_rng import InjectorRNG + from omnibase_core.services.replay.injector_time import InjectorTime + from omnibase_core.services.replay.recorder_effect import RecorderEffect + from omnibase_core.services.replay.service_config_override_injector import ( + ServiceConfigOverrideInjector, + ) + +.. versionadded:: 0.4.0 + Added Replay Infrastructure (OMN-1116) + Added Configuration Override Injection (OMN-1205) +""" + +__all__: list[str] = [] diff --git a/src/omnibase_core/pipeline/replay/injector_rng.py b/src/omnibase_core/services/replay/injector_rng.py similarity index 98% rename from src/omnibase_core/pipeline/replay/injector_rng.py rename to src/omnibase_core/services/replay/injector_rng.py index 07ba3fa045..7fbe107a3a 100644 --- a/src/omnibase_core/pipeline/replay/injector_rng.py +++ b/src/omnibase_core/services/replay/injector_rng.py @@ -15,7 +15,7 @@ Usage: .. code-block:: python - from omnibase_core.pipeline.replay import InjectorRNG + from omnibase_core.services.replay.injector_rng import InjectorRNG # Replay mode: use specific seed rng = InjectorRNG(seed=42) diff --git a/src/omnibase_core/pipeline/replay/injector_time.py b/src/omnibase_core/services/replay/injector_time.py similarity index 98% rename from src/omnibase_core/pipeline/replay/injector_time.py rename to src/omnibase_core/services/replay/injector_time.py index 3711239c3b..8a946c7e8a 100644 --- a/src/omnibase_core/pipeline/replay/injector_time.py +++ b/src/omnibase_core/services/replay/injector_time.py @@ -15,7 +15,7 @@ Usage: .. code-block:: python - from omnibase_core.pipeline.replay import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime from datetime import datetime, timezone # Production mode: returns current time diff --git a/src/omnibase_core/pipeline/replay/recorder_effect.py b/src/omnibase_core/services/replay/recorder_effect.py similarity index 99% rename from src/omnibase_core/pipeline/replay/recorder_effect.py rename to src/omnibase_core/services/replay/recorder_effect.py index 249bc13eae..98c53df561 100644 --- a/src/omnibase_core/pipeline/replay/recorder_effect.py +++ b/src/omnibase_core/services/replay/recorder_effect.py @@ -22,7 +22,7 @@ Usage: .. code-block:: python - from omnibase_core.pipeline.replay import RecorderEffect + from omnibase_core.services.replay.recorder_effect import RecorderEffect from omnibase_core.enums.replay import EnumRecorderMode # Production mode (default): pass-through, no recording @@ -121,7 +121,7 @@ class RecorderEffect: is_replaying: Whether the recorder is in replay mode. Example: - >>> from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect + >>> from omnibase_core.services.replay.recorder_effect import RecorderEffect >>> from omnibase_core.enums.replay import EnumRecorderMode >>> # Production mode (default) >>> recorder = RecorderEffect() diff --git a/src/omnibase_core/services/replay/sentinel_missing.py b/src/omnibase_core/services/replay/sentinel_missing.py new file mode 100644 index 0000000000..0be83ae46b --- /dev/null +++ b/src/omnibase_core/services/replay/sentinel_missing.py @@ -0,0 +1,47 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Sentinel value for missing configuration paths. + +This module provides a singleton sentinel to distinguish "path not found" +from "value is None" during config override traversal. + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + + +class _Missing: + """Singleton sentinel class for missing values (distinct from None). + + Used during path traversal to distinguish between: + - Path exists but value is None + - Path does not exist (returns MISSING) + + Thread Safety: + Singleton pattern is thread-safe for reads. The instance is created + at module load time. + """ + + _instance: "_Missing | None" = None + + def __new__(cls) -> "_Missing": + """Return singleton instance.""" + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __repr__(self) -> str: + """Return string representation.""" + return "" + + def __bool__(self) -> bool: + """Always return False for truthiness checks.""" + return False + + +# Module-level singleton instance +MISSING = _Missing() + +__all__ = ["MISSING"] diff --git a/src/omnibase_core/services/replay/service_config_override_injector.py b/src/omnibase_core/services/replay/service_config_override_injector.py new file mode 100644 index 0000000000..50da1775ef --- /dev/null +++ b/src/omnibase_core/services/replay/service_config_override_injector.py @@ -0,0 +1,889 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +"""Service for injecting configuration overrides during replay. + +This module provides the ServiceConfigOverrideInjector class for validating, +previewing, and applying configuration overrides to handler configurations, +environment variables, and replay contexts. + +Thread Safety: + ServiceConfigOverrideInjector is stateless (only strict_mode set at init) + and thread-safe. All methods are pure functions that do not modify shared + state. The apply() method always deep-copies configurations before + modification, ensuring the original is never mutated. + +Design: + The service uses a MISSING sentinel to distinguish between: + - Path not found (returns MISSING) + - Path exists but value is None (returns None) + + This enables accurate validation and preview generation. + +Architecture: + Configuration override injection operates at three injection points: + + 1. HANDLER_CONFIG: Patches handler configuration models (Pydantic or dict) + before handler execution begins. + + 2. ENVIRONMENT: Creates an environment variable overlay dict without + mutating os.environ. Caller is responsible for applying the overlay. + + 3. CONTEXT: Patches replay context fields on a copy, not the original. + +Example: + >>> from omnibase_core.services.replay.service_config_override_injector import ( + ... ServiceConfigOverrideInjector, + ... ) + >>> from omnibase_core.models.replay import ModelConfigOverride, ModelConfigOverrideSet + >>> from omnibase_core.enums.replay import EnumOverrideInjectionPoint + >>> + >>> injector = ServiceConfigOverrideInjector() + >>> override = ModelConfigOverride( + ... path="retry.max_attempts", + ... value=5, + ... injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ... ) + >>> override_set = ModelConfigOverrideSet(overrides=(override,)) + >>> config = {"retry": {"max_attempts": 3}} + >>> + >>> # Validate overrides + >>> validation = injector.validate(override_set, config) + >>> print(validation.is_valid) + True + >>> + >>> # Preview changes + >>> preview = injector.preview(override_set, config) + >>> print(preview.to_markdown()) + ... + >>> + >>> # Apply overrides + >>> result = injector.apply(override_set, config) + >>> print(result.patched_config) + {'retry': {'max_attempts': 5}} + +Related: + - OMN-1205: Configuration Override Injection + - ModelConfigOverrideSet: Collection of overrides + - ModelReplayContext: Replay context for deterministic execution + +.. versionadded:: 0.4.0 + Added Configuration Override Injection (OMN-1205) +""" + +import copy +import logging +import re +import types +from collections.abc import Mapping +from typing import Any + +from pydantic import BaseModel + +from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, +) +from omnibase_core.models.replay.model_config_override_field_preview import ( + ModelConfigOverrideFieldPreview, +) +from omnibase_core.models.replay.model_config_override_preview import ( + ModelConfigOverridePreview, +) +from omnibase_core.models.replay.model_config_override_result import ( + ModelConfigOverrideResult, +) +from omnibase_core.models.replay.model_config_override_set import ( + ModelConfigOverrideSet, +) +from omnibase_core.models.replay.model_config_override_validation import ( + ModelConfigOverrideValidation, +) +from omnibase_core.services.replay.sentinel_missing import MISSING + +logger = logging.getLogger(__name__) + + +class ServiceConfigOverrideInjector: + """Service for validating, previewing, and applying configuration overrides. + + Provides validate/preview/apply operations for config overrides during replay. + All operations are stateless and thread-safe. + + Attributes: + strict_mode: If True, validation fails on any warning. Default False. + MAX_PATH_DEPTH: Maximum allowed path depth to prevent DoS attacks. + + Thread Safety: + This service is stateless (only strict_mode config set at init). + All methods are pure functions that do not modify shared state. + The apply() method always deep-copies before modification. + + Example: + >>> injector = ServiceConfigOverrideInjector() + >>> result = injector.validate(override_set, config) + >>> if result.is_valid: + ... applied = injector.apply(override_set, config) + """ + + # Maximum path depth to prevent DoS via deeply nested paths + MAX_PATH_DEPTH: int = 20 + + def __init__(self, strict_mode: bool = False) -> None: + """Initialize the config override injector. + + Args: + strict_mode: If True, validation fails on any warning. + Default False allows warnings without failing validation. + """ + self.strict_mode = strict_mode + + def _is_frozen_pydantic_model(self, obj: Any) -> bool: + """Check if an object is a frozen Pydantic model. + + Frozen Pydantic models cannot be modified after creation. This method + detects frozen models by checking the model_config attribute for + frozen=True setting. + + Args: + obj: The object to check. + + Returns: + True if the object is a frozen Pydantic BaseModel, False otherwise. + + Examples: + >>> from pydantic import BaseModel, ConfigDict + >>> class FrozenModel(BaseModel): + ... model_config = ConfigDict(frozen=True) + ... value: int + >>> injector = ServiceConfigOverrideInjector() + >>> injector._is_frozen_pydantic_model(FrozenModel(value=1)) + True + >>> class MutableModel(BaseModel): + ... value: int + >>> injector._is_frozen_pydantic_model(MutableModel(value=1)) + False + """ + if not isinstance(obj, BaseModel): + return False + + # Check model_config for frozen setting + model_config = getattr(obj, "model_config", None) + if model_config is None: + return False + + # model_config can be a dict or ConfigDict + if isinstance(model_config, dict): + return model_config.get("frozen", False) is True + + # For ConfigDict or other mapping types + if isinstance(model_config, Mapping): + return model_config.get("frozen", False) is True + + return False + + def _is_valid_path_syntax(self, path: str) -> tuple[bool, str]: + """Validate dot-path syntax. + + Checks that the path follows valid dot-notation syntax: + - Segments separated by dots + - Each segment is a valid identifier or numeric index + - Path depth does not exceed MAX_PATH_DEPTH + + Args: + path: The dot-separated path to validate. + + Returns: + Tuple of (is_valid, error_message). If valid, error_message is empty. + + Examples: + >>> injector = ServiceConfigOverrideInjector() + >>> injector._is_valid_path_syntax("config.timeout") + (True, '') + >>> injector._is_valid_path_syntax("items.0.name") + (True, '') + >>> injector._is_valid_path_syntax("") + (False, 'Path cannot be empty') + """ + if not path: + return (False, "Path cannot be empty") + + if not path.strip(): + return (False, "Path cannot be whitespace only") + + parts = path.split(".") + if len(parts) > self.MAX_PATH_DEPTH: + return ( + False, + f"Path depth {len(parts)} exceeds maximum {self.MAX_PATH_DEPTH}", + ) + + # Check each segment + for i, part in enumerate(parts): + if not part: + return (False, f"Empty segment at position {i}") + + # Allow numeric segments for array indices + if part.isdigit(): + continue + + # Check identifier format for non-numeric segments + if not re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", part): + return ( + False, + f"Invalid path segment '{part}' at position {i}: " + "must be a valid identifier or numeric index", + ) + + return (True, "") + + def _get_value_at_path(self, obj: Any, path: str | None) -> Any: + """Get value at a dot-separated path. + + Traverses nested structures (dict, Pydantic model, list/tuple) to + retrieve the value at the specified path. Returns MISSING sentinel + if the path does not exist. + + Args: + obj: The object to traverse (dict, Pydantic model, or list). + path: The dot-separated path to the value. + + Returns: + The value at the path, or MISSING if the path does not exist. + + Thread Safety: + This is a pure function that does not modify the input object. + + Examples: + >>> injector = ServiceConfigOverrideInjector() + >>> data = {"config": {"timeout": 30}} + >>> injector._get_value_at_path(data, "config.timeout") + 30 + >>> injector._get_value_at_path(data, "config.missing") + + """ + # Handle None/invalid inputs gracefully + if obj is None or path is None: + return MISSING + + try: + parts = path.split(".") + except AttributeError: + # boundary-ok: path is not a string-like object + return MISSING + + current: Any = obj + + for part in parts: + if current is MISSING: + return MISSING + + try: + if isinstance(current, dict): + current = current.get(part, MISSING) + elif isinstance(current, BaseModel): + current = getattr(current, part, MISSING) + elif isinstance(current, (list, tuple)) and part.isdigit(): + idx = int(part) + if 0 <= idx < len(current): + current = current[idx] + else: + current = MISSING + else: + # Try getattr for other objects + current = getattr(current, part, MISSING) + except (AttributeError, IndexError, KeyError, TypeError) as e: + # boundary-ok: handle edge cases like non-subscriptable types + logger.debug( + "Path traversal failed at '%s': %s", + part, + str(e), + ) + return MISSING + + return current + + def _set_value_at_path(self, obj: Any, path: str, value: Any) -> tuple[bool, str]: + """Set value at a dot-separated path. + + Traverses nested structures and sets the value at the specified path. + Creates intermediate dictionaries if they don't exist. + + Args: + obj: The object to modify (must be a dict or mutable structure). + path: The dot-separated path to set. + value: The value to set. + + Returns: + Tuple of (success, error_message). If successful, error_message is empty. + + Warning: + This method modifies the input object in place. Always use on a + deep copy to avoid mutating the original. + + Examples: + >>> injector = ServiceConfigOverrideInjector() + >>> data = {"config": {"timeout": 30}} + >>> injector._set_value_at_path(data, "config.timeout", 60) + (True, '') + >>> data["config"]["timeout"] + 60 + """ + # Handle None object gracefully + if obj is None: + return (False, "Cannot set value on None object") + + parts = path.split(".") + + current: Any = obj + + # Navigate to parent of target + for i, part in enumerate(parts[:-1]): + try: + if isinstance(current, dict): + if part not in current: + # Create intermediate dict + current[part] = {} + current = current[part] + elif isinstance(current, BaseModel): + # Pydantic models are typically frozen, need special handling + child = getattr(current, part, MISSING) + if child is MISSING: + return ( + False, + f"Cannot create path segment '{part}' in frozen Pydantic model", + ) + current = child + elif isinstance(current, (list, tuple)) and part.isdigit(): + idx = int(part) + if 0 <= idx < len(current): + current = current[idx] + else: + return ( + False, + f"Index {idx} out of bounds for list at '{'.'.join(parts[:i])}'", + ) + else: + return ( + False, + f"Cannot traverse non-container type {type(current).__name__} " + f"at path segment '{part}'", + ) + except (AttributeError, IndexError, KeyError, TypeError) as e: + # boundary-ok: handle edge cases during path navigation + return ( + False, + f"Path navigation failed at '{part}': {e}", + ) + + # Set the final value + final_part = parts[-1] + try: + if isinstance(current, dict): + current[final_part] = value + return (True, "") + elif isinstance(current, BaseModel): + # Pydantic models: need to check if field exists and is settable + if not hasattr(current, final_part): + return ( + False, + f"Field '{final_part}' does not exist on Pydantic model " + f"{type(current).__name__}", + ) + try: + setattr(current, final_part, value) + return (True, "") + except (AttributeError, TypeError, ValueError) as e: + # catch-all-ok: Pydantic models may reject values for various reasons + return ( + False, + f"Cannot set field '{final_part}' on Pydantic model: {e}", + ) + elif isinstance(current, list) and final_part.isdigit(): + idx = int(final_part) + if 0 <= idx < len(current): + current[idx] = value + return (True, "") + else: + return ( + False, + f"Index {idx} out of bounds for list", + ) + else: + return ( + False, + f"Cannot set value on {type(current).__name__}", + ) + except (AttributeError, IndexError, KeyError, TypeError) as e: + # boundary-ok: handle edge cases during value assignment + return ( + False, + f"Failed to set value at '{final_part}': {e}", + ) + + def _deep_copy_config(self, config: Any) -> Any: + """Deep copy a configuration object. + + Handles both dict and Pydantic model configurations. + For mutable Pydantic models, uses model_copy(deep=True). + For frozen Pydantic models, converts to dict since frozen models + cannot be modified after creation. + + Args: + config: The configuration to copy. + + Returns: + A deep copy of the configuration. For frozen Pydantic models, + returns a dict representation that can be modified. + + Thread Safety: + This is a pure function that creates a new object. + + Note: + Frozen Pydantic models are detected early and converted to dicts + to allow modification. This is intentional - if you need to apply + overrides to a frozen model, the result must be a dict since the + original frozen model's copy would also be frozen. + """ + if isinstance(config, BaseModel): + # Check for frozen models first - they need special handling + if self._is_frozen_pydantic_model(config): + # Frozen models cannot be modified, so convert to dict + # This allows overrides to be applied to the dict representation + logger.debug( + "Converting frozen Pydantic model %s to dict for override application", + type(config).__name__, + ) + return copy.deepcopy(config.model_dump()) + # Mutable Pydantic model - use model_copy + return config.model_copy(deep=True) + else: + return copy.deepcopy(config) + + def validate( + self, + overrides: ModelConfigOverrideSet | None, + config: Any, + injection_point: EnumOverrideInjectionPoint | None = None, + ) -> ModelConfigOverrideValidation: + """Validate configuration overrides against a config. + + Checks path existence, type compatibility, and override consistency. + Does NOT modify the original configuration. + + Args: + overrides: The set of overrides to validate. If None, returns + a validation result with is_valid=False. + config: The target configuration to validate against. If None, + all paths will be marked as non-existent. + injection_point: Optional filter to only validate overrides for + a specific injection point. If None, validates all overrides. + + Returns: + ModelConfigOverrideValidation with validation results. + + Thread Safety: + Pure function - does not modify any state. + + Example: + >>> validation = injector.validate(override_set, config) + >>> if validation.is_valid: + ... print("All overrides are valid") + >>> else: + ... for v in validation.violations: + ... print(f"Error: {v}") + """ + # Handle None overrides gracefully + if overrides is None: + return ModelConfigOverrideValidation( + is_valid=False, + violations=("Overrides cannot be None",), + warnings=(), + suggestions=(), + paths_validated=0, + type_checks_passed=0, + ) + + violations: list[str] = [] + warnings: list[str] = [] + suggestions: list[str] = [] + paths_validated = 0 + type_checks_passed = 0 + + # Handle None config - all paths will be treated as non-existent + if config is None: + warnings.append("Config is None - all paths will be treated as new") + + # Filter overrides by injection point if specified + overrides_to_check = overrides.overrides + if injection_point is not None: + overrides_to_check = tuple( + o for o in overrides.overrides if o.injection_point == injection_point + ) + + for override in overrides_to_check: + paths_validated += 1 + + # Validate path syntax + is_valid_syntax, syntax_error = self._is_valid_path_syntax(override.path) + if not is_valid_syntax: + violations.append( + f"Invalid path syntax '{override.path}': {syntax_error}" + ) + continue + + # Check if path exists + current_value = self._get_value_at_path(config, override.path) + + if current_value is MISSING: + # Path doesn't exist - this could be creating a new path + if override.injection_point == EnumOverrideInjectionPoint.ENVIRONMENT: + # Environment overrides can create new keys + suggestions.append( + f"Path '{override.path}' will be created as new environment variable" + ) + else: + warnings.append( + f"Path '{override.path}' does not exist in config (will be created)" + ) + else: + # Path exists - check type compatibility + current_type = type(current_value).__name__ + new_type = type(override.value).__name__ + + if current_type != new_type: + # Type mismatch + if override.value is None: + # Setting to None is usually acceptable + suggestions.append( + f"Path '{override.path}' will be set to None " + f"(was {current_type})" + ) + elif isinstance(current_value, (int, float)) and isinstance( + override.value, (int, float) + ): + # Numeric type conversion is acceptable + type_checks_passed += 1 + else: + warnings.append( + f"Type mismatch at '{override.path}': " + f"changing from {current_type} to {new_type}" + ) + else: + type_checks_passed += 1 + + # Determine overall validity + is_valid = len(violations) == 0 + if self.strict_mode and warnings: + is_valid = False + + return ModelConfigOverrideValidation( + is_valid=is_valid, + violations=tuple(violations), + warnings=tuple(warnings), + suggestions=tuple(suggestions), + paths_validated=paths_validated, + type_checks_passed=type_checks_passed, + ) + + def preview( + self, + overrides: ModelConfigOverrideSet | None, + config: Any, + injection_point: EnumOverrideInjectionPoint | None = None, + ) -> ModelConfigOverridePreview: + """Preview configuration changes before applying. + + Generates a patch-like diff showing before/after state for each + override without modifying the original configuration. + + Args: + overrides: The set of overrides to preview. If None, returns + an empty preview. + config: The target configuration to preview against. If None, + all paths will be marked as not existing. + injection_point: Optional filter to only preview overrides for + a specific injection point. If None, previews all overrides. + + Returns: + ModelConfigOverridePreview with before/after previews. + + Thread Safety: + Pure function - does not modify any state. + + Example: + >>> preview = injector.preview(override_set, config) + >>> print(preview.to_markdown()) + ## Configuration Override Preview + | Path | Injection Point | Before | After | Status | + ... + """ + # Handle None overrides gracefully + if overrides is None: + return ModelConfigOverridePreview( + field_previews=(), + paths_not_found=(), + type_mismatches=(), + ) + + field_previews: list[ModelConfigOverrideFieldPreview] = [] + paths_not_found: list[str] = [] + type_mismatches: list[str] = [] + + # Filter overrides by injection point if specified + overrides_to_preview = overrides.overrides + if injection_point is not None: + overrides_to_preview = tuple( + o for o in overrides.overrides if o.injection_point == injection_point + ) + + for override in overrides_to_preview: + # Validate path syntax first + is_valid_syntax, _ = self._is_valid_path_syntax(override.path) + if not is_valid_syntax: + # Skip invalid paths in preview + continue + + # Get current value + current_value = self._get_value_at_path(config, override.path) + path_exists = current_value is not MISSING + + # Track paths not found + if not path_exists: + paths_not_found.append(override.path) + old_value: Any = None # Use None for display + else: + old_value = current_value + + # Check for type mismatch + if path_exists and current_value is not None: + current_type = type(current_value).__name__ + new_type = type(override.value).__name__ + if current_type != new_type: + # Allow numeric conversions + if not ( + isinstance(current_value, (int, float)) + and isinstance(override.value, (int, float)) + ): + type_mismatches.append( + f"{override.path}: {current_type} -> {new_type}" + ) + + # Create field preview + field_preview = ModelConfigOverrideFieldPreview( + path=override.path, + injection_point=override.injection_point, + old_value=old_value, + new_value=override.value, + value_type=type(override.value).__name__, + path_exists=path_exists, + ) + field_previews.append(field_preview) + + return ModelConfigOverridePreview( + field_previews=tuple(field_previews), + paths_not_found=tuple(paths_not_found), + type_mismatches=tuple(type_mismatches), + ) + + def apply( + self, + overrides: ModelConfigOverrideSet | None, + config: Any, + injection_point: EnumOverrideInjectionPoint | None = None, + ) -> ModelConfigOverrideResult: + """Apply configuration overrides to a config. + + Creates a deep copy of the configuration and applies all overrides. + The original configuration is NEVER modified. + + Args: + overrides: The set of overrides to apply. If None, returns + an error result. + config: The target configuration to patch. Will NOT be modified. + If None, returns an error result. + injection_point: Optional filter to only apply overrides for + a specific injection point. If None, applies all overrides. + + Returns: + ModelConfigOverrideResult with the patched configuration. + + Thread Safety: + Pure function - creates a deep copy before modification. + The original config is never mutated. + + Example: + >>> result = injector.apply(override_set, config) + >>> if result.success: + ... new_config = result.patched_config + ... print(f"Applied {result.overrides_applied} overrides") + """ + # Handle None overrides gracefully + if overrides is None: + return ModelConfigOverrideResult( + success=False, + patched_config=config, + overrides_applied=0, + paths_created=(), + errors=("Overrides cannot be None",), + ) + + # Handle None config gracefully + if config is None: + return ModelConfigOverrideResult( + success=False, + patched_config=None, + overrides_applied=0, + paths_created=(), + errors=("Config cannot be None",), + ) + + # Deep copy the config to avoid mutating the original + patched_config = self._deep_copy_config(config) + + errors: list[str] = [] + paths_created: list[str] = [] + overrides_applied = 0 + + # Filter overrides by injection point if specified + overrides_to_apply = overrides.overrides + if injection_point is not None: + overrides_to_apply = tuple( + o for o in overrides.overrides if o.injection_point == injection_point + ) + + for override in overrides_to_apply: + # Validate path syntax + is_valid_syntax, syntax_error = self._is_valid_path_syntax(override.path) + if not is_valid_syntax: + errors.append(f"Invalid path '{override.path}': {syntax_error}") + continue + + # Check if path exists (for tracking paths_created) + current_value = self._get_value_at_path(patched_config, override.path) + path_existed = current_value is not MISSING + + # Apply the override + success, error = self._set_value_at_path( + patched_config, override.path, override.value + ) + + if success: + overrides_applied += 1 + if not path_existed: + paths_created.append(override.path) + else: + errors.append(f"Failed to set '{override.path}': {error}") + + return ModelConfigOverrideResult( + success=len(errors) == 0, + patched_config=patched_config, + overrides_applied=overrides_applied, + paths_created=tuple(paths_created), + errors=tuple(errors), + ) + + def apply_environment_overlay( + self, + overrides: ModelConfigOverrideSet | None, + ) -> types.MappingProxyType[str, str]: + """Create an immutable environment variable overlay from overrides. + + Extracts ENVIRONMENT injection point overrides and creates an immutable + mapping that can be used as an environment overlay. Does NOT mutate + os.environ. + + The caller is responsible for applying the overlay to their environment + context (e.g., passing to subprocess or using with os.environ.update()). + + Environment Variable Naming Derivation: + The environment variable name is derived from the override path + using the **last segment** of the dot-separated path, converted + to **uppercase**. This allows for logical grouping in paths while + producing conventional environment variable names. + + Path Segment Extraction:: + + path.split(".")[-1].upper() + + Examples: + - "timeout" -> "TIMEOUT" + - "llm.temperature" -> "TEMPERATURE" + - "config.retry.count" -> "COUNT" + - "my_setting" -> "MY_SETTING" + + Rationale: + - Paths may be hierarchical for organization/grouping purposes + - Environment variable names are traditionally SCREAMING_CASE + - Using the last segment prevents overly long env var names + - Underscores in path segments are preserved (my_var -> MY_VAR) + + Value Conversion: + - None values become empty strings ("") + - Boolean True becomes "true", False becomes "false" + - All other values use str() conversion + + Args: + overrides: The set of overrides to extract environment variables from. + If None, returns an empty mapping. + + Returns: + Immutable mapping of environment variable names to string values. + Only includes overrides with injection_point=ENVIRONMENT. + Returns MappingProxyType to prevent callers from mutating the result. + + Thread Safety: + Pure function - returns an immutable mapping, never mutates os.environ. + + Example: + >>> from omnibase_core.models.replay import ModelConfigOverride, ModelConfigOverrideSet + >>> from omnibase_core.enums.replay import EnumOverrideInjectionPoint + >>> override = ModelConfigOverride( + ... path="llm.openai.api_key", # Grouped path + ... value="sk-test-123", + ... injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ... ) + >>> override_set = ModelConfigOverrideSet(overrides=(override,)) + >>> injector = ServiceConfigOverrideInjector() + >>> overlay = injector.apply_environment_overlay(override_set) + >>> dict(overlay) + {'API_KEY': 'sk-test-123'} # Last segment, uppercased + >>> + >>> # Pass to subprocess (dict() creates a mutable copy if needed) + >>> subprocess.run(cmd, env={**os.environ, **overlay}) + >>> + >>> # Or use as context manager (user-provided) + >>> with env_context(overlay): + ... run_handler() + >>> + >>> # Attempting to mutate raises TypeError + >>> overlay["NEW_VAR"] = "value" # Raises TypeError + """ + # Handle None overrides gracefully + if overrides is None: + return types.MappingProxyType({}) + + env_overlay: dict[str, str] = {} + + for override in overrides.overrides: + if override.injection_point != EnumOverrideInjectionPoint.ENVIRONMENT: + continue + + # Environment variable names are the path (or last segment) + # Path can be the full env var name or nested for grouping + parts = override.path.split(".") + env_name = parts[-1].upper() # Use last segment, uppercase by convention + + # Convert value to string + if override.value is None: + value_str = "" + elif isinstance(override.value, bool): + value_str = "true" if override.value else "false" + else: + value_str = str(override.value) + + env_overlay[env_name] = value_str + + return types.MappingProxyType(env_overlay) + + +__all__ = ["ServiceConfigOverrideInjector", "MISSING"] diff --git a/tests/unit/models/replay/test_model_config_overrides.py b/tests/unit/models/replay/test_model_config_overrides.py new file mode 100644 index 0000000000..2f8e14ed21 --- /dev/null +++ b/tests/unit/models/replay/test_model_config_overrides.py @@ -0,0 +1,468 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Unit tests for ModelConfigOverride and ModelConfigOverrideSet. + +Tests cover: +- Model immutability (frozen) +- Field validation (path min_length) +- Default values +- Injection point grouping (by_injection_point property) +- Immutable update pattern (with_override method) +- Serialization roundtrip (JSON export/import) + +.. versionadded:: 0.4.0 + Added as part of Configuration Override Injection (OMN-1205) +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest +from pydantic import ValidationError + +if TYPE_CHECKING: + from omnibase_core.models.replay import ModelConfigOverride + + +@pytest.fixture +def handler_config_override() -> ModelConfigOverride: + """Create a sample handler config override.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + from omnibase_core.models.replay import ModelConfigOverride + + return ModelConfigOverride( + path="llm.temperature", + value=0.7, + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ) + + +@pytest.fixture +def environment_override() -> ModelConfigOverride: + """Create a sample environment override.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + from omnibase_core.models.replay import ModelConfigOverride + + return ModelConfigOverride( + path="API_KEY", + value="secret-key-123", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ) + + +@pytest.fixture +def context_override() -> ModelConfigOverride: + """Create a sample context override.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + from omnibase_core.models.replay import ModelConfigOverride + + return ModelConfigOverride( + path="replay.timeout_seconds", + value=60, + injection_point=EnumOverrideInjectionPoint.CONTEXT, + ) + + +@pytest.mark.unit +class TestModelConfigOverrideCreation: + """Test ModelConfigOverride creation and defaults.""" + + def test_create_with_defaults(self) -> None: + """Can create override with default injection point.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + from omnibase_core.models.replay import ModelConfigOverride + + override = ModelConfigOverride(path="test.path", value=42) + assert override.path == "test.path" + assert override.value == 42 + assert override.injection_point == EnumOverrideInjectionPoint.HANDLER_CONFIG + + def test_create_with_explicit_injection_point(self) -> None: + """Can specify injection point explicitly.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + from omnibase_core.models.replay import ModelConfigOverride + + override = ModelConfigOverride( + path="API_KEY", + value="secret", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ) + assert override.injection_point == EnumOverrideInjectionPoint.ENVIRONMENT + + def test_create_with_complex_value(self) -> None: + """Can create override with complex nested value.""" + from omnibase_core.models.replay import ModelConfigOverride + + complex_value: dict[str, Any] = { + "nested": {"key": "value"}, + "list": [1, 2, 3], + } + override = ModelConfigOverride(path="config.section", value=complex_value) + assert override.value == complex_value + + def test_create_with_none_value(self) -> None: + """Can create override with None value (to clear a config).""" + from omnibase_core.models.replay import ModelConfigOverride + + override = ModelConfigOverride(path="optional.field", value=None) + assert override.value is None + + +@pytest.mark.unit +class TestModelConfigOverrideImmutability: + """Test ModelConfigOverride immutability characteristics.""" + + def test_model_is_frozen( + self, handler_config_override: ModelConfigOverride + ) -> None: + """Test that ModelConfigOverride is frozen (immutable).""" + with pytest.raises(ValidationError): + handler_config_override.path = "modified.path" + + def test_cannot_modify_value( + self, handler_config_override: ModelConfigOverride + ) -> None: + """Test that value field cannot be reassigned.""" + with pytest.raises(ValidationError): + handler_config_override.value = 999 + + def test_cannot_modify_injection_point( + self, handler_config_override: ModelConfigOverride + ) -> None: + """Test that injection_point field cannot be reassigned.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + + with pytest.raises(ValidationError): + handler_config_override.injection_point = ( + EnumOverrideInjectionPoint.ENVIRONMENT + ) + + +@pytest.mark.unit +class TestModelConfigOverrideValidation: + """Test ModelConfigOverride field validation.""" + + def test_path_min_length_empty_string(self) -> None: + """Path must be at least 1 character - empty string rejected.""" + from omnibase_core.models.replay import ModelConfigOverride + + with pytest.raises(ValidationError) as exc_info: + ModelConfigOverride(path="", value=1) + + assert "path" in str(exc_info.value).lower() + + def test_path_min_length_single_char_valid(self) -> None: + """Path with single character is valid.""" + from omnibase_core.models.replay import ModelConfigOverride + + override = ModelConfigOverride(path="x", value=1) + assert override.path == "x" + + def test_extra_fields_forbidden(self) -> None: + """Test that extra fields are forbidden.""" + from omnibase_core.models.replay import ModelConfigOverride + + with pytest.raises(ValidationError) as exc_info: + ModelConfigOverride( + path="test", + value=1, + unknown_field="value", # type: ignore[call-arg] + ) + + assert "extra" in str(exc_info.value).lower() or "unknown_field" in str( + exc_info.value + ) + + def test_path_required(self) -> None: + """Test that path is required.""" + from omnibase_core.models.replay import ModelConfigOverride + + with pytest.raises(ValidationError) as exc_info: + ModelConfigOverride(value=1) # type: ignore[call-arg] + + assert "path" in str(exc_info.value) + + def test_value_required(self) -> None: + """Test that value is required.""" + from omnibase_core.models.replay import ModelConfigOverride + + with pytest.raises(ValidationError) as exc_info: + ModelConfigOverride(path="test") # type: ignore[call-arg] + + assert "value" in str(exc_info.value) + + +@pytest.mark.unit +class TestModelConfigOverrideSerialization: + """Test ModelConfigOverride serialization and deserialization.""" + + def test_serialization_roundtrip( + self, handler_config_override: ModelConfigOverride + ) -> None: + """Test that model can be serialized and deserialized.""" + from omnibase_core.models.replay import ModelConfigOverride + + # Serialize to dict + data = handler_config_override.model_dump() + + # Deserialize back + restored = ModelConfigOverride.model_validate(data) + + assert restored.path == handler_config_override.path + assert restored.value == handler_config_override.value + assert restored.injection_point == handler_config_override.injection_point + + def test_json_serialization_roundtrip( + self, handler_config_override: ModelConfigOverride + ) -> None: + """Test that model can be serialized to JSON and back.""" + from omnibase_core.models.replay import ModelConfigOverride + + # Serialize to JSON + json_str = handler_config_override.model_dump_json() + + # Deserialize back + restored = ModelConfigOverride.model_validate_json(json_str) + + assert restored.path == handler_config_override.path + assert restored.value == handler_config_override.value + + +@pytest.mark.unit +class TestModelConfigOverrideSetCreation: + """Test ModelConfigOverrideSet creation.""" + + def test_create_empty(self) -> None: + """Can create empty override set.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + override_set = ModelConfigOverrideSet() + assert len(override_set.overrides) == 0 + + def test_create_with_single_override( + self, handler_config_override: ModelConfigOverride + ) -> None: + """Can create with single override.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + override_set = ModelConfigOverrideSet(overrides=(handler_config_override,)) + assert len(override_set.overrides) == 1 + assert override_set.overrides[0] == handler_config_override + + def test_create_with_multiple_overrides( + self, + handler_config_override: ModelConfigOverride, + environment_override: ModelConfigOverride, + context_override: ModelConfigOverride, + ) -> None: + """Can create with multiple overrides.""" + overrides = ( + handler_config_override, + environment_override, + context_override, + ) + from omnibase_core.models.replay import ModelConfigOverrideSet + + override_set = ModelConfigOverrideSet(overrides=overrides) + assert len(override_set.overrides) == 3 + + +@pytest.mark.unit +class TestModelConfigOverrideSetImmutability: + """Test ModelConfigOverrideSet immutability characteristics.""" + + def test_model_is_frozen( + self, handler_config_override: ModelConfigOverride + ) -> None: + """Test that ModelConfigOverrideSet is frozen (immutable).""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + override_set = ModelConfigOverrideSet(overrides=(handler_config_override,)) + with pytest.raises(ValidationError): + override_set.overrides = () + + +@pytest.mark.unit +class TestModelConfigOverrideSetByInjectionPoint: + """Test ModelConfigOverrideSet.by_injection_point property.""" + + def test_by_injection_point_groups_correctly( + self, + handler_config_override: ModelConfigOverride, + environment_override: ModelConfigOverride, + context_override: ModelConfigOverride, + ) -> None: + """by_injection_point property groups overrides by type.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + from omnibase_core.models.replay import ( + ModelConfigOverride, + ModelConfigOverrideSet, + ) + + # Add another handler config override + second_handler = ModelConfigOverride( + path="llm.max_tokens", + value=1000, + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ) + overrides = ( + handler_config_override, + environment_override, + context_override, + second_handler, + ) + override_set = ModelConfigOverrideSet(overrides=overrides) + by_point = override_set.by_injection_point + + assert len(by_point[EnumOverrideInjectionPoint.HANDLER_CONFIG]) == 2 + assert len(by_point[EnumOverrideInjectionPoint.ENVIRONMENT]) == 1 + assert len(by_point[EnumOverrideInjectionPoint.CONTEXT]) == 1 + + def test_by_injection_point_empty_set(self) -> None: + """by_injection_point returns empty dict for empty set.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + override_set = ModelConfigOverrideSet() + by_point = override_set.by_injection_point + + assert by_point == {} + + def test_by_injection_point_single_type_only( + self, handler_config_override: ModelConfigOverride + ) -> None: + """by_injection_point works with single injection point type.""" + from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, + ) + from omnibase_core.models.replay import ( + ModelConfigOverride, + ModelConfigOverrideSet, + ) + + second_handler = ModelConfigOverride( + path="another.path", + value="value", + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ) + override_set = ModelConfigOverrideSet( + overrides=(handler_config_override, second_handler) + ) + by_point = override_set.by_injection_point + + assert len(by_point) == 1 + assert len(by_point[EnumOverrideInjectionPoint.HANDLER_CONFIG]) == 2 + + +@pytest.mark.unit +class TestModelConfigOverrideSetWithOverride: + """Test ModelConfigOverrideSet.with_override method.""" + + def test_with_override_returns_new_set( + self, handler_config_override: ModelConfigOverride + ) -> None: + """with_override returns new set (immutable update).""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + original = ModelConfigOverrideSet() + updated = original.with_override(handler_config_override) + + assert len(original.overrides) == 0 # Original unchanged + assert len(updated.overrides) == 1 + assert updated is not original + + def test_with_override_appends_to_existing( + self, + handler_config_override: ModelConfigOverride, + environment_override: ModelConfigOverride, + ) -> None: + """with_override appends to existing overrides.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + original = ModelConfigOverrideSet(overrides=(handler_config_override,)) + updated = original.with_override(environment_override) + + assert len(original.overrides) == 1 # Original unchanged + assert len(updated.overrides) == 2 + assert updated.overrides[0] == handler_config_override + assert updated.overrides[1] == environment_override + + def test_with_override_preserves_order( + self, + handler_config_override: ModelConfigOverride, + environment_override: ModelConfigOverride, + context_override: ModelConfigOverride, + ) -> None: + """with_override preserves insertion order.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + set1 = ModelConfigOverrideSet().with_override(handler_config_override) + set2 = set1.with_override(environment_override) + set3 = set2.with_override(context_override) + + assert set3.overrides[0] == handler_config_override + assert set3.overrides[1] == environment_override + assert set3.overrides[2] == context_override + + +@pytest.mark.unit +class TestModelConfigOverrideSetSerialization: + """Test ModelConfigOverrideSet serialization and deserialization.""" + + def test_serialization_roundtrip( + self, + handler_config_override: ModelConfigOverride, + environment_override: ModelConfigOverride, + ) -> None: + """Test that model can be serialized and deserialized.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + override_set = ModelConfigOverrideSet( + overrides=(handler_config_override, environment_override) + ) + + # Serialize to dict + data = override_set.model_dump() + + # Deserialize back + restored = ModelConfigOverrideSet.model_validate(data) + + assert len(restored.overrides) == 2 + assert restored.overrides[0].path == handler_config_override.path + assert restored.overrides[1].path == environment_override.path + + def test_json_serialization_roundtrip( + self, + handler_config_override: ModelConfigOverride, + environment_override: ModelConfigOverride, + ) -> None: + """Test that model can be serialized to JSON and back.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + override_set = ModelConfigOverrideSet( + overrides=(handler_config_override, environment_override) + ) + + # Serialize to JSON + json_str = override_set.model_dump_json() + + # Deserialize back + restored = ModelConfigOverrideSet.model_validate_json(json_str) + + assert len(restored.overrides) == 2 diff --git a/tests/unit/models/replay/test_model_replay_input.py b/tests/unit/models/replay/test_model_replay_input.py new file mode 100644 index 0000000000..218beada98 --- /dev/null +++ b/tests/unit/models/replay/test_model_replay_input.py @@ -0,0 +1,392 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Unit tests for ModelReplayInput. + +Tests cover: +- Model immutability (frozen) +- Default UUID generation for input_id +- Generic type preservation +- has_overrides property logic +- Immutable update pattern (with_overrides method) +- Serialization roundtrip (JSON export/import) + +.. versionadded:: 0.4.0 + Added as part of Configuration Override Injection (OMN-1205) +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any +from uuid import UUID + +import pytest +from pydantic import ValidationError + +if TYPE_CHECKING: + from omnibase_core.models.replay import ModelConfigOverride, ModelConfigOverrideSet + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + +@pytest.fixture +def sample_data() -> dict[str, Any]: + """Create sample input data.""" + return { + "event_type": "user.created", + "payload": {"user_id": 123, "email": "test@example.com"}, + "timestamp": "2024-06-15T12:30:45Z", + } + + +@pytest.fixture +def simple_override() -> ModelConfigOverride: + """Create a simple config override.""" + from omnibase_core.models.replay import ModelConfigOverride + + return ModelConfigOverride(path="test.path", value=42) + + +@pytest.fixture +def override_set(simple_override: ModelConfigOverride) -> ModelConfigOverrideSet: + """Create an override set with one override.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + + return ModelConfigOverrideSet(overrides=(simple_override,)) + + +@pytest.fixture +def replay_input_with_data( + sample_data: dict[str, Any], +) -> ModelReplayInput[dict[str, Any]]: + """Create a replay input with data only.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + return ModelReplayInput(data=sample_data) + + +@pytest.fixture +def replay_input_with_overrides( + sample_data: dict[str, Any], + override_set: ModelConfigOverrideSet, +) -> ModelReplayInput[dict[str, Any]]: + """Create a replay input with data and overrides.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + return ModelReplayInput(data=sample_data, config_overrides=override_set) + + +@pytest.mark.unit +class TestModelReplayInputCreation: + """Test ModelReplayInput creation and defaults.""" + + def test_create_with_data_only(self, sample_data: dict[str, Any]) -> None: + """Can create with just data.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + replay_input = ModelReplayInput(data=sample_data) + + assert replay_input.data == sample_data + assert replay_input.config_overrides is None + + def test_default_input_id_generated(self, sample_data: dict[str, Any]) -> None: + """Test that input_id is auto-generated as UUID.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + replay_input = ModelReplayInput(data=sample_data) + + assert replay_input.input_id is not None + assert isinstance(replay_input.input_id, UUID) + + def test_unique_input_ids_for_different_instances( + self, sample_data: dict[str, Any] + ) -> None: + """Test that each instance gets a unique input_id.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + input1 = ModelReplayInput(data=sample_data) + input2 = ModelReplayInput(data=sample_data) + + assert input1.input_id != input2.input_id + + def test_create_with_explicit_input_id(self, sample_data: dict[str, Any]) -> None: + """Can create with explicit input_id.""" + from uuid import UUID + + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + explicit_id = UUID("12345678-1234-5678-1234-567812345678") + replay_input = ModelReplayInput(data=sample_data, input_id=explicit_id) + + assert replay_input.input_id == explicit_id + + def test_create_with_overrides( + self, + sample_data: dict[str, Any], + override_set: ModelConfigOverrideSet, + ) -> None: + """Can create with config overrides.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + replay_input = ModelReplayInput( + data=sample_data, + config_overrides=override_set, + ) + + assert replay_input.config_overrides is not None + assert len(replay_input.config_overrides.overrides) == 1 + + +@pytest.mark.unit +class TestModelReplayInputImmutability: + """Test ModelReplayInput immutability characteristics.""" + + def test_model_is_frozen( + self, replay_input_with_data: ModelReplayInput[dict[str, Any]] + ) -> None: + """Test that ModelReplayInput is frozen (immutable).""" + with pytest.raises(ValidationError): + replay_input_with_data.data = {"new": "data"} + + def test_cannot_modify_input_id( + self, replay_input_with_data: ModelReplayInput[dict[str, Any]] + ) -> None: + """Test that input_id field cannot be reassigned.""" + from uuid import uuid4 + + with pytest.raises(ValidationError): + replay_input_with_data.input_id = uuid4() + + def test_cannot_modify_config_overrides( + self, + replay_input_with_overrides: ModelReplayInput[dict[str, Any]], + override_set: ModelConfigOverrideSet, + ) -> None: + """Test that config_overrides field cannot be reassigned.""" + with pytest.raises(ValidationError): + replay_input_with_overrides.config_overrides = override_set + + +@pytest.mark.unit +class TestModelReplayInputValidation: + """Test ModelReplayInput field validation.""" + + def test_data_required(self) -> None: + """Test that data is required.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + with pytest.raises(ValidationError) as exc_info: + ModelReplayInput() # type: ignore[call-arg] + + assert "data" in str(exc_info.value) + + def test_extra_fields_forbidden(self, sample_data: dict[str, Any]) -> None: + """Test that extra fields are forbidden.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + with pytest.raises(ValidationError) as exc_info: + ModelReplayInput( + data=sample_data, + unknown_field="value", # type: ignore[call-arg] + ) + + assert "extra" in str(exc_info.value).lower() or "unknown_field" in str( + exc_info.value + ) + + +@pytest.mark.unit +class TestModelReplayInputHasOverrides: + """Test ModelReplayInput.has_overrides property.""" + + def test_has_overrides_false_when_none( + self, replay_input_with_data: ModelReplayInput[dict[str, Any]] + ) -> None: + """has_overrides is False when config_overrides is None.""" + assert replay_input_with_data.has_overrides is False + + def test_has_overrides_false_when_empty(self, sample_data: dict[str, Any]) -> None: + """has_overrides is False when empty override set.""" + from omnibase_core.models.replay import ModelConfigOverrideSet + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + replay_input = ModelReplayInput( + data=sample_data, + config_overrides=ModelConfigOverrideSet(), + ) + assert replay_input.has_overrides is False + + def test_has_overrides_true_when_present( + self, + replay_input_with_overrides: ModelReplayInput[dict[str, Any]], + ) -> None: + """has_overrides is True when overrides present.""" + assert replay_input_with_overrides.has_overrides is True + + +@pytest.mark.unit +class TestModelReplayInputWithOverrides: + """Test ModelReplayInput.with_overrides method.""" + + def test_with_overrides_returns_new_instance( + self, + replay_input_with_data: ModelReplayInput[dict[str, Any]], + override_set: ModelConfigOverrideSet, + ) -> None: + """with_overrides returns new instance (immutable update).""" + updated = replay_input_with_data.with_overrides(override_set) + + assert replay_input_with_data.config_overrides is None # Original unchanged + assert updated.config_overrides is not None + assert updated is not replay_input_with_data + + def test_with_overrides_preserves_input_id( + self, + replay_input_with_data: ModelReplayInput[dict[str, Any]], + override_set: ModelConfigOverrideSet, + ) -> None: + """with_overrides preserves the input_id.""" + updated = replay_input_with_data.with_overrides(override_set) + + assert updated.input_id == replay_input_with_data.input_id + + def test_with_overrides_preserves_data( + self, + replay_input_with_data: ModelReplayInput[dict[str, Any]], + override_set: ModelConfigOverrideSet, + sample_data: dict[str, Any], + ) -> None: + """with_overrides preserves the original data.""" + updated = replay_input_with_data.with_overrides(override_set) + + assert updated.data == sample_data + + def test_with_overrides_replaces_existing( + self, + replay_input_with_overrides: ModelReplayInput[dict[str, Any]], + ) -> None: + """with_overrides replaces existing overrides.""" + from omnibase_core.models.replay import ( + ModelConfigOverride, + ModelConfigOverrideSet, + ) + + new_override = ModelConfigOverride(path="new.path", value="new_value") + new_set = ModelConfigOverrideSet(overrides=(new_override,)) + + updated = replay_input_with_overrides.with_overrides(new_set) + + assert updated.config_overrides is not None + assert len(updated.config_overrides.overrides) == 1 + assert updated.config_overrides.overrides[0].path == "new.path" + + +@pytest.mark.unit +class TestModelReplayInputGenericType: + """Test ModelReplayInput generic type behavior.""" + + def test_generic_type_with_dict(self) -> None: + """Generic type is preserved with dict.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + data_dict: dict[str, int] = {"count": 42, "total": 100} + replay_input: ModelReplayInput[dict[str, int]] = ModelReplayInput( + data=data_dict + ) + assert replay_input.data["count"] == 42 + assert replay_input.data["total"] == 100 + + def test_generic_type_with_list(self) -> None: + """Generic type works with list data.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + data_list = [1, 2, 3, 4, 5] + replay_input: ModelReplayInput[list[int]] = ModelReplayInput(data=data_list) + assert replay_input.data == [1, 2, 3, 4, 5] + + def test_generic_type_with_string(self) -> None: + """Generic type works with string data.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + data_str = "simple string data" + replay_input: ModelReplayInput[str] = ModelReplayInput(data=data_str) + assert replay_input.data == "simple string data" + + def test_generic_type_with_nested_structure(self) -> None: + """Generic type works with complex nested data.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + nested_data: dict[str, list[dict[str, int]]] = { + "items": [{"a": 1}, {"b": 2}], + } + replay_input: ModelReplayInput[dict[str, list[dict[str, int]]]] = ( + ModelReplayInput(data=nested_data) + ) + assert replay_input.data["items"][0]["a"] == 1 + + +@pytest.mark.unit +class TestModelReplayInputSerialization: + """Test ModelReplayInput serialization and deserialization.""" + + def test_serialization_roundtrip_without_overrides( + self, replay_input_with_data: ModelReplayInput[dict[str, Any]] + ) -> None: + """Test that model can be serialized and deserialized without overrides.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + # Serialize to dict + data = replay_input_with_data.model_dump() + + # Deserialize back + restored = ModelReplayInput.model_validate(data) + + assert restored.input_id == replay_input_with_data.input_id + assert restored.data == replay_input_with_data.data + assert restored.config_overrides is None + + def test_serialization_roundtrip_with_overrides( + self, replay_input_with_overrides: ModelReplayInput[dict[str, Any]] + ) -> None: + """Test that model can be serialized and deserialized with overrides.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + # Serialize to dict + data = replay_input_with_overrides.model_dump() + + # Deserialize back + restored = ModelReplayInput.model_validate(data) + + assert restored.input_id == replay_input_with_overrides.input_id + assert restored.data == replay_input_with_overrides.data + assert restored.config_overrides is not None + assert ( + len(restored.config_overrides.overrides) + == len(replay_input_with_overrides.config_overrides.overrides) # type: ignore[union-attr] + ) + + def test_json_serialization_roundtrip( + self, replay_input_with_overrides: ModelReplayInput[dict[str, Any]] + ) -> None: + """Test that model can be serialized to JSON and back.""" + from omnibase_core.models.replay.model_replay_input import ModelReplayInput + + # Serialize to JSON + json_str = replay_input_with_overrides.model_dump_json() + + # Deserialize back + restored = ModelReplayInput.model_validate_json(json_str) + + assert restored.input_id == replay_input_with_overrides.input_id + assert restored.data == replay_input_with_overrides.data + + def test_serialized_input_id_is_string( + self, replay_input_with_data: ModelReplayInput[dict[str, Any]] + ) -> None: + """Test that input_id serializes to string in JSON.""" + import json + + json_str = replay_input_with_data.model_dump_json() + data = json.loads(json_str) + + assert isinstance(data["input_id"], str) diff --git a/tests/unit/pipeline/replay/test_injector_rng.py b/tests/unit/pipeline/replay/test_injector_rng.py index f537462f7b..175ce1eede 100644 --- a/tests/unit/pipeline/replay/test_injector_rng.py +++ b/tests/unit/pipeline/replay/test_injector_rng.py @@ -335,7 +335,7 @@ def test_import_from_pipeline_replay_module(self): def test_import_from_injector_file(self): """Test direct import from injector file.""" - from omnibase_core.pipeline.replay.injector_rng import InjectorRNG + from omnibase_core.services.replay.injector_rng import InjectorRNG assert InjectorRNG is not None diff --git a/tests/unit/pipeline/replay/test_injector_time.py b/tests/unit/pipeline/replay/test_injector_time.py index 7e8fcc537e..830e020b11 100644 --- a/tests/unit/pipeline/replay/test_injector_time.py +++ b/tests/unit/pipeline/replay/test_injector_time.py @@ -26,13 +26,13 @@ import pytest if TYPE_CHECKING: - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime @pytest.fixture def production_injector() -> InjectorTime: """Create an injector in production mode (no fixed time).""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime return InjectorTime() @@ -46,7 +46,7 @@ def fixed_time() -> datetime: @pytest.fixture def replay_injector(fixed_time: datetime) -> InjectorTime: """Create an injector in replay mode with fixed time.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime return InjectorTime(fixed_time=fixed_time) @@ -125,7 +125,7 @@ def test_multiple_calls_return_same_fixed_time( def test_fixed_time_preserves_timezone(self, fixed_time: datetime) -> None: """Test that the fixed time preserves the UTC timezone.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime injector = InjectorTime(fixed_time=fixed_time) result = injector.now() @@ -156,7 +156,7 @@ def test_utc_now_returns_utc_timezone_replay( def test_fixed_time_without_timezone_gets_utc(self) -> None: """Test that naive datetime fixed_time raises or gets UTC.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime # Create naive datetime (no timezone) naive_time = datetime(2024, 6, 15, 12, 30, 45) @@ -177,10 +177,10 @@ class TestInjectorTimeProtocolCompliance: def test_implements_protocol(self) -> None: """Test that InjectorTime implements ProtocolTimeService.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime from omnibase_core.protocols.replay.protocol_time_service import ( ProtocolTimeService, ) + from omnibase_core.services.replay.injector_time import InjectorTime injector = InjectorTime() @@ -235,7 +235,7 @@ class TestInjectorTimeEdgeCases: def test_epoch_fixed_time(self) -> None: """Test with Unix epoch as fixed time.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime epoch = datetime(1970, 1, 1, 0, 0, 0, tzinfo=UTC) injector = InjectorTime(fixed_time=epoch) @@ -245,7 +245,7 @@ def test_epoch_fixed_time(self) -> None: def test_far_future_fixed_time(self) -> None: """Test with far future date as fixed time.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime future = datetime(2100, 12, 31, 23, 59, 59, tzinfo=UTC) injector = InjectorTime(fixed_time=future) @@ -255,7 +255,7 @@ def test_far_future_fixed_time(self) -> None: def test_microsecond_precision(self) -> None: """Test that microsecond precision is preserved.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime precise_time = datetime(2024, 6, 15, 12, 30, 45, microsecond=123456, tzinfo=UTC) injector = InjectorTime(fixed_time=precise_time) diff --git a/tests/unit/pipeline/replay/test_recorder_effect.py b/tests/unit/pipeline/replay/test_recorder_effect.py index d228ae7599..77a93bac40 100644 --- a/tests/unit/pipeline/replay/test_recorder_effect.py +++ b/tests/unit/pipeline/replay/test_recorder_effect.py @@ -30,10 +30,8 @@ if TYPE_CHECKING: from omnibase_core.models.replay.model_effect_record import ModelEffectRecord - from omnibase_core.pipeline.replay.recorder_effect import ( - RecorderEffect, - ) from omnibase_core.protocols.replay.protocol_time_service import ProtocolTimeService + from omnibase_core.services.replay.recorder_effect import RecorderEffect @pytest.fixture @@ -45,7 +43,7 @@ def fixed_time() -> datetime: @pytest.fixture def mock_time_service(fixed_time: datetime) -> ProtocolTimeService: """Create a mock time service that returns fixed time.""" - from omnibase_core.pipeline.replay.injector_time import InjectorTime + from omnibase_core.services.replay.injector_time import InjectorTime return InjectorTime(fixed_time=fixed_time) @@ -73,7 +71,7 @@ def sample_result() -> dict[str, Any]: def pass_through_recorder() -> RecorderEffect: """Create a recorder in pass-through mode.""" from omnibase_core.enums.replay import EnumRecorderMode - from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect + from omnibase_core.services.replay.recorder_effect import RecorderEffect return RecorderEffect(mode=EnumRecorderMode.PASS_THROUGH) @@ -84,7 +82,7 @@ def recording_recorder( ) -> RecorderEffect: """Create a recorder in recording mode.""" from omnibase_core.enums.replay import EnumRecorderMode - from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect + from omnibase_core.services.replay.recorder_effect import RecorderEffect return RecorderEffect( mode=EnumRecorderMode.RECORDING, @@ -119,7 +117,7 @@ def sample_records(fixed_time: datetime) -> list[ModelEffectRecord]: def replay_recorder(sample_records: list[ModelEffectRecord]) -> RecorderEffect: """Create a recorder in replay mode with pre-recorded data.""" from omnibase_core.enums.replay import EnumRecorderMode - from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect + from omnibase_core.services.replay.recorder_effect import RecorderEffect return RecorderEffect( mode=EnumRecorderMode.REPLAYING, @@ -386,10 +384,10 @@ class TestRecorderEffectProtocolCompliance: def test_implements_protocol(self) -> None: """Test that RecorderEffect implements ProtocolEffectRecorder.""" - from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect from omnibase_core.protocols.replay.protocol_effect_recorder import ( ProtocolEffectRecorder, ) + from omnibase_core.services.replay.recorder_effect import RecorderEffect recorder = RecorderEffect() @@ -452,11 +450,9 @@ class TestRecorderEffectIntentMatching: def test_intent_matching_for_replay(self, fixed_time: datetime) -> None: """Test that replay correctly matches intents.""" + from omnibase_core.enums.replay import EnumRecorderMode from omnibase_core.models.replay.model_effect_record import ModelEffectRecord - from omnibase_core.pipeline.replay.recorder_effect import ( - EnumRecorderMode, - RecorderEffect, - ) + from omnibase_core.services.replay.recorder_effect import RecorderEffect records = [ ModelEffectRecord( @@ -491,11 +487,9 @@ def test_intent_matching_for_replay(self, fixed_time: datetime) -> None: def test_intent_matching_is_exact(self, fixed_time: datetime) -> None: """Test that intent matching is exact (not partial).""" + from omnibase_core.enums.replay import EnumRecorderMode from omnibase_core.models.replay.model_effect_record import ModelEffectRecord - from omnibase_core.pipeline.replay.recorder_effect import ( - EnumRecorderMode, - RecorderEffect, - ) + from omnibase_core.services.replay.recorder_effect import RecorderEffect records = [ ModelEffectRecord( @@ -535,7 +529,7 @@ def test_mode_enum_values(self) -> None: def test_default_mode_is_pass_through(self) -> None: """Test that default mode is PASS_THROUGH.""" - from omnibase_core.pipeline.replay.recorder_effect import RecorderEffect + from omnibase_core.services.replay.recorder_effect import RecorderEffect recorder = RecorderEffect() assert recorder.is_recording is False diff --git a/tests/unit/services/replay/__init__.py b/tests/unit/services/replay/__init__.py new file mode 100644 index 0000000000..2d597294c2 --- /dev/null +++ b/tests/unit/services/replay/__init__.py @@ -0,0 +1,15 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Unit tests for replay services. + +This package contains unit tests for replay infrastructure services: + +- ServiceConfigOverrideInjector: Configuration override injection (OMN-1205) + +OMN-1205: Configuration Override Injection for deterministic replay. +""" + +# Mark this as a test namespace package +__all__: list[str] = [] diff --git a/tests/unit/services/replay/test_service_config_override_injector.py b/tests/unit/services/replay/test_service_config_override_injector.py new file mode 100644 index 0000000000..a9ac47d608 --- /dev/null +++ b/tests/unit/services/replay/test_service_config_override_injector.py @@ -0,0 +1,1083 @@ +# SPDX-FileCopyrightText: 2025 OmniNode Team +# +# SPDX-License-Identifier: Apache-2.0 +""" +Tests for ServiceConfigOverrideInjector. + +Tests the configuration override injection service for OMN-1205. +Validates the service's ability to validate, preview, and apply configuration +overrides while maintaining immutability and thread-safety guarantees. + +Test Categories: + - TestValidate: Validation of override paths and types + - TestPreview: Preview generation with before/after state + - TestApply: Application of overrides with immutability guarantees + - TestEnvironmentOverlay: Environment variable overlay generation + - TestPathTraversal: Nested path handling (dicts, lists, models) + - TestPydanticModelConfig: Pydantic model support + - TestConcurrentAccess: Thread-safety verification + +OMN-1205: Configuration Override Injection +""" + +import concurrent.futures +import copy +import os +import threading +from typing import Any + +import pytest +from pydantic import BaseModel, ConfigDict + +from omnibase_core.enums.replay.enum_override_injection_point import ( + EnumOverrideInjectionPoint, +) +from omnibase_core.models.replay import ( + ModelConfigOverride, + ModelConfigOverrideSet, +) +from omnibase_core.services.replay.service_config_override_injector import ( + MISSING, + ServiceConfigOverrideInjector, +) + +# ============================================================================= +# FIXTURES +# ============================================================================= + + +@pytest.fixture +def injector() -> ServiceConfigOverrideInjector: + """Create a standard (non-strict) config override injector.""" + return ServiceConfigOverrideInjector() + + +@pytest.fixture +def strict_injector() -> ServiceConfigOverrideInjector: + """Create a strict mode config override injector.""" + return ServiceConfigOverrideInjector(strict_mode=True) + + +@pytest.fixture +def sample_dict_config() -> dict[str, Any]: + """Create a sample nested dictionary configuration.""" + return { + "llm": { + "openai": {"temperature": 0.9, "model": "gpt-4"}, + "max_tokens": 1000, + }, + "retry": {"max_attempts": 5, "backoff_factor": 2.0}, + "debug": False, + } + + +@pytest.fixture +def sample_overrides() -> ModelConfigOverrideSet: + """Create a sample override set with two overrides.""" + return ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="llm.openai.temperature", value=0.5), + ModelConfigOverride(path="retry.max_attempts", value=3), + ) + ) + + +@pytest.fixture +def config_with_list() -> dict[str, Any]: + """Create a configuration with list elements.""" + return { + "items": [ + {"name": "first", "enabled": True}, + {"name": "second", "enabled": False}, + {"name": "third", "enabled": True}, + ], + "tags": ["tag1", "tag2", "tag3"], + } + + +class SamplePydanticConfig(BaseModel): + """Sample Pydantic model for testing override application. + + Note: + Uses from_attributes=True for pytest-xdist compatibility. When running + tests in parallel, xdist workers import classes independently, causing + class identity differences. from_attributes=True allows Pydantic to + accept valid instances from different worker processes. + """ + + model_config = ConfigDict(extra="forbid", from_attributes=True) + + timeout: int = 30 + debug: bool = False + name: str = "default" + + +class FrozenPydanticConfig(BaseModel): + """Frozen Pydantic model for testing immutability. + + Note: + Uses from_attributes=True for pytest-xdist compatibility. + """ + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + timeout: int = 30 + debug: bool = False + + +class NestedPydanticConfig(BaseModel): + """Nested Pydantic model for testing deep paths. + + Note: + Uses from_attributes=True for pytest-xdist compatibility. + """ + + model_config = ConfigDict(extra="forbid", from_attributes=True) + + inner: SamplePydanticConfig = SamplePydanticConfig() + level: int = 1 + + +# ============================================================================= +# TEST VALIDATE +# ============================================================================= + + +@pytest.mark.unit +class TestValidate: + """Tests for the validate() method.""" + + def test_validate_existing_paths_returns_valid( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + sample_overrides: ModelConfigOverrideSet, + ) -> None: + """Validate returns is_valid=True when all paths exist.""" + validation = injector.validate(sample_overrides, sample_dict_config) + + assert validation.is_valid is True + assert len(validation.violations) == 0 + assert validation.paths_validated == 2 + assert validation.type_checks_passed == 2 + + def test_validate_missing_path_non_strict_returns_warning( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Validate returns warning (not violation) for missing path in non-strict mode.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="nonexistent.path", value="test"),) + ) + + validation = injector.validate(overrides, sample_dict_config) + + assert validation.is_valid is True # Non-strict allows warnings + assert len(validation.violations) == 0 + assert len(validation.warnings) == 1 + assert "nonexistent.path" in validation.warnings[0] + assert "does not exist" in validation.warnings[0] + + def test_validate_missing_path_strict_returns_violation( + self, + strict_injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Validate returns is_valid=False for missing path in strict mode.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="nonexistent.path", value="test"),) + ) + + validation = strict_injector.validate(overrides, sample_dict_config) + + # Strict mode treats warnings as failures + assert validation.is_valid is False + assert len(validation.warnings) >= 1 + assert "nonexistent.path" in validation.warnings[0] + + def test_validate_type_mismatch_returns_warning( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Validate returns warning when override type doesn't match existing value.""" + overrides = ModelConfigOverrideSet( + overrides=( + # debug is bool, trying to set to string + ModelConfigOverride(path="debug", value="not_a_bool"), + ) + ) + + validation = injector.validate(overrides, sample_dict_config) + + # Type mismatch produces warning, not violation (non-strict) + assert validation.is_valid is True + assert len(validation.warnings) >= 1 + assert any("Type mismatch" in w for w in validation.warnings) + + def test_validate_invalid_path_syntax( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Validate returns violation for invalid path syntax.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="..invalid", value="test"),) + ) + + validation = injector.validate(overrides, sample_dict_config) + + assert validation.is_valid is False + assert len(validation.violations) >= 1 + assert any("Invalid path syntax" in v for v in validation.violations) + + def test_validate_numeric_conversion_allowed( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Validate allows int/float conversion without warning.""" + overrides = ModelConfigOverrideSet( + overrides=( + # max_tokens is int (1000), setting to float + ModelConfigOverride(path="llm.max_tokens", value=1500.0), + ) + ) + + validation = injector.validate(overrides, sample_dict_config) + + assert validation.is_valid is True + # Numeric conversion is acceptable, should pass type check + assert validation.type_checks_passed == 1 + + +# ============================================================================= +# TEST PREVIEW +# ============================================================================= + + +@pytest.mark.unit +class TestPreview: + """Tests for the preview() method.""" + + def test_preview_shows_before_after( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + sample_overrides: ModelConfigOverrideSet, + ) -> None: + """Preview shows old and new values for each override.""" + preview = injector.preview(sample_overrides, sample_dict_config) + + assert len(preview.field_previews) == 2 + + # Check temperature override + temp_preview = next( + p for p in preview.field_previews if p.path == "llm.openai.temperature" + ) + assert temp_preview.old_value == 0.9 + assert temp_preview.new_value == 0.5 + assert temp_preview.path_exists is True + + # Check retry override + retry_preview = next( + p for p in preview.field_previews if p.path == "retry.max_attempts" + ) + assert retry_preview.old_value == 5 + assert retry_preview.new_value == 3 + assert retry_preview.path_exists is True + + def test_preview_identifies_new_paths( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Preview identifies paths that don't exist (will be created).""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="new_setting", value="new_value"), + ModelConfigOverride(path="another.new.path", value=123), + ) + ) + + preview = injector.preview(overrides, sample_dict_config) + + assert len(preview.paths_not_found) == 2 + assert "new_setting" in preview.paths_not_found + assert "another.new.path" in preview.paths_not_found + + # Field previews should show path_exists=False + for fp in preview.field_previews: + assert fp.path_exists is False + + def test_preview_to_markdown_generates_table( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + sample_overrides: ModelConfigOverrideSet, + ) -> None: + """Preview to_markdown() generates valid markdown table.""" + preview = injector.preview(sample_overrides, sample_dict_config) + markdown = preview.to_markdown() + + # Check header + assert "## Configuration Override Preview" in markdown + assert "| Path | Injection Point | Before | After | Status |" in markdown + assert "|------|-----------------|--------|-------|--------|" in markdown + + # Check content + assert "llm.openai.temperature" in markdown + assert "retry.max_attempts" in markdown + + def test_preview_type_mismatches_tracked( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Preview tracks type mismatches for review.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="debug", value="string_not_bool"),) + ) + + preview = injector.preview(overrides, sample_dict_config) + + assert len(preview.type_mismatches) == 1 + assert "debug" in preview.type_mismatches[0] + + def test_preview_filters_by_injection_point( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Preview filters overrides by injection point when specified.""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride( + path="llm.openai.temperature", + value=0.5, + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ), + ModelConfigOverride( + path="env_var", + value="test", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ModelConfigOverride( + path="context_var", + value="ctx", + injection_point=EnumOverrideInjectionPoint.CONTEXT, + ), + ) + ) + + # Only preview HANDLER_CONFIG overrides + preview = injector.preview( + overrides, + sample_dict_config, + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ) + + assert len(preview.field_previews) == 1 + assert preview.field_previews[0].path == "llm.openai.temperature" + assert ( + preview.field_previews[0].injection_point + == EnumOverrideInjectionPoint.HANDLER_CONFIG + ) + + # Preview all (no filter) + preview_all = injector.preview(overrides, sample_dict_config) + assert len(preview_all.field_previews) == 3 + + +# ============================================================================= +# TEST APPLY +# ============================================================================= + + +@pytest.mark.unit +class TestApply: + """Tests for the apply() method - critical immutability tests.""" + + def test_apply_does_not_modify_original( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + sample_overrides: ModelConfigOverrideSet, + ) -> None: + """Apply NEVER modifies the original configuration.""" + original_copy = copy.deepcopy(sample_dict_config) + + result = injector.apply(sample_overrides, sample_dict_config) + + # Original must remain unchanged + assert sample_dict_config == original_copy + assert sample_dict_config["llm"]["openai"]["temperature"] == 0.9 + assert sample_dict_config["retry"]["max_attempts"] == 5 + + # Patched config should have new values + assert result.patched_config["llm"]["openai"]["temperature"] == 0.5 + assert result.patched_config["retry"]["max_attempts"] == 3 + + def test_apply_returns_new_object( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + sample_overrides: ModelConfigOverrideSet, + ) -> None: + """Apply returns a new object, not the original.""" + result = injector.apply(sample_overrides, sample_dict_config) + + assert result.patched_config is not sample_dict_config + assert result.patched_config["llm"] is not sample_dict_config["llm"] + assert ( + result.patched_config["llm"]["openai"] + is not sample_dict_config["llm"]["openai"] + ) + + def test_apply_preserves_unspecified_fields( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Apply preserves fields not specified in overrides.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="llm.openai.temperature", value=0.1),) + ) + + result = injector.apply(overrides, sample_dict_config) + + # Modified field + assert result.patched_config["llm"]["openai"]["temperature"] == 0.1 + + # Preserved fields + assert result.patched_config["llm"]["openai"]["model"] == "gpt-4" + assert result.patched_config["llm"]["max_tokens"] == 1000 + assert result.patched_config["retry"]["max_attempts"] == 5 + assert result.patched_config["retry"]["backoff_factor"] == 2.0 + assert result.patched_config["debug"] is False + + def test_apply_tracks_created_paths( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Apply tracks paths that were created (didn't exist before).""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="new_setting", value="created"), + ModelConfigOverride(path="llm.openai.temperature", value=0.5), + ) + ) + + result = injector.apply(overrides, sample_dict_config) + + assert result.success is True + assert "new_setting" in result.paths_created + # temperature existed, should not be in paths_created + assert "llm.openai.temperature" not in result.paths_created + assert result.overrides_applied == 2 + + def test_apply_reports_errors_for_invalid_paths( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Apply reports errors for invalid path syntax.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="..bad.path", value="test"),) + ) + + result = injector.apply(overrides, sample_dict_config) + + assert result.success is False + assert len(result.errors) >= 1 + assert any("Invalid path" in e for e in result.errors) + + +# ============================================================================= +# TEST ENVIRONMENT OVERLAY +# ============================================================================= + + +@pytest.mark.unit +class TestEnvironmentOverlay: + """Tests for the apply_environment_overlay() method.""" + + def test_creates_env_dict_not_modifying_os_environ( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Environment overlay creates dict without touching os.environ. + + Environment variable names are converted to SCREAMING_SNAKE_CASE + (uppercase) by convention. The last segment of the path is used + as the variable name. + """ + original_environ = dict(os.environ) + + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride( + path="test_var", + value="test_value", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ModelConfigOverride( + path="api.key", + value="secret123", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ) + ) + + overlay = injector.apply_environment_overlay(overrides) + + # os.environ must remain unchanged + assert dict(os.environ) == original_environ + + # Overlay should contain environment variables in SCREAMING_SNAKE_CASE + # The service converts names to uppercase (last segment of path) + assert "TEST_VAR" in overlay + assert overlay["TEST_VAR"] == "test_value" + assert "KEY" in overlay # Last segment of "api.key", uppercased + assert overlay["KEY"] == "secret123" + + def test_env_overlay_only_includes_environment_injection_point( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Environment overlay only includes ENVIRONMENT injection point overrides.""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride( + path="env_var", + value="included", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ModelConfigOverride( + path="handler_var", + value="excluded", + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ), + ) + ) + + overlay = injector.apply_environment_overlay(overrides) + + assert len(overlay) == 1 + assert "ENV_VAR" in overlay + + def test_env_overlay_converts_values_to_strings( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Environment overlay converts all values to strings.""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride( + path="int_var", + value=42, + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ModelConfigOverride( + path="bool_var", + value=True, + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ModelConfigOverride( + path="none_var", + value=None, + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ) + ) + + overlay = injector.apply_environment_overlay(overrides) + + assert overlay["INT_VAR"] == "42" + assert overlay["BOOL_VAR"] == "true" + assert overlay["NONE_VAR"] == "" + + def test_env_overlay_is_immutable( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Environment overlay returns immutable MappingProxyType. + + The returned overlay cannot be mutated by callers, preventing + accidental modification of the overlay that could affect other + parts of the system. + """ + import types + + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride( + path="test_var", + value="test_value", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ) + ) + + overlay = injector.apply_environment_overlay(overrides) + + # Verify return type is MappingProxyType + assert isinstance(overlay, types.MappingProxyType) + + # Verify it's readable + assert overlay["TEST_VAR"] == "test_value" + + # Verify mutation raises TypeError + with pytest.raises(TypeError): + overlay["NEW_VAR"] = "should_fail" # type: ignore[index] + + with pytest.raises(TypeError): + del overlay["TEST_VAR"] # type: ignore[misc] + + # Verify dict() can create mutable copy for subprocess use + mutable_copy = dict(overlay) + mutable_copy["NEW_VAR"] = "can_add" + assert "NEW_VAR" in mutable_copy + assert "NEW_VAR" not in overlay + + +# ============================================================================= +# TEST PATH TRAVERSAL +# ============================================================================= + + +@pytest.mark.unit +class TestPathTraversal: + """Tests for nested path handling.""" + + def test_deep_nested_path( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Handles deeply nested paths like a.b.c.d.""" + config: dict[str, Any] = {"a": {"b": {"c": {"d": {"value": "original"}}}}} + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="a.b.c.d.value", value="modified"),) + ) + + result = injector.apply(overrides, config) + + assert result.success is True + assert result.patched_config["a"]["b"]["c"]["d"]["value"] == "modified" + # Original unchanged + assert config["a"]["b"]["c"]["d"]["value"] == "original" + + def test_list_index_path( + self, + injector: ServiceConfigOverrideInjector, + config_with_list: dict[str, Any], + ) -> None: + """Handles list index paths like items.0.name.""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="items.0.name", value="modified_first"), + ModelConfigOverride(path="items.1.enabled", value=True), + ModelConfigOverride(path="tags.2", value="modified_tag"), + ) + ) + + result = injector.apply(overrides, config_with_list) + + assert result.success is True + assert result.patched_config["items"][0]["name"] == "modified_first" + assert result.patched_config["items"][1]["enabled"] is True + assert result.patched_config["tags"][2] == "modified_tag" + + # Original unchanged + assert config_with_list["items"][0]["name"] == "first" + assert config_with_list["items"][1]["enabled"] is False + assert config_with_list["tags"][2] == "tag3" + + def test_list_index_out_of_bounds( + self, + injector: ServiceConfigOverrideInjector, + config_with_list: dict[str, Any], + ) -> None: + """Handles out-of-bounds list index gracefully.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="items.99.name", value="test"),) + ) + + result = injector.apply(overrides, config_with_list) + + assert result.success is False + assert len(result.errors) >= 1 + assert any("out of bounds" in e.lower() for e in result.errors) + + def test_create_intermediate_dicts( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Creates intermediate dictionaries when path doesn't exist.""" + config: dict[str, Any] = {"existing": "value"} + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="new.nested.path", value="created"),) + ) + + result = injector.apply(overrides, config) + + assert result.success is True + assert result.patched_config["new"]["nested"]["path"] == "created" + assert "new.nested.path" in result.paths_created + + +# ============================================================================= +# TEST PYDANTIC MODEL CONFIG +# ============================================================================= + + +@pytest.mark.unit +class TestPydanticModelConfig: + """Tests for Pydantic model support.""" + + def test_apply_to_pydantic_model( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Apply works with mutable Pydantic models.""" + config = SamplePydanticConfig(timeout=30, debug=False, name="original") + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="timeout", value=60), + ModelConfigOverride(path="name", value="modified"), + ) + ) + + result = injector.apply(overrides, config) + + assert result.success is True + assert result.patched_config.timeout == 60 + assert result.patched_config.name == "modified" + assert result.patched_config.debug is False # Preserved + + # Original unchanged (Pydantic model_copy creates new instance) + assert config.timeout == 30 + assert config.name == "original" + + def test_apply_to_frozen_pydantic_model_converts_to_dict( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Apply converts frozen Pydantic models to dict for modification. + + Frozen Pydantic models cannot be modified in place. The service + detects frozen models early and converts them to dicts, allowing + the overrides to be applied successfully to the dict representation. + """ + config = FrozenPydanticConfig(timeout=30, debug=False) + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="timeout", value=60),) + ) + + result = injector.apply(overrides, config) + + # Frozen models are converted to dicts, allowing modification + assert result.success is True + assert result.overrides_applied == 1 + # Result is a dict, not the original frozen model + assert isinstance(result.patched_config, dict) + assert result.patched_config["timeout"] == 60 + assert result.patched_config["debug"] is False + # Original frozen model is unchanged + assert config.timeout == 30 + + def test_validate_pydantic_model_paths( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Validate works with Pydantic model paths.""" + config = SamplePydanticConfig(timeout=30) + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="timeout", value=60), + ModelConfigOverride(path="nonexistent", value="test"), + ) + ) + + validation = injector.validate(overrides, config) + + # First override is valid + # Second doesn't exist - warning + assert len(validation.warnings) >= 1 + assert any("nonexistent" in w for w in validation.warnings) + + def test_apply_to_nested_pydantic_model( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Apply handles nested Pydantic models.""" + config = NestedPydanticConfig() + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="inner.timeout", value=120), + ModelConfigOverride(path="level", value=5), + ) + ) + + result = injector.apply(overrides, config) + + assert result.success is True + assert result.patched_config.inner.timeout == 120 + assert result.patched_config.level == 5 + + def test_is_frozen_pydantic_model_detection( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """The _is_frozen_pydantic_model helper correctly detects frozen models.""" + # Test frozen model detection + frozen_config = FrozenPydanticConfig(timeout=30, debug=False) + assert injector._is_frozen_pydantic_model(frozen_config) is True + + # Test mutable model detection + mutable_config = SamplePydanticConfig(timeout=30) + assert injector._is_frozen_pydantic_model(mutable_config) is False + + # Test non-Pydantic objects + assert injector._is_frozen_pydantic_model({"timeout": 30}) is False + assert injector._is_frozen_pydantic_model([1, 2, 3]) is False + assert injector._is_frozen_pydantic_model("string") is False + assert injector._is_frozen_pydantic_model(None) is False + assert injector._is_frozen_pydantic_model(42) is False + + +# ============================================================================= +# TEST CONCURRENT ACCESS +# ============================================================================= + + +@pytest.mark.unit +class TestConcurrentAccess: + """Tests for thread-safety.""" + + def test_concurrent_apply_no_race_condition( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Multiple concurrent applies don't cause race conditions.""" + config: dict[str, Any] = {"shared": {"counter": 0}, "value": "original"} + + results: list[Any] = [] + errors: list[Exception] = [] + lock = threading.Lock() + + def apply_override(index: int) -> None: + """Apply an override and store result.""" + try: + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="value", value=f"modified_{index}"), + ModelConfigOverride(path=f"new_field_{index}", value=index), + ) + ) + result = injector.apply(overrides, config) + with lock: + results.append((index, result)) + except Exception as e: + with lock: + errors.append(e) + + # Run concurrent applies + with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: + futures = [executor.submit(apply_override, i) for i in range(20)] + concurrent.futures.wait(futures) + + # No errors should occur + assert len(errors) == 0, f"Errors occurred: {errors}" + + # All applies should succeed + assert len(results) == 20 + + # Original config must be unchanged + assert config["value"] == "original" + assert config["shared"]["counter"] == 0 + assert "new_field_0" not in config + + # Each result should have independent patched config + for index, result in results: + assert result.success is True + assert result.patched_config["value"] == f"modified_{index}" + assert result.patched_config[f"new_field_{index}"] == index + + def test_concurrent_validate_no_race_condition( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Multiple concurrent validates are thread-safe.""" + config: dict[str, Any] = {"field1": "value1", "field2": 100} + + results: list[Any] = [] + lock = threading.Lock() + + def validate_override(index: int) -> None: + """Validate an override and store result.""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride(path="field1", value=f"new_{index}"), + ModelConfigOverride(path="field2", value=index), + ) + ) + validation = injector.validate(overrides, config) + with lock: + results.append((index, validation)) + + with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: + futures = [executor.submit(validate_override, i) for i in range(20)] + concurrent.futures.wait(futures) + + # All validations should succeed + assert len(results) == 20 + for _, validation in results: + assert validation.is_valid is True + assert validation.paths_validated == 2 + + def test_injector_instance_reuse_thread_safe( + self, + ) -> None: + """Single injector instance can be safely reused across threads.""" + injector = ServiceConfigOverrideInjector() + configs = [{"value": i} for i in range(10)] + + results: list[Any] = [] + lock = threading.Lock() + + def process_config(config: dict[str, Any], index: int) -> None: + """Process a config with the shared injector.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="value", value=index * 10),) + ) + result = injector.apply(overrides, config) + with lock: + results.append((config["value"], result.patched_config["value"])) + + with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: + futures = [ + executor.submit(process_config, configs[i], i) for i in range(10) + ] + concurrent.futures.wait(futures) + + assert len(results) == 10 + + # Verify each original config retained its value + for i in range(10): + assert configs[i]["value"] == i + + +# ============================================================================= +# TEST EDGE CASES +# ============================================================================= + + +@pytest.mark.unit +class TestEdgeCases: + """Tests for edge cases and boundary conditions.""" + + def test_empty_override_set( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Handles empty override set gracefully.""" + overrides = ModelConfigOverrideSet(overrides=()) + + validation = injector.validate(overrides, sample_dict_config) + assert validation.is_valid is True + assert validation.paths_validated == 0 + + preview = injector.preview(overrides, sample_dict_config) + assert len(preview.field_previews) == 0 + + result = injector.apply(overrides, sample_dict_config) + assert result.success is True + assert result.overrides_applied == 0 + assert result.patched_config == sample_dict_config + + def test_none_value_override( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Handles setting values to None.""" + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path="llm.openai.temperature", value=None),) + ) + + result = injector.apply(overrides, sample_dict_config) + + assert result.success is True + assert result.patched_config["llm"]["openai"]["temperature"] is None + + def test_missing_sentinel_behavior( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """MISSING sentinel is distinct from None.""" + assert MISSING is not None + assert bool(MISSING) is False + assert repr(MISSING) == "" + + # Same instance always + from omnibase_core.services.replay.sentinel_missing import _Missing + + assert _Missing() is MISSING + + def test_injection_point_filtering( + self, + injector: ServiceConfigOverrideInjector, + sample_dict_config: dict[str, Any], + ) -> None: + """Filtering by injection point works correctly.""" + overrides = ModelConfigOverrideSet( + overrides=( + ModelConfigOverride( + path="llm.openai.temperature", + value=0.5, + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ), + ModelConfigOverride( + path="env_var", + value="test", + injection_point=EnumOverrideInjectionPoint.ENVIRONMENT, + ), + ) + ) + + # Only validate HANDLER_CONFIG + validation = injector.validate( + overrides, + sample_dict_config, + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ) + assert validation.paths_validated == 1 + + # Only apply HANDLER_CONFIG + result = injector.apply( + overrides, + sample_dict_config, + injection_point=EnumOverrideInjectionPoint.HANDLER_CONFIG, + ) + assert result.overrides_applied == 1 + assert result.patched_config["llm"]["openai"]["temperature"] == 0.5 + + def test_path_depth_limit( + self, + injector: ServiceConfigOverrideInjector, + ) -> None: + """Path depth is limited to prevent DoS.""" + # Create a path exceeding MAX_PATH_DEPTH (20) + deep_path = ".".join([f"level{i}" for i in range(25)]) + overrides = ModelConfigOverrideSet( + overrides=(ModelConfigOverride(path=deep_path, value="test"),) + ) + + validation = injector.validate(overrides, {}) + + assert validation.is_valid is False + assert any("depth" in v.lower() for v in validation.violations)