feat(models): CDQA-08 — add ModelGoldenPath and golden_path to ModelTicketContract [OMN-2980]

pyproject.toml

@@ -57,6 +57,11 @@ testpaths = ["tests"]
 python_files = ["test_*.py", "*_test.py"]
 python_classes = ["Test*"]
 python_functions = ["test_*"]
+markers = [
+    "unit: Unit tests (fast, isolated)",
+    "integration: Integration tests",
+    "slow: Slow-running tests",
+]
 addopts = [
     "--strict-markers",
     "--strict-config",

src/onex_change_control/__init__.py

@@ -6,16 +6,29 @@
 
 Downstream packages can import and use the models directly:
     from onex_change_control import ModelDayClose, ModelTicketContract
+    from onex_change_control import ModelGoldenPath, ModelGoldenPathAssertion
+    from onex_change_control import ModelGoldenPathInput, ModelGoldenPathOutput
 """
 
 from typing import Final
 
-from onex_change_control.models import ModelDayClose, ModelTicketContract
+from onex_change_control.models import (
+    ModelDayClose,
+    ModelGoldenPath,
+    ModelGoldenPathAssertion,
+    ModelGoldenPathInput,
+    ModelGoldenPathOutput,
+    ModelTicketContract,
+)
 
 __version__: Final[str] = "0.1.0"
 
 __all__ = [
     "ModelDayClose",
+    "ModelGoldenPath",
+    "ModelGoldenPathAssertion",
+    "ModelGoldenPathInput",
+    "ModelGoldenPathOutput",
     "ModelTicketContract",
     "__version__",
 ]

src/onex_change_control/models/__init__.py

@@ -1,12 +1,23 @@
 """ONEX Change Control Models.
 
-This module provides Pydantic schema models for drift control artifacts.
+Exports ModelGoldenPath, ModelTicketContract, ModelDayClose, and related
+supporting types used for drift detection and governance workflows.
 """
 
 from onex_change_control.models.model_day_close import ModelDayClose
+from onex_change_control.models.model_golden_path import (
+    ModelGoldenPath,
+    ModelGoldenPathAssertion,
+    ModelGoldenPathInput,
+    ModelGoldenPathOutput,
+)
 from onex_change_control.models.model_ticket_contract import ModelTicketContract
 
 __all__ = [
     "ModelDayClose",
+    "ModelGoldenPath",
+    "ModelGoldenPathAssertion",
+    "ModelGoldenPathInput",
+    "ModelGoldenPathOutput",
     "ModelTicketContract",
 ]

src/onex_change_control/models/model_golden_path.py

@@ -0,0 +1,138 @@
+"""Golden Path Model.
+
+Pydantic schema model for golden path event chain test declarations.
+A golden path test verifies that a specific input event produces the expected
+output event, enabling automated contract verification of node pipelines.
+"""
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+# Security constraints to prevent DoS attacks
+_MAX_STRING_LENGTH = 10000  # Max length for string fields
+_MAX_LIST_ITEMS = 1000  # Max items in lists
+
+
+class ModelGoldenPathAssertion(BaseModel, frozen=True):
+    """A single assertion on an output event field.
+
+    Specifies a field path, comparison operator, and expected value.
+    Assertions are evaluated against the output event produced by running
+    the golden path test.
+    """
+
+    field: str = Field(
+        ...,
+        description=(
+            "Dot-separated field path on the output event "
+            "(e.g., 'status' or 'data.result')"
+        ),
+        max_length=_MAX_STRING_LENGTH,
+    )
+    op: Literal["eq", "neq", "gte", "lte", "in", "contains"] = Field(
+        ...,
+        description="Comparison operator: eq | neq | gte | lte | in | contains",
+    )
+    value: Any = Field(
+        ...,
+        description="Expected value to compare against",
+    )
+
+
+class ModelGoldenPathInput(BaseModel, frozen=True):
+    """Input specification for a golden path test.
+
+    Describes the Kafka topic and fixture file to use as the input event.
+    """
+
+    topic: str = Field(
+        ...,
+        description="Kafka topic to publish the input event to",
+        max_length=_MAX_STRING_LENGTH,
+    )
+    fixture: str = Field(
+        ...,
+        description="Path to the JSON fixture file relative to the repo root",
+        max_length=_MAX_STRING_LENGTH,
+    )
+    input_correlation_id_field: str = Field(
+        default="correlation_id",
+        description="Field name in the fixture that holds the correlation ID",
+        max_length=_MAX_STRING_LENGTH,
+    )
+
+
+class ModelGoldenPathOutput(BaseModel, frozen=True):
+    """Output specification for a golden path test.
+
+    Describes the Kafka topic to listen on for the output event, with optional
+    schema validation and field-level assertions.
+    """
+
+    topic: str = Field(
+        ...,
+        description="Kafka topic to consume the output event from",
+        max_length=_MAX_STRING_LENGTH,
+    )
+    output_correlation_id_field: str = Field(
+        default="correlation_id",
+        description="Field name in the output event that holds the correlation ID",
+        max_length=_MAX_STRING_LENGTH,
+    )
+    schema_name: str | None = Field(
+        default=None,
+        description=(
+            "Optional Pydantic model class name for output validation. "
+            "When present and importable, the runner validates the output event "
+            "against this schema. When not importable, "
+            "schema_validation_status is set to 'skipped'."
+        ),
+        max_length=_MAX_STRING_LENGTH,
+    )
+    assertions: list[ModelGoldenPathAssertion] = Field(
+        default_factory=list,
+        description="Field-level assertions to evaluate against the output event",
+        max_length=_MAX_LIST_ITEMS,
+    )
+
+
+class ModelGoldenPath(BaseModel, frozen=True):
+    """Golden path event chain test declaration.
+
+    Declares a full input-to-output contract test for a node pipeline. The golden
+    path runner publishes the input fixture to the input topic, waits for a matching
+    output event on the output topic, and evaluates all assertions.
+
+    The timeout_ms field lives here (not in input or output) as the single source
+    of truth for the test timeout. The infra field controls whether real Kafka or
+    a mock is used.
+    """
+
+    input: ModelGoldenPathInput = Field(
+        ...,
+        description="Input event specification",
+    )
+    output: ModelGoldenPathOutput = Field(
+        ...,
+        description="Output event specification",
+    )
+    timeout_ms: int = Field(
+        default=30000,
+        description="Timeout in milliseconds for the full input-to-output round trip",
+        ge=1,
+    )
+    infra: Literal["real", "mock"] = Field(
+        default="real",
+        description=(
+            "Infrastructure mode: 'real' uses live Kafka, "
+            "'mock' uses an in-process stub"
+        ),
+    )
+    test_file: str | None = Field(
+        default=None,
+        description=(
+            "Optional path to the pytest golden path file relative to the repo root"
+        ),
+        max_length=_MAX_STRING_LENGTH,
+    )

src/onex_change_control/models/model_ticket_contract.py

@@ -7,6 +7,7 @@
 
 from onex_change_control.enums.enum_evidence_kind import EnumEvidenceKind
 from onex_change_control.enums.enum_interface_surface import EnumInterfaceSurface
+from onex_change_control.models.model_golden_path import ModelGoldenPath
 from onex_change_control.validation.patterns import SEMVER_PATTERN
 
 # Security constraints to prevent DoS attacks
@@ -130,6 +131,14 @@ class ModelTicketContract(BaseModel):
         ...,
         description="Emergency bypass configuration",
     )
+    golden_path: ModelGoldenPath | None = Field(
+        default=None,
+        description=(
+            "Optional golden path event chain test declaration. "
+            "When present, declares an input-to-output contract test for the "
+            "node pipeline associated with this ticket."
+        ),
+    )
 
     @field_validator("schema_version")
     @classmethod

templates/ticket_contract.template.yaml

@@ -49,3 +49,28 @@ emergency_bypass:
   enabled: false  # Set to true only in genuine emergencies
   justification: ""  # Required if enabled=true: explain why bypass is necessary
   follow_up_ticket_id: ""  # Required if enabled=true: ticket that will properly address this
+
+# Optional: Declare a golden path event chain test (omit if not a node-based ticket).
+# When present, declares an input-to-output contract test for the node pipeline.
+# The runner publishes the input fixture to the input topic, waits for a matching
+# output event on the output topic, and evaluates all assertions.
+#
+# golden_path:
+#   timeout_ms: 30000   # Single source of truth for round-trip timeout (default: 30000ms)
+#   infra: real         # "real" uses live Kafka, "mock" uses an in-process stub
+#   input:
+#     topic: onex.cmd.{service}.{event}.v1
+#     fixture: tests/fixtures/golden/{name}.json
+#     input_correlation_id_field: correlation_id  # Optional, default: correlation_id
+#   output:
+#     topic: onex.evt.{service}.{event}.v1
+#     output_correlation_id_field: correlation_id  # Optional, default: correlation_id
+#     schema_name: ModelFoo        # Optional; validates output if importable, skips if not
+#     assertions:
+#       - field: status
+#         op: eq
+#         value: completed
+#       - field: data.count
+#         op: gte
+#         value: 1
+#   test_file: tests/golden/test_{ticket_id_snake}_golden.py  # Optional

tests/unit/__init__.py

@@ -0,0 +1 @@
+"""Unit tests for onex_change_control."""