diff --git a/src/onex_change_control/enums/__init__.py b/src/onex_change_control/enums/__init__.py index b4b0ced0586..cb97c3de0b4 100644 --- a/src/onex_change_control/enums/__init__.py +++ b/src/onex_change_control/enums/__init__.py @@ -7,6 +7,8 @@ EnumAutopilotCycleStatus, EnumAutopilotStepStatus, ) +from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict +from onex_change_control.enums.enum_compliance_violation import EnumComplianceViolation from onex_change_control.enums.enum_db_boundary import ( EnumDbBoundaryExceptionStatus, EnumDbBoundaryReasonCategory, @@ -27,6 +29,7 @@ from onex_change_control.enums.enum_integration_surface import EnumIntegrationSurface from onex_change_control.enums.enum_interface_surface import EnumInterfaceSurface from onex_change_control.enums.enum_invariant_status import EnumInvariantStatus +from onex_change_control.enums.enum_migration_status import EnumMigrationStatus from onex_change_control.enums.enum_pr_state import EnumPRState from onex_change_control.enums.enum_probe_reason import EnumProbeReason from onex_change_control.enums.enum_probe_status import EnumProbeStatus @@ -34,6 +37,8 @@ __all__ = [ "EnumAutopilotCycleStatus", "EnumAutopilotStepStatus", + "EnumComplianceVerdict", + "EnumComplianceViolation", "EnumDbBoundaryExceptionStatus", "EnumDbBoundaryReasonCategory", "EnumDocReferenceType", @@ -50,6 +55,7 @@ "EnumIntegrationSurface", "EnumInterfaceSurface", "EnumInvariantStatus", + "EnumMigrationStatus", "EnumPRState", "EnumProbeReason", "EnumProbeStatus", diff --git a/src/onex_change_control/enums/enum_compliance_verdict.py b/src/onex_change_control/enums/enum_compliance_verdict.py new file mode 100644 index 00000000000..b7bee5451a5 --- /dev/null +++ b/src/onex_change_control/enums/enum_compliance_verdict.py @@ -0,0 +1,40 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Compliance Verdict Enum. + +Verdict categories for handler contract compliance audits. +""" + +from enum import Enum, unique + + +@unique +class EnumComplianceVerdict(str, Enum): + """Verdict for a handler's contract compliance status. + + - COMPLIANT: Handler fully adheres to its contract declarations. + - IMPERATIVE: Handler bypasses the contract system (hardcoded topics, etc.). + - HYBRID: Handler partially uses contracts but has some imperative wiring. + - ALLOWLISTED: Handler has known violations tracked in the allowlist. + - MISSING_CONTRACT: No contract.yaml exists for the handler's node. + """ + + COMPLIANT = "compliant" + """Handler fully adheres to its contract declarations.""" + + IMPERATIVE = "imperative" + """Handler bypasses the contract system.""" + + HYBRID = "hybrid" + """Handler partially uses contracts but has some imperative wiring.""" + + ALLOWLISTED = "allowlisted" + """Handler has known violations tracked in the allowlist.""" + + MISSING_CONTRACT = "missing_contract" + """No contract.yaml exists for the handler's node.""" + + def __str__(self) -> str: + """Return the string value for serialization.""" + return self.value diff --git a/src/onex_change_control/enums/enum_compliance_violation.py b/src/onex_change_control/enums/enum_compliance_violation.py new file mode 100644 index 00000000000..6d01f05589a --- /dev/null +++ b/src/onex_change_control/enums/enum_compliance_violation.py @@ -0,0 +1,46 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Compliance Violation Enum. + +Specific violation types for handler contract compliance checks. +""" + +from enum import Enum, unique + + +@unique +class EnumComplianceViolation(str, Enum): + """Specific contract compliance violation types. + + Each value represents a distinct way a handler can bypass + the contract-declared dispatch system. + """ + + HARDCODED_TOPIC = "hardcoded_topic" + """Topic string literal in handler instead of contract declaration.""" + + UNDECLARED_TRANSPORT = "undeclared_transport" + """Handler uses a transport (DB, HTTP, Kafka) not declared in contract.""" + + LOGIC_IN_NODE = "logic_in_node" + """Business logic found in node.py instead of handler.""" + + MISSING_HANDLER_ROUTING = "missing_handler_routing" + """Handler exists in handlers/ but is not in contract.yaml handler_routing.""" + + UNDECLARED_PUBLISH = "undeclared_publish" + """Handler publishes to topic not in contract event_bus.publish_topics.""" + + UNDECLARED_SUBSCRIBE = "undeclared_subscribe" + """Handler subscribes to topic not in contract event_bus.subscribe_topics.""" + + DIRECT_DB_ACCESS = "direct_db_access" + """Handler constructs DB connections directly instead of using injected services.""" + + UNREGISTERED_HANDLER = "unregistered_handler" + """Handler file in handlers/ directory but not importable or registered.""" + + def __str__(self) -> str: + """Return the string value for serialization.""" + return self.value diff --git a/src/onex_change_control/enums/enum_migration_status.py b/src/onex_change_control/enums/enum_migration_status.py new file mode 100644 index 00000000000..9df42ab5f8f --- /dev/null +++ b/src/onex_change_control/enums/enum_migration_status.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Migration Status Enum. + +Status tracking for imperative-to-declarative handler migrations. +""" + +from enum import Enum, unique + + +@unique +class EnumMigrationStatus(str, Enum): + """Status of a handler's migration from imperative to declarative. + + Lifecycle: PENDING -> GENERATED -> VALIDATED -> DEPLOYED -> RETIRED + + - PENDING: Migration identified but not yet started. + - GENERATED: Migration spec generated with contract.yaml changes. + - VALIDATED: Migration validated via contract-dispatch equivalence. + - DEPLOYED: Migrated handler deployed and serving traffic. + - RETIRED: Old imperative wiring fully removed. + """ + + PENDING = "pending" + """Migration identified but not yet started.""" + + GENERATED = "generated" + """Migration spec generated with contract.yaml changes.""" + + VALIDATED = "validated" + """Migration validated via contract-dispatch equivalence.""" + + DEPLOYED = "deployed" + """Migrated handler deployed and serving traffic.""" + + RETIRED = "retired" + """Old imperative wiring fully removed.""" + + def __str__(self) -> str: + """Return the string value for serialization.""" + return self.value diff --git a/src/onex_change_control/models/__init__.py b/src/onex_change_control/models/__init__.py index e809c884840..ec23756330f 100644 --- a/src/onex_change_control/models/__init__.py +++ b/src/onex_change_control/models/__init__.py @@ -16,6 +16,10 @@ ModelAutopilotCycleRecord, ModelAutopilotStepResult, ) +from onex_change_control.models.model_compliance_sweep_report import ( + ModelComplianceSweepReport, + ModelRepoComplianceBreakdown, +) from onex_change_control.models.model_contract_drift_input import ( ModelContractDriftInput, ) @@ -70,10 +74,17 @@ ModelGoldenPathInput, ModelGoldenPathOutput, ) +from onex_change_control.models.model_handler_compliance_result import ( + ModelHandlerComplianceResult, +) from onex_change_control.models.model_integration_record import ( ModelIntegrationProbeResult, ModelIntegrationRecord, ) +from onex_change_control.models.model_migration_spec import ( + ModelMigrationSpec, + ModelMigrationValidationResult, +) from onex_change_control.models.model_ticket_contract import ( ModelDodCheck, ModelDodEvidenceItem, @@ -85,6 +96,7 @@ "EnumAutopilotStepStatus", "ModelAutopilotCycleRecord", "ModelAutopilotStepResult", + "ModelComplianceSweepReport", "ModelContractDriftInput", "ModelContractDriftOutput", "ModelDayClose", @@ -116,8 +128,12 @@ "ModelGoldenPathAssertion", "ModelGoldenPathInput", "ModelGoldenPathOutput", + "ModelHandlerComplianceResult", "ModelIntegrationProbeResult", "ModelIntegrationRecord", + "ModelMigrationSpec", + "ModelMigrationValidationResult", + "ModelRepoComplianceBreakdown", "ModelRepoDocSummary", "ModelTicketContract", ] diff --git a/src/onex_change_control/models/model_compliance_sweep_report.py b/src/onex_change_control/models/model_compliance_sweep_report.py new file mode 100644 index 00000000000..90a69376b38 --- /dev/null +++ b/src/onex_change_control/models/model_compliance_sweep_report.py @@ -0,0 +1,74 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Compliance Sweep Report Model. + +Aggregated sweep report across all handlers in one or more repos. +""" + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + +from onex_change_control.models.model_handler_compliance_result import ( + ModelHandlerComplianceResult, +) + + +class ModelRepoComplianceBreakdown(BaseModel): + """Per-repo breakdown of compliance results.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + repo: str = Field(..., description="Repository name") + total_handlers: int = Field(..., description="Total handlers scanned", ge=0) + compliant: int = Field(..., description="Compliant handler count", ge=0) + imperative: int = Field(..., description="Imperative handler count", ge=0) + hybrid: int = Field(..., description="Hybrid handler count", ge=0) + top_violations: list[str] = Field( + default_factory=list, + description="Most common violation types in this repo", + ) + + +class ModelComplianceSweepReport(BaseModel): + """Aggregated compliance sweep report. + + Contains counts, ratios, per-repo breakdown, and full audit details + for all handlers scanned across one or more repositories. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + timestamp: datetime = Field( + ..., + description="When the sweep was performed", + ) + total_handlers: int = Field(..., description="Total handlers scanned", ge=0) + compliant_count: int = Field(..., description="Compliant handler count", ge=0) + imperative_count: int = Field(..., description="Imperative handler count", ge=0) + hybrid_count: int = Field(..., description="Hybrid handler count", ge=0) + allowlisted_count: int = Field(..., description="Allowlisted handler count", ge=0) + missing_contract_count: int = Field( + ..., description="Missing contract handler count", ge=0 + ) + compliant_pct: float = Field( + ..., + description="Percentage of compliant handlers (0-100)", + ) + violation_histogram: dict[str, int] = Field( + default_factory=dict, + description="Count per EnumComplianceViolation value", + ) + per_repo: dict[str, ModelRepoComplianceBreakdown] = Field( + default_factory=dict, + description="Breakdown by repository", + ) + results: list[ModelHandlerComplianceResult] = Field( + default_factory=list, + description="Full audit details", + ) + new_violations_since_last: list[str] = Field( + default_factory=list, + description="Handlers that gained violations since last sweep", + ) diff --git a/src/onex_change_control/models/model_handler_compliance_result.py b/src/onex_change_control/models/model_handler_compliance_result.py new file mode 100644 index 00000000000..0a7e1a09e9c --- /dev/null +++ b/src/onex_change_control/models/model_handler_compliance_result.py @@ -0,0 +1,84 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Handler Compliance Result Model. + +Single handler audit result for contract compliance checks. +""" + +from pydantic import BaseModel, ConfigDict, Field + +from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict +from onex_change_control.enums.enum_compliance_violation import EnumComplianceViolation + + +class ModelHandlerComplianceResult(BaseModel): + """Audit result for a single handler's contract compliance. + + Captures the cross-reference between a handler's actual behavior + (topics used, transports imported, routing registration) and what + is declared in its contract.yaml. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + handler_path: str = Field( + ..., + description="Relative path to handler file", + ) + node_dir: str = Field( + ..., + description="Parent node directory", + ) + repo: str = Field( + ..., + description="Repository name", + ) + contract_path: str | None = Field( + default=None, + description="Path to associated contract.yaml (None if missing)", + ) + violations: list[EnumComplianceViolation] = Field( + default_factory=list, + description="Specific violations found", + ) + violation_details: list[str] = Field( + default_factory=list, + description="Human-readable detail per violation", + ) + declared_topics: list[str] = Field( + default_factory=list, + description="Topics declared in contract.yaml", + ) + used_topics: list[str] = Field( + default_factory=list, + description="Topics referenced in handler source code", + ) + undeclared_topics: list[str] = Field( + default_factory=list, + description="Topics used but not declared", + ) + declared_transports: list[str] = Field( + default_factory=list, + description="Transports declared in contract capabilities", + ) + used_transports: list[str] = Field( + default_factory=list, + description="Transports detected in handler imports/calls", + ) + undeclared_transports: list[str] = Field( + default_factory=list, + description="Transports used but not declared", + ) + handler_in_routing: bool = Field( + default=False, + description="Whether handler is registered in contract.yaml handler_routing", + ) + verdict: EnumComplianceVerdict = Field( + ..., + description="Overall compliance verdict", + ) + allowlisted: bool = Field( + default=False, + description="Whether this handler is in the allowlist", + ) diff --git a/src/onex_change_control/models/model_migration_spec.py b/src/onex_change_control/models/model_migration_spec.py new file mode 100644 index 00000000000..f07a7ae2b23 --- /dev/null +++ b/src/onex_change_control/models/model_migration_spec.py @@ -0,0 +1,133 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Migration Spec Model. + +Migration plan for a single handler's imperative-to-declarative transition. +""" + +from __future__ import annotations + +from typing import Self + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from onex_change_control.enums.enum_compliance_violation import ( # noqa: TC001 + EnumComplianceViolation, +) +from onex_change_control.enums.enum_migration_status import EnumMigrationStatus + + +class ModelMigrationSpec(BaseModel): + """Migration plan for one handler. + + Captures the violations to fix, specific contract.yaml changes needed, + handler code changes, and current migration status. + + Lifecycle: PENDING -> GENERATED -> VALIDATED -> DEPLOYED -> RETIRED + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + handler_path: str = Field( + ..., + description="Path to imperative handler", + ) + node_dir: str = Field( + ..., + description="Parent node directory", + ) + contract_path: str = Field( + ..., + description="Path to contract.yaml (existing or to-be-created)", + ) + violations: list[EnumComplianceViolation] = Field( + default_factory=list, + description="Violations to fix", + ) + contract_changes: list[str] = Field( + default_factory=list, + description="Specific contract.yaml changes needed", + ) + handler_changes: list[str] = Field( + default_factory=list, + description="Specific handler changes needed", + ) + estimated_complexity: int = Field( + ..., + description="1-5 scale based on violation count and type", + ge=1, + le=5, + ) + status: EnumMigrationStatus = Field( + default=EnumMigrationStatus.PENDING, + description="Current migration status", + ) + ticket_id: str | None = Field( + default=None, + description="Linear ticket ID tracking this migration", + ) + + +class ModelMigrationValidationResult(BaseModel): + """Before/after validation result for a handler migration. + + Verifies that the migrated handler can be loaded via contract + dispatch and produces equivalent outputs. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + handler_path: str = Field( + ..., + description="Path to migrated handler", + ) + contract_dispatch_loads: bool = Field( + ..., + description="Whether handler loads successfully via contract routing", + ) + test_inputs_count: int = Field( + ..., + description="Number of test inputs run", + ge=0, + ) + tests_passed: int = Field( + ..., + description="Number of tests that produced equivalent output", + ge=0, + ) + tests_failed: int = Field( + ..., + description="Number of tests that produced different output", + ge=0, + ) + failure_details: list[str] = Field( + default_factory=list, + description="Details of any failed equivalence tests", + ) + passed: bool = Field( + ..., + description="Overall pass/fail: all tests passed and dispatch loads", + ) + + @model_validator(mode="after") + def check_consistency(self) -> Self: + """Enforce test count and passed-flag invariants.""" + if self.tests_passed + self.tests_failed != self.test_inputs_count: + msg = ( + f"tests_passed ({self.tests_passed}) " + f"+ tests_failed ({self.tests_failed}) " + f"!= test_inputs_count ({self.test_inputs_count})" + ) + raise ValueError(msg) + if self.passed and ( + not self.contract_dispatch_loads + or self.tests_failed != 0 + or self.tests_passed != self.test_inputs_count + ): + msg = ( + "passed=True requires contract_dispatch_loads=True, " + "tests_failed=0, and all tests passed" + ) + raise ValueError(msg) + return self diff --git a/src/onex_change_control/scanners/__init__.py b/src/onex_change_control/scanners/__init__.py index 7d3f20a057b..c11f15889cd 100644 --- a/src/onex_change_control/scanners/__init__.py +++ b/src/onex_change_control/scanners/__init__.py @@ -1,4 +1,4 @@ # SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. # SPDX-License-Identifier: MIT -"""Doc freshness scanners for reference extraction and resolution.""" +"""ONEX Change Control scanners for doc freshness and handler compliance.""" diff --git a/src/onex_change_control/scanners/handler_contract_compliance.py b/src/onex_change_control/scanners/handler_contract_compliance.py new file mode 100644 index 00000000000..3386df23a0f --- /dev/null +++ b/src/onex_change_control/scanners/handler_contract_compliance.py @@ -0,0 +1,516 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Handler-contract cross-reference scanner. + +AST-based analysis of handler Python files and their contract.yaml +declarations to detect contract compliance violations. + +Uses ast.walk() to find string literals and import statements, +skipping docstrings and comments to reduce false positives. +""" + +from __future__ import annotations + +import ast +import re +from typing import TYPE_CHECKING, Any + +import yaml + +from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict +from onex_change_control.enums.enum_compliance_violation import EnumComplianceViolation +from onex_change_control.models.model_handler_compliance_result import ( + ModelHandlerComplianceResult, +) + +if TYPE_CHECKING: + from pathlib import Path + +# --- Topic detection patterns --- + +# Known ONEX topic patterns +_ONEX_TOPIC_RE = re.compile(r"onex\.evt\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+\.v\d+") + +# Known bare topic names used in the platform +_BARE_TOPIC_NAMES: frozenset[str] = frozenset( + { + "agent-actions", + "agent-transformation-events", + "agent.routing.requested.v1", + "agent.routing.completed.v1", + "agent.routing.failed.v1", + "router-performance-metrics", + "documentation-changed", + } +) + +# --- Transport detection --- + +# Call-site transport indicators (precise call-site detection only) +_TRANSPORT_CALLSITE_PATTERNS: dict[str, str] = { + "session.execute": "DATABASE", + "session.query": "DATABASE", + "asyncpg.connect": "DATABASE", + "asyncpg.create_pool": "DATABASE", + "psycopg.connect": "DATABASE", + "create_engine": "DATABASE", + "create_async_engine": "DATABASE", + "AsyncClient": "HTTP", + "httpx.get": "HTTP", + "httpx.post": "HTTP", + "requests.get": "HTTP", + "requests.post": "HTTP", + "KafkaProducer": "KAFKA", + "KafkaConsumer": "KAFKA", + "AIOKafkaProducer": "KAFKA", + "AIOKafkaConsumer": "KAFKA", + "QdrantClient": "QDRANT", +} + +_IMPERATIVE_VIOLATION_THRESHOLD = 2 + + +def parse_contract_topics(contract_path: Path) -> tuple[list[str], list[str]]: + """Extract publish and subscribe topics from contract.yaml. + + Returns: + Tuple of (publish_topics, subscribe_topics). + """ + data = _load_yaml(contract_path) + if data is None: + return [], [] + + event_bus = data.get("event_bus", {}) or {} + publish = event_bus.get("publish_topics", []) or [] + subscribe = event_bus.get("subscribe_topics", []) or [] + return list(publish), list(subscribe) + + +def parse_contract_transports(contract_path: Path) -> list[str]: + """Extract declared transport types from contract.yaml. + + Looks at metadata.transport_type and handler_routing.handlers[].handler_type. + """ + data = _load_yaml(contract_path) + if data is None: + return [] + + transports: list[str] = [] + + # Check metadata.transport_type + metadata = data.get("metadata", {}) or {} + if transport := metadata.get("transport_type"): + transports.append(str(transport).upper()) + + # Check handler_routing.handlers[].handler_type + handler_routing = data.get("handler_routing", {}) or {} + for handler_entry in handler_routing.get("handlers", []) or []: + handler_info = handler_entry.get("handler", {}) or {} + if handler_type := handler_info.get("handler_type"): + transports.append(str(handler_type).upper()) + + # Infer transports from node_type and topics + _infer_kafka_transport(data, transports) + + return transports + + +def _infer_kafka_transport(data: dict[str, Any], transports: list[str]) -> None: + """Add KAFKA transport if EFFECT node has declared topics.""" + node_type = data.get("node_type", "") + if "EFFECT" not in str(node_type): + return + event_bus = data.get("event_bus", {}) or {} + has_topics = event_bus.get("publish_topics") or event_bus.get("subscribe_topics") + if has_topics and "KAFKA" not in transports: + transports.append("KAFKA") + + +def parse_contract_handler_routing( + contract_path: Path, +) -> list[dict[str, Any]]: + """Extract handler routing entries from contract.yaml. + + Returns list of handler entries with name, module, operation. + """ + data = _load_yaml(contract_path) + if data is None: + return [] + + handler_routing = data.get("handler_routing", {}) or {} + entries: list[dict[str, Any]] = [] + for handler_entry in handler_routing.get("handlers", []) or []: + handler_info = handler_entry.get("handler", {}) or {} + entries.append( + { + "name": handler_info.get("name", ""), + "module": handler_info.get("module", ""), + "operation": handler_entry.get("operation", ""), + } + ) + return entries + + +def scan_handler_topics(handler_path: Path) -> list[str]: + """Find topic string literals in handler code using AST. + + Only flags string literals inside function bodies, + skipping docstrings and comments. + """ + source = handler_path.read_text(encoding="utf-8") + try: + tree = ast.parse(source) + except SyntaxError: + return [] + + topics: list[str] = [] + docstring_nodes = _get_docstring_nodes(tree) + + for node in ast.walk(tree): + if not (isinstance(node, ast.Constant) and isinstance(node.value, str)): + continue + if id(node) in docstring_nodes: + continue + value = node.value + if _ONEX_TOPIC_RE.search(value) or value in _BARE_TOPIC_NAMES: + topics.append(value) + + return sorted(set(topics)) + + +def scan_handler_transports(handler_path: Path) -> list[str]: + """Detect transport usage in handler via AST. + + Only returns transports confirmed by call-site usage, not bare imports. + Import-only references (including TYPE_CHECKING guards) are excluded + to avoid false positives. + + Returns deduplicated list of transport types found at call sites. + """ + source = handler_path.read_text(encoding="utf-8") + try: + tree = ast.parse(source) + except SyntaxError: + return [] + + callsite_transports: set[str] = set() + + for node in ast.walk(tree): + _check_callsite_transports(node, callsite_transports) + + return sorted(callsite_transports) + + +def _check_callsite_transports(node: ast.AST, transports: set[str]) -> None: + """Check call-site and attribute nodes for transport usage.""" + call_str: str | None = None + if isinstance(node, ast.Call): + call_str = _get_call_string(node) + elif isinstance(node, ast.Attribute): + call_str = _get_attribute_string(node) + + if call_str: + for pattern, transport in _TRANSPORT_CALLSITE_PATTERNS.items(): + if pattern in call_str: + transports.add(transport) + + +def scan_node_py_logic(node_py_path: Path) -> list[str]: + """Return custom method names in node.py beyond __init__. + + A clean declarative node should only have __init__ calling super().__init__. + Any other methods indicate business logic in the wrong layer. + """ + if not node_py_path.exists(): + return [] + + source = node_py_path.read_text(encoding="utf-8") + try: + tree = ast.parse(source) + except SyntaxError: + return [] + + custom_methods: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef): + for item in node.body: + if ( + isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)) + and item.name != "__init__" + ): + custom_methods.append(item.name) + + return custom_methods + + +def cross_reference( + node_dir: Path, + repo: str, + allowlisted_paths: frozenset[str] | None = None, +) -> list[ModelHandlerComplianceResult]: + """Run all checks for all handlers in a node directory. + + Args: + node_dir: Path to node directory containing contract.yaml and handlers/. + repo: Repository name. + allowlisted_paths: Set of handler paths that are allowlisted. + + Returns: + List of compliance results, one per handler file. + """ + if allowlisted_paths is None: + allowlisted_paths = frozenset() + + contract_ctx = _build_contract_context(node_dir) + node_logic = scan_node_py_logic(node_dir / "node.py") + + handlers_dir = node_dir / "handlers" + if not handlers_dir.exists(): + return [] + + handler_files = sorted( + f + for f in handlers_dir.rglob("*.py") + if f.name != "__init__.py" and not f.name.startswith("_") + ) + + return [ + _audit_handler( + handler_file=hf, + node_dir=node_dir, + repo=repo, + contract_ctx=contract_ctx, + node_logic=node_logic, + allowlisted_paths=allowlisted_paths, + ) + for hf in handler_files + ] + + +def _build_contract_context( + node_dir: Path, +) -> dict[str, Any]: + """Extract all contract info needed for handler auditing.""" + contract_path = node_dir / "contract.yaml" + has_contract = contract_path.exists() + if not has_contract: + return { + "has_contract": False, + "declared_topics": [], + "declared_transports": [], + "routed_modules": set(), + "contract_path": None, + } + + publish_topics, subscribe_topics = parse_contract_topics(contract_path) + routing_entries = parse_contract_handler_routing(contract_path) + return { + "has_contract": True, + "declared_topics": sorted(set(publish_topics + subscribe_topics)), + "declared_transports": parse_contract_transports(contract_path), + "routed_modules": {e["module"] for e in routing_entries}, + "contract_path": contract_path, + } + + +def _audit_handler( # noqa: PLR0913 + handler_file: Path, + node_dir: Path, + repo: str, + contract_ctx: dict[str, Any], + node_logic: list[str], + allowlisted_paths: frozenset[str], +) -> ModelHandlerComplianceResult: + """Audit a single handler file against its contract context.""" + base_dir = node_dir.parent.parent.parent + rel_path = str(handler_file.relative_to(base_dir)) + is_allowlisted = rel_path in allowlisted_paths + + if not contract_ctx["has_contract"]: + return ModelHandlerComplianceResult( + handler_path=rel_path, + node_dir=str(node_dir.relative_to(base_dir)), + repo=repo, + contract_path=None, + verdict=EnumComplianceVerdict.MISSING_CONTRACT, + allowlisted=is_allowlisted, + ) + + violations, violation_details = _collect_violations( + handler_file, node_dir, contract_ctx, node_logic + ) + used_topics = scan_handler_topics(handler_file) + used_transports = scan_handler_transports(handler_file) + undeclared = [t for t in used_topics if t not in contract_ctx["declared_topics"]] + undeclared_transports = [ + t for t in used_transports if t not in contract_ctx["declared_transports"] + ] + handler_module = _infer_module_path(handler_file, node_dir) + in_routing = handler_module in contract_ctx["routed_modules"] + + verdict = _determine_verdict(violations, is_allowlisted=is_allowlisted) + contract_rel = str(contract_ctx["contract_path"].relative_to(base_dir)) + + return ModelHandlerComplianceResult( + handler_path=rel_path, + node_dir=str(node_dir.relative_to(base_dir)), + repo=repo, + contract_path=contract_rel, + violations=violations, + violation_details=violation_details, + declared_topics=contract_ctx["declared_topics"], + used_topics=used_topics, + undeclared_topics=undeclared, + declared_transports=contract_ctx["declared_transports"], + used_transports=used_transports, + undeclared_transports=undeclared_transports, + handler_in_routing=in_routing, + verdict=verdict, + allowlisted=is_allowlisted, + ) + + +def _collect_violations( + handler_file: Path, + node_dir: Path, + contract_ctx: dict[str, Any], + node_logic: list[str], +) -> tuple[list[EnumComplianceViolation], list[str]]: + """Collect all violations for a handler file.""" + violations: list[EnumComplianceViolation] = [] + details: list[str] = [] + + # Check 1: Topic compliance — any hardcoded topic literal is a violation, + # even if declared in the contract. Declaration and hardcoding are orthogonal: + # handlers should use contract-driven dispatch, not string literals. + used_topics = scan_handler_topics(handler_file) + for topic in used_topics: + violations.append(EnumComplianceViolation.HARDCODED_TOPIC) + if topic in contract_ctx["declared_topics"]: + details.append( + f"hardcoded topic '{topic}' (declared but should use contract dispatch)" + ) + else: + details.append(f"hardcoded topic '{topic}' not in contract") + + # Check 2: Transport compliance + used_transports = scan_handler_transports(handler_file) + for transport in used_transports: + if transport not in contract_ctx["declared_transports"]: + violations.append(EnumComplianceViolation.UNDECLARED_TRANSPORT) + details.append(f"undeclared transport {transport} used in handler") + + # Check 3: Handler routing registration + handler_module = _infer_module_path(handler_file, node_dir) + if handler_module not in contract_ctx["routed_modules"]: + violations.append(EnumComplianceViolation.MISSING_HANDLER_ROUTING) + details.append("handler not registered in contract handler_routing") + + # Check 4: Logic in node.py + if node_logic: + violations.append(EnumComplianceViolation.LOGIC_IN_NODE) + details.append(f"node.py has custom methods: {', '.join(node_logic)}") + + return violations, details + + +def _determine_verdict( + violations: list[EnumComplianceViolation], + *, + is_allowlisted: bool, +) -> EnumComplianceVerdict: + """Determine the compliance verdict from violations.""" + if is_allowlisted: + return EnumComplianceVerdict.ALLOWLISTED + if not violations: + return EnumComplianceVerdict.COMPLIANT + if len(violations) >= _IMPERATIVE_VIOLATION_THRESHOLD: + return EnumComplianceVerdict.IMPERATIVE + return EnumComplianceVerdict.HYBRID + + +# --- Internal helpers --- + + +def _load_yaml(path: Path) -> dict[str, Any] | None: + """Load and return a YAML file as a dict.""" + if not path.exists(): + return None + with path.open(encoding="utf-8") as f: + data = yaml.safe_load(f) + if not isinstance(data, dict): + return None + return data + + +def _get_docstring_nodes(tree: ast.Module) -> set[int]: + """Collect ids of AST nodes that are docstrings.""" + docstring_ids: set[int] = set() + for node in ast.walk(tree): + if isinstance( + node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Module) + ) and ( + node.body + and isinstance(node.body[0], ast.Expr) + and isinstance(node.body[0].value, ast.Constant) + and isinstance(node.body[0].value.value, str) + ): + docstring_ids.add(id(node.body[0].value)) + return docstring_ids + + +def _get_call_string(node: ast.Call) -> str | None: + """Extract a dotted string representation of a Call node's function.""" + if isinstance(node.func, ast.Name): + return node.func.id + if isinstance(node.func, ast.Attribute): + return _get_attribute_string(node.func) + return None + + +def _get_attribute_string(node: ast.Attribute) -> str | None: + """Extract dotted string from an Attribute node.""" + parts: list[str] = [node.attr] + current: ast.expr = node.value + while isinstance(current, ast.Attribute): + parts.append(current.attr) + current = current.value + if isinstance(current, ast.Name): + parts.append(current.id) + parts.reverse() + return ".".join(parts) + + +def _infer_module_path(handler_file: Path, node_dir: Path) -> str: + """Infer the Python module path for a handler file. + + E.g., for handler at: + .../src/omnibase_infra/nodes/node_foo/handlers/handler_bar.py + Returns: + omnibase_infra.nodes.node_foo.handlers.handler_bar + """ + # Walk up to find 'src/' directory + parts = handler_file.parts + src_idx = None + for i, part in enumerate(parts): + if part == "src": + src_idx = i + break + + if src_idx is not None: + module_parts = parts[src_idx + 1 :] + else: + # Fallback: use relative to node_dir parent + try: + rel = handler_file.relative_to(node_dir.parent.parent) + module_parts = rel.parts + except ValueError: + return "" + + # Strip .py extension from last part + last = module_parts[-1] + if last.endswith(".py"): + module_parts = (*module_parts[:-1], last[:-3]) + + return ".".join(module_parts) diff --git a/src/onex_change_control/validators/arch_handler_contract_compliance.py b/src/onex_change_control/validators/arch_handler_contract_compliance.py new file mode 100644 index 00000000000..325646b4d92 --- /dev/null +++ b/src/onex_change_control/validators/arch_handler_contract_compliance.py @@ -0,0 +1,223 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""CI validator: arch-handler-contract-compliance. + +Scans all node directories in a repo for handler contract compliance +violations. Exits non-zero if any new (non-allowlisted) violation is found. + +Usage: + uv run python -m onex_change_control.validators.arch_handler_contract_compliance \ + --repo-root . --allowlist-path arch-handler-contract-compliance-allowlist.yaml + + # Generate initial allowlist from current violations + uv run python -m onex_change_control.validators.arch_handler_contract_compliance \ + --repo-root . --generate-allowlist +""" + +from __future__ import annotations + +import argparse +import json +import logging +import pathlib +import sys +from typing import TYPE_CHECKING, Any + +import yaml + +if TYPE_CHECKING: + from pathlib import Path + +from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict +from onex_change_control.scanners.handler_contract_compliance import cross_reference + +logger = logging.getLogger(__name__) + + +def _find_node_dirs(repo_root: Path) -> list[Path]: + """Find all node directories (containing contract.yaml or handlers/).""" + src_dir = repo_root / "src" + if not src_dir.exists(): + return [] + + node_dirs: list[Path] = [] + for nodes_dir in src_dir.rglob("nodes"): + if not nodes_dir.is_dir(): + continue + for child in sorted(nodes_dir.iterdir()): + if child.is_dir() and child.name.startswith("node_"): + handlers_dir = child / "handlers" + if handlers_dir.exists(): + node_dirs.append(child) + + return node_dirs + + +def _load_allowlist(allowlist_path: Path) -> dict[str, list[str]]: + """Load allowlist YAML, returning handler_path -> list of violation types.""" + if not allowlist_path.exists(): + return {} + + with allowlist_path.open(encoding="utf-8") as f: + data = yaml.safe_load(f) + + if not isinstance(data, dict): + return {} + + result: dict[str, list[str]] = {} + for entry in data.get("allowlisted_handlers", []) or []: + path = entry.get("path", "") + violations = entry.get("violations", []) + if path: + result[path] = violations + + return result + + +def _infer_repo_name(repo_root: Path) -> str: + """Infer repository name from the repo root directory.""" + src_dir = repo_root / "src" + if src_dir.exists(): + for child in src_dir.iterdir(): + if child.is_dir() and not child.name.startswith("."): + return child.name + return repo_root.name + + +def _write(msg: str) -> None: + """Write a line to stdout (CI output).""" + sys.stdout.write(msg + "\n") + + +def run_scan( + repo_root: Path, + allowlist_path: Path | None = None, + *, + generate_allowlist: bool = False, + output_json: bool = False, +) -> int: + """Run the handler contract compliance scan. + + Returns: + Exit code: 0 if clean, 1 if violations found. + """ + repo_name = _infer_repo_name(repo_root) + node_dirs = _find_node_dirs(repo_root) + + if not node_dirs: + _write(f"No node directories found in {repo_root}") + return 0 + + # Load allowlist + allowlist = _load_allowlist(allowlist_path) if allowlist_path else {} + allowlisted_paths = frozenset(allowlist.keys()) + + # Scan all nodes + all_results = [] + for node_dir in node_dirs: + results = cross_reference( + node_dir=node_dir, + repo=repo_name, + allowlisted_paths=allowlisted_paths, + ) + all_results.extend(results) + + if generate_allowlist: + _output_allowlist(all_results) + return 0 + + if output_json: + json_output = [r.model_dump(mode="json") for r in all_results] + _write(json.dumps(json_output, indent=2)) + + # Count violations + new_violations = [r for r in all_results if r.violations and not r.allowlisted] + + compliant = sum( + 1 + for r in all_results + if not r.violations and r.verdict == EnumComplianceVerdict.COMPLIANT + ) + allowlisted_count = sum(1 for r in all_results if r.allowlisted) + total = len(all_results) + + _write(f"\n=== Handler Contract Compliance: {repo_name} ===") + _write(f"Total handlers: {total}") + _write(f"Compliant: {compliant}") + _write(f"Allowlisted: {allowlisted_count}") + _write(f"New violations: {len(new_violations)}") + + if new_violations: + _write("\n--- New violations (not allowlisted) ---") + for r in new_violations: + _write(f"\n {r.handler_path}") + _write(f" Verdict: {r.verdict}") + for detail in r.violation_details: + _write(f" - {detail}") + return 1 + + return 0 + + +def _output_allowlist( + results: list[Any], +) -> None: + """Output current violations as allowlist YAML.""" + entries = [] + for r in results: + if r.violations: + entries.append( + { + "path": r.handler_path, + "violations": [str(v) for v in r.violations], + "ticket": "# migration pending", + } + ) + + allowlist = {"allowlisted_handlers": entries} + _write(yaml.dump(allowlist, default_flow_style=False, sort_keys=False)) + + +def main() -> None: + """CLI entry point.""" + logging.basicConfig(level=logging.INFO) + parser = argparse.ArgumentParser( + description="Handler contract compliance validator" + ) + parser.add_argument( + "--repo-root", + type=pathlib.Path, + required=True, + help="Repository root path", + ) + parser.add_argument( + "--allowlist-path", + type=pathlib.Path, + default=None, + help="Path to allowlist YAML", + ) + parser.add_argument( + "--generate-allowlist", + action="store_true", + help="Output current violations as allowlist", + ) + parser.add_argument( + "--json", + dest="output_json", + action="store_true", + help="Output JSON report", + ) + + args = parser.parse_args() + exit_code = run_scan( + repo_root=args.repo_root, + allowlist_path=args.allowlist_path, + generate_allowlist=args.generate_allowlist, + output_json=args.output_json, + ) + sys.exit(exit_code) + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_compliance_models.py b/tests/unit/test_compliance_models.py new file mode 100644 index 00000000000..4d507552368 --- /dev/null +++ b/tests/unit/test_compliance_models.py @@ -0,0 +1,378 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Tests for handler contract compliance models and enums.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict +from onex_change_control.enums.enum_compliance_violation import EnumComplianceViolation +from onex_change_control.enums.enum_migration_status import EnumMigrationStatus +from onex_change_control.models.model_compliance_sweep_report import ( + ModelComplianceSweepReport, + ModelRepoComplianceBreakdown, +) +from onex_change_control.models.model_handler_compliance_result import ( + ModelHandlerComplianceResult, +) +from onex_change_control.models.model_migration_spec import ( + ModelMigrationSpec, + ModelMigrationValidationResult, +) + +# --- Enum Tests --- + + +class TestEnumComplianceVerdict: + def test_all_values(self) -> None: + assert set(EnumComplianceVerdict) == { + EnumComplianceVerdict.COMPLIANT, + EnumComplianceVerdict.IMPERATIVE, + EnumComplianceVerdict.HYBRID, + EnumComplianceVerdict.ALLOWLISTED, + EnumComplianceVerdict.MISSING_CONTRACT, + } + + def test_str_serialization(self) -> None: + assert str(EnumComplianceVerdict.COMPLIANT) == "compliant" + assert str(EnumComplianceVerdict.IMPERATIVE) == "imperative" + + def test_is_str_enum(self) -> None: + assert isinstance(EnumComplianceVerdict.COMPLIANT, str) + + +class TestEnumComplianceViolation: + def test_all_values(self) -> None: + assert len(EnumComplianceViolation) == 8 + + def test_str_serialization(self) -> None: + assert str(EnumComplianceViolation.HARDCODED_TOPIC) == "hardcoded_topic" + assert str(EnumComplianceViolation.DIRECT_DB_ACCESS) == "direct_db_access" + + def test_is_str_enum(self) -> None: + assert isinstance(EnumComplianceViolation.HARDCODED_TOPIC, str) + + +class TestEnumMigrationStatus: + def test_all_values(self) -> None: + assert set(EnumMigrationStatus) == { + EnumMigrationStatus.PENDING, + EnumMigrationStatus.GENERATED, + EnumMigrationStatus.VALIDATED, + EnumMigrationStatus.DEPLOYED, + EnumMigrationStatus.RETIRED, + } + + def test_str_serialization(self) -> None: + assert str(EnumMigrationStatus.PENDING) == "pending" + + +# --- Model Tests --- + + +@pytest.fixture +def compliant_result() -> ModelHandlerComplianceResult: + return ModelHandlerComplianceResult( + handler_path="src/omnibase_infra/nodes/node_foo/handlers/handler_bar.py", + node_dir="src/omnibase_infra/nodes/node_foo", + repo="omnibase_infra", + contract_path="src/omnibase_infra/nodes/node_foo/contract.yaml", + violations=[], + violation_details=[], + declared_topics=["onex.evt.foo.v1"], + used_topics=["onex.evt.foo.v1"], + undeclared_topics=[], + declared_transports=["KAFKA"], + used_transports=["KAFKA"], + undeclared_transports=[], + handler_in_routing=True, + verdict=EnumComplianceVerdict.COMPLIANT, + allowlisted=False, + ) + + +@pytest.fixture +def imperative_result() -> ModelHandlerComplianceResult: + return ModelHandlerComplianceResult( + handler_path="src/omnibase_infra/nodes/node_baz/handlers/handler_qux.py", + node_dir="src/omnibase_infra/nodes/node_baz", + repo="omnibase_infra", + contract_path="src/omnibase_infra/nodes/node_baz/contract.yaml", + violations=[ + EnumComplianceViolation.HARDCODED_TOPIC, + EnumComplianceViolation.UNDECLARED_TRANSPORT, + ], + violation_details=[ + "hardcoded topic 'agent-actions' at line 47", + "undeclared transport HTTP (httpx import at line 3)", + ], + declared_topics=["onex.evt.baz.v1"], + used_topics=["onex.evt.baz.v1", "agent-actions"], + undeclared_topics=["agent-actions"], + declared_transports=["KAFKA"], + used_transports=["KAFKA", "HTTP"], + undeclared_transports=["HTTP"], + handler_in_routing=True, + verdict=EnumComplianceVerdict.IMPERATIVE, + allowlisted=False, + ) + + +class TestModelHandlerComplianceResult: + def test_frozen(self, compliant_result: ModelHandlerComplianceResult) -> None: + with pytest.raises(ValidationError): + compliant_result.repo = "changed" # type: ignore[misc] + + def test_extra_forbid(self) -> None: + with pytest.raises(ValidationError): + ModelHandlerComplianceResult( + handler_path="x", + node_dir="x", + repo="x", + verdict=EnumComplianceVerdict.COMPLIANT, + extra_field="bad", # type: ignore[call-arg] + ) + + def test_roundtrip_json( + self, imperative_result: ModelHandlerComplianceResult + ) -> None: + json_str = imperative_result.model_dump_json() + parsed = json.loads(json_str) + restored = ModelHandlerComplianceResult.model_validate(parsed) + assert restored == imperative_result + + def test_compliant_has_no_violations( + self, compliant_result: ModelHandlerComplianceResult + ) -> None: + assert compliant_result.violations == [] + assert compliant_result.verdict == EnumComplianceVerdict.COMPLIANT + + def test_imperative_has_violations( + self, imperative_result: ModelHandlerComplianceResult + ) -> None: + assert len(imperative_result.violations) == 2 + assert imperative_result.verdict == EnumComplianceVerdict.IMPERATIVE + + def test_missing_contract(self) -> None: + result = ModelHandlerComplianceResult( + handler_path="x", + node_dir="x", + repo="x", + contract_path=None, + verdict=EnumComplianceVerdict.MISSING_CONTRACT, + ) + assert result.contract_path is None + + +class TestModelComplianceSweepReport: + def test_empty_report(self) -> None: + report = ModelComplianceSweepReport( + timestamp=datetime.now(tz=UTC), + total_handlers=0, + compliant_count=0, + imperative_count=0, + hybrid_count=0, + allowlisted_count=0, + missing_contract_count=0, + compliant_pct=0.0, + ) + assert report.total_handlers == 0 + assert report.compliant_pct == 0.0 + + def test_percentage_accepts_boundary_values(self) -> None: + report = ModelComplianceSweepReport( + timestamp=datetime.now(tz=UTC), + total_handlers=10, + compliant_count=10, + imperative_count=0, + hybrid_count=0, + allowlisted_count=0, + missing_contract_count=0, + compliant_pct=100.0, + ) + assert report.compliant_pct == 100.0 + + zero_report = ModelComplianceSweepReport( + timestamp=datetime.now(tz=UTC), + total_handlers=0, + compliant_count=0, + imperative_count=0, + hybrid_count=0, + allowlisted_count=0, + missing_contract_count=0, + compliant_pct=0.0, + ) + assert zero_report.compliant_pct == 0.0 + + def test_frozen(self) -> None: + report = ModelComplianceSweepReport( + timestamp=datetime.now(tz=UTC), + total_handlers=0, + compliant_count=0, + imperative_count=0, + hybrid_count=0, + allowlisted_count=0, + missing_contract_count=0, + compliant_pct=0.0, + ) + with pytest.raises(ValidationError): + report.total_handlers = 5 # type: ignore[misc] + + def test_roundtrip_json( + self, compliant_result: ModelHandlerComplianceResult + ) -> None: + breakdown = ModelRepoComplianceBreakdown( + repo="omnibase_infra", + total_handlers=1, + compliant=1, + imperative=0, + hybrid=0, + top_violations=[], + ) + report = ModelComplianceSweepReport( + timestamp=datetime(2026, 3, 27, tzinfo=UTC), + total_handlers=1, + compliant_count=1, + imperative_count=0, + hybrid_count=0, + allowlisted_count=0, + missing_contract_count=0, + compliant_pct=100.0, + violation_histogram={}, + per_repo={"omnibase_infra": breakdown}, + results=[compliant_result], + ) + json_str = report.model_dump_json() + parsed = json.loads(json_str) + restored = ModelComplianceSweepReport.model_validate(parsed) + assert restored.total_handlers == 1 + assert len(restored.results) == 1 + + def test_violation_histogram(self) -> None: + report = ModelComplianceSweepReport( + timestamp=datetime.now(tz=UTC), + total_handlers=5, + compliant_count=2, + imperative_count=3, + hybrid_count=0, + allowlisted_count=0, + missing_contract_count=0, + compliant_pct=40.0, + violation_histogram={ + "hardcoded_topic": 3, + "undeclared_transport": 1, + }, + ) + assert report.violation_histogram["hardcoded_topic"] == 3 + + +class TestModelMigrationSpec: + def test_frozen(self) -> None: + spec = ModelMigrationSpec( + handler_path="x", + node_dir="x", + contract_path="x/contract.yaml", + estimated_complexity=1, + ) + with pytest.raises(ValidationError): + spec.status = EnumMigrationStatus.GENERATED # type: ignore[misc] + + def test_roundtrip_json(self) -> None: + spec = ModelMigrationSpec( + handler_path="src/omnibase_infra/nodes/node_foo/handlers/handler_bar.py", + node_dir="src/omnibase_infra/nodes/node_foo", + contract_path="src/omnibase_infra/nodes/node_foo/contract.yaml", + violations=[EnumComplianceViolation.HARDCODED_TOPIC], + contract_changes=["add 'agent-actions' to event_bus.publish_topics"], + handler_changes=["replace hardcoded topic with contract lookup"], + estimated_complexity=2, + status=EnumMigrationStatus.GENERATED, + ticket_id="OMN-1234", + ) + json_str = spec.model_dump_json() + parsed = json.loads(json_str) + restored = ModelMigrationSpec.model_validate(parsed) + assert restored == spec + + def test_complexity_bounds(self) -> None: + with pytest.raises(ValidationError): + ModelMigrationSpec( + handler_path="x", + node_dir="x", + contract_path="x", + estimated_complexity=0, + ) + with pytest.raises(ValidationError): + ModelMigrationSpec( + handler_path="x", + node_dir="x", + contract_path="x", + estimated_complexity=6, + ) + + def test_default_status(self) -> None: + spec = ModelMigrationSpec( + handler_path="x", + node_dir="x", + contract_path="x", + estimated_complexity=1, + ) + assert spec.status == EnumMigrationStatus.PENDING + assert spec.ticket_id is None + + +class TestModelMigrationValidationResult: + def test_frozen(self) -> None: + result = ModelMigrationValidationResult( + handler_path="x", + contract_dispatch_loads=True, + test_inputs_count=5, + tests_passed=5, + tests_failed=0, + passed=True, + ) + with pytest.raises(ValidationError): + result.passed = False # type: ignore[misc] + + def test_roundtrip_json(self) -> None: + result = ModelMigrationValidationResult( + handler_path="src/omnibase_infra/nodes/node_foo/handlers/handler_bar.py", + contract_dispatch_loads=True, + test_inputs_count=3, + tests_passed=2, + tests_failed=1, + failure_details=["Input 3: expected output A, got output B"], + passed=False, + ) + json_str = result.model_dump_json() + parsed = json.loads(json_str) + restored = ModelMigrationValidationResult.model_validate(parsed) + assert restored == result + + def test_count_mismatch_rejected(self) -> None: + with pytest.raises(ValidationError, match="test_inputs_count"): + ModelMigrationValidationResult( + handler_path="x", + contract_dispatch_loads=True, + test_inputs_count=5, + tests_passed=3, + tests_failed=1, + passed=False, + ) + + def test_passed_requires_all_green(self) -> None: + with pytest.raises(ValidationError, match="passed=True requires"): + ModelMigrationValidationResult( + handler_path="x", + contract_dispatch_loads=False, + test_inputs_count=5, + tests_passed=5, + tests_failed=0, + passed=True, + ) diff --git a/tests/unit/test_handler_contract_compliance_scanner.py b/tests/unit/test_handler_contract_compliance_scanner.py new file mode 100644 index 00000000000..548d5b89bf6 --- /dev/null +++ b/tests/unit/test_handler_contract_compliance_scanner.py @@ -0,0 +1,356 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Tests for handler contract compliance scanner.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict +from onex_change_control.enums.enum_compliance_violation import EnumComplianceViolation +from onex_change_control.scanners.handler_contract_compliance import ( + cross_reference, + parse_contract_handler_routing, + parse_contract_topics, + parse_contract_transports, + scan_handler_topics, + scan_handler_transports, + scan_node_py_logic, +) + +# --- Fixtures --- + + +@pytest.fixture +def tmp_node_dir(tmp_path: Path) -> Path: + """Create a minimal node directory structure under src/.""" + src = tmp_path / "src" / "test_repo" / "nodes" / "node_test" + src.mkdir(parents=True) + handlers = src / "handlers" + handlers.mkdir() + (handlers / "__init__.py").write_text("") + return src + + +@pytest.fixture +def compliant_contract(tmp_node_dir: Path) -> Path: + """Write a compliant contract.yaml.""" + contract = tmp_node_dir / "contract.yaml" + contract.write_text( + """ +name: "node_test" +node_type: "EFFECT_GENERIC" +contract_version: + major: 1 + minor: 0 + patch: 0 +event_bus: + publish_topics: + - "onex.evt.test.computed.v1" + subscribe_topics: + - "onex.evt.test.requested.v1" +handler_routing: + routing_strategy: "operation_match" + handlers: + - operation: "compute" + handler: + name: "HandlerTest" + module: "test_repo.nodes.node_test.handlers.handler_test" +""" + ) + return contract + + +@pytest.fixture +def compliant_handler(tmp_node_dir: Path) -> Path: + """Write a compliant handler (no hardcoded topics, no undeclared transports).""" + handler = tmp_node_dir / "handlers" / "handler_test.py" + handler.write_text( + ''' +"""Handler for test compute operation.""" + +from __future__ import annotations + + +class HandlerTest: + """Test handler - fully compliant.""" + + async def handle(self, input_data: dict) -> dict: + """Process input and return result.""" + return {"status": "ok"} +''' + ) + return handler + + +@pytest.fixture +def imperative_handler(tmp_node_dir: Path) -> Path: + """Write a handler with hardcoded topics and undeclared transports.""" + handler = tmp_node_dir / "handlers" / "handler_imperative.py" + handler.write_text( + ''' +"""Imperative handler with violations.""" + +from __future__ import annotations + +import httpx + + +class HandlerImperative: + """Handler with contract violations.""" + + async def handle(self, input_data: dict) -> dict: + """Process with hardcoded topic and HTTP transport.""" + topic = "agent-actions" + another = "onex.evt.bad.topic.v1" + async with httpx.AsyncClient() as client: + resp = await client.get("http://example.com") + return {"topic": topic, "another": another} +''' + ) + return handler + + +@pytest.fixture +def clean_node_py(tmp_node_dir: Path) -> Path: + """Write a clean declarative node.py.""" + node_py = tmp_node_dir / "node.py" + node_py.write_text( + ''' +"""Declarative node.""" + +class NodeTest: + """Clean declarative node.""" + + def __init__(self, container): + super().__init__(container) +''' + ) + return node_py + + +@pytest.fixture +def logic_node_py(tmp_node_dir: Path) -> Path: + """Write a node.py with custom logic.""" + node_py = tmp_node_dir / "node.py" + node_py.write_text( + ''' +"""Node with custom logic.""" + +class NodeTest: + """Node that has business logic (violation).""" + + def __init__(self, container): + super().__init__(container) + + def process_data(self, data): + """This should be in a handler.""" + return data + + def validate_input(self, input_data): + """This should also be in a handler.""" + return True +''' + ) + return node_py + + +# --- parse_contract_topics --- + + +class TestParseContractTopics: + def test_extracts_topics(self, compliant_contract: Path) -> None: + publish, subscribe = parse_contract_topics(compliant_contract) + assert "onex.evt.test.computed.v1" in publish + assert "onex.evt.test.requested.v1" in subscribe + + def test_missing_file(self, tmp_path: Path) -> None: + publish, subscribe = parse_contract_topics(tmp_path / "nonexistent.yaml") + assert publish == [] + assert subscribe == [] + + def test_no_event_bus(self, tmp_node_dir: Path) -> None: + contract = tmp_node_dir / "contract.yaml" + contract.write_text("name: test\nnode_type: COMPUTE_GENERIC\n") + publish, subscribe = parse_contract_topics(contract) + assert publish == [] + assert subscribe == [] + + +# --- parse_contract_transports --- + + +class TestParseContractTransports: + def test_infers_kafka_from_topics(self, compliant_contract: Path) -> None: + transports = parse_contract_transports(compliant_contract) + assert "KAFKA" in transports + + def test_empty_for_no_transports(self, tmp_node_dir: Path) -> None: + contract = tmp_node_dir / "contract.yaml" + contract.write_text("name: test\nnode_type: COMPUTE_GENERIC\n") + transports = parse_contract_transports(contract) + assert transports == [] + + +# --- parse_contract_handler_routing --- + + +class TestParseContractHandlerRouting: + def test_extracts_routing(self, compliant_contract: Path) -> None: + entries = parse_contract_handler_routing(compliant_contract) + assert len(entries) == 1 + assert entries[0]["name"] == "HandlerTest" + assert entries[0]["module"] == "test_repo.nodes.node_test.handlers.handler_test" + + +# --- scan_handler_topics --- + + +class TestScanHandlerTopics: + def test_finds_hardcoded_topics(self, imperative_handler: Path) -> None: + topics = scan_handler_topics(imperative_handler) + assert "agent-actions" in topics + assert "onex.evt.bad.topic.v1" in topics + + def test_no_topics_in_compliant(self, compliant_handler: Path) -> None: + topics = scan_handler_topics(compliant_handler) + assert topics == [] + + def test_skips_docstrings(self, tmp_path: Path) -> None: + handler = tmp_path / "handler_docstring.py" + handler.write_text( + ''' +"""This handler uses agent-actions topic.""" + +class Handler: + """Processes onex.evt.test.computed.v1 events.""" + pass +''' + ) + topics = scan_handler_topics(handler) + assert topics == [] + + +# --- scan_handler_transports --- + + +class TestScanHandlerTransports: + def test_detects_http_transport(self, imperative_handler: Path) -> None: + transports = scan_handler_transports(imperative_handler) + assert "HTTP" in transports + + def test_no_transports_in_compliant(self, compliant_handler: Path) -> None: + transports = scan_handler_transports(compliant_handler) + assert transports == [] + + def test_detects_database_transport(self, tmp_path: Path) -> None: + handler = tmp_path / "handler_db.py" + handler.write_text( + """ +import asyncpg + +async def query(): + conn = await asyncpg.connect() +""" + ) + transports = scan_handler_transports(handler) + assert "DATABASE" in transports + + +# --- scan_node_py_logic --- + + +class TestScanNodePyLogic: + def test_clean_node(self, clean_node_py: Path) -> None: + methods = scan_node_py_logic(clean_node_py) + assert methods == [] + + def test_logic_in_node(self, logic_node_py: Path) -> None: + methods = scan_node_py_logic(logic_node_py) + assert "process_data" in methods + assert "validate_input" in methods + + def test_missing_file(self, tmp_path: Path) -> None: + methods = scan_node_py_logic(tmp_path / "nonexistent.py") + assert methods == [] + + +# --- cross_reference --- + + +class TestCrossReference: + def test_compliant_handler( + self, + tmp_node_dir: Path, + compliant_contract: Path, # noqa: ARG002 + compliant_handler: Path, # noqa: ARG002 + clean_node_py: Path, # noqa: ARG002 + ) -> None: + results = cross_reference(tmp_node_dir, "test_repo") + assert len(results) == 1 + assert results[0].verdict == EnumComplianceVerdict.COMPLIANT + assert results[0].violations == [] + + def test_imperative_handler( + self, + tmp_node_dir: Path, + compliant_contract: Path, # noqa: ARG002 + imperative_handler: Path, # noqa: ARG002 + clean_node_py: Path, # noqa: ARG002 + ) -> None: + results = cross_reference(tmp_node_dir, "test_repo") + violating = [ + r for r in results if r.handler_path.endswith("handler_imperative.py") + ] + assert len(violating) == 1 + r = violating[0] + assert r.verdict == EnumComplianceVerdict.IMPERATIVE + assert EnumComplianceViolation.HARDCODED_TOPIC in r.violations + assert EnumComplianceViolation.UNDECLARED_TRANSPORT in r.violations + + def test_missing_contract(self, tmp_node_dir: Path) -> None: + handler = tmp_node_dir / "handlers" / "handler_orphan.py" + handler.write_text("class HandlerOrphan: pass\n") + results = cross_reference(tmp_node_dir, "test_repo") + assert len(results) == 1 + assert results[0].verdict == EnumComplianceVerdict.MISSING_CONTRACT + + def test_logic_in_node_reported( + self, + tmp_node_dir: Path, + compliant_contract: Path, # noqa: ARG002 + compliant_handler: Path, # noqa: ARG002 + logic_node_py: Path, # noqa: ARG002 + ) -> None: + results = cross_reference(tmp_node_dir, "test_repo") + assert len(results) == 1 + assert EnumComplianceViolation.LOGIC_IN_NODE in results[0].violations + + def test_allowlisted_handler( + self, + tmp_node_dir: Path, + compliant_contract: Path, # noqa: ARG002 + imperative_handler: Path, # noqa: ARG002 + ) -> None: + handler_path = str( + (tmp_node_dir / "handlers" / "handler_imperative.py").relative_to( + tmp_node_dir.parent.parent.parent + ) + ) + results = cross_reference( + tmp_node_dir, + "test_repo", + allowlisted_paths=frozenset({handler_path}), + ) + imperative = [ + r for r in results if r.handler_path.endswith("handler_imperative.py") + ] + assert len(imperative) == 1 + assert imperative[0].verdict == EnumComplianceVerdict.ALLOWLISTED + assert imperative[0].allowlisted is True