From fea9e1cdb89c7585cb9c4124c6a661a5c1b305f7 Mon Sep 17 00:00:00 2001 From: ElliotSun Date: Sat, 29 Aug 2026 09:18:38 +1000 Subject: [PATCH 1/3] feat(cli): standardize CLI and CI outcomes around governance decisions (#30) - Add ProcessOutcome and CliExitCode vocabulary in semapact.interfaces.outcomes - Implement centralized outcome and exit code mapping functions - Update CLI top-level exception handling to map exceptions to standardized exit codes - Remove hardcoded sys.exit calls in command modules to route through central handler - Add comprehensive unit, matrix, and subprocess-level acceptance tests --- semapact/interfaces/__init__.py | 19 +- semapact/interfaces/cli.py | 24 +- semapact/interfaces/commands/export_cmd.py | 30 +- semapact/interfaces/commands/plan_cmd.py | 133 +++--- semapact/interfaces/outcomes.py | 106 +++++ tests/interfaces/test_cli_outcomes.py | 175 ++++++++ .../test_cli_subprocess_outcomes.py | 388 ++++++++++++++++++ tests/test_semapact_cli.py | 5 +- 8 files changed, 780 insertions(+), 100 deletions(-) create mode 100644 semapact/interfaces/outcomes.py create mode 100644 tests/interfaces/test_cli_outcomes.py create mode 100644 tests/interfaces/test_cli_subprocess_outcomes.py diff --git a/semapact/interfaces/__init__.py b/semapact/interfaces/__init__.py index 7ebe827..89635e0 100644 --- a/semapact/interfaces/__init__.py +++ b/semapact/interfaces/__init__.py @@ -1,3 +1,20 @@ from semapact.interfaces.cli import main +from semapact.interfaces.outcomes import ( + CliExitCode, + ProcessOutcome, + exit_code_from_exception, + exit_code_from_outcome, + outcome_from_exception, + outcome_from_gate_result, +) + +__all__ = [ + "CliExitCode", + "ProcessOutcome", + "exit_code_from_exception", + "exit_code_from_outcome", + "main", + "outcome_from_exception", + "outcome_from_gate_result", +] -__all__ = ["main"] diff --git a/semapact/interfaces/cli.py b/semapact/interfaces/cli.py index 765adfe..ce8eaf1 100644 --- a/semapact/interfaces/cli.py +++ b/semapact/interfaces/cli.py @@ -342,7 +342,8 @@ def main() -> int: except ImportError: import sys print("āŒ TUI requires the 'tui' extra. Install it via: pip install \"semapact[tui]\"", file=sys.stderr) - return 1 + from semapact.interfaces.outcomes import CliExitCode + return int(CliExitCode.RUNTIME_ERROR) from semapact.tui.app import SemaPactTUI app = SemaPactTUI() app.run() @@ -435,25 +436,34 @@ def main() -> int: print(json.dumps(payload, indent=2, sort_keys=True)) return 0 + from semapact.interfaces.outcomes import CliExitCode parser.error(f"Unknown command: {args.command}") - return 2 + return int(CliExitCode.VALIDATION_FAILED) except KeyboardInterrupt: return 130 except SystemExit as exc: - return exc.code if isinstance(exc.code, int) else 1 + from semapact.interfaces.outcomes import CliExitCode + return exc.code if isinstance(exc.code, int) else int(CliExitCode.RUNTIME_ERROR) except Exception as exc: - from semapact.exceptions import SemaPactError import logging - - if isinstance(exc, SemaPactError): + from semapact.exceptions import ( + GovernanceBlockedError, + GovernanceReviewRequiredError, + SemaPactError, + ) + from semapact.interfaces.outcomes import exit_code_from_exception + + if isinstance(exc, (GovernanceBlockedError, GovernanceReviewRequiredError)): + logging.getLogger("semapact").info("Governance decision: %s", exc) + elif isinstance(exc, SemaPactError): logging.getLogger("semapact").error("Fatal error: %s", exc) else: logging.getLogger("semapact").error( "Fatal error: %s", exc, exc_info=True ) print(f"āŒ {exc}", file=__import__("sys").stderr) - return 1 + return exit_code_from_exception(exc) if __name__ == "__main__": diff --git a/semapact/interfaces/commands/export_cmd.py b/semapact/interfaces/commands/export_cmd.py index 593c482..6c22de3 100644 --- a/semapact/interfaces/commands/export_cmd.py +++ b/semapact/interfaces/commands/export_cmd.py @@ -109,28 +109,22 @@ def run_export(args: argparse.Namespace) -> str: def run_export_ge(args: argparse.Namespace) -> str: - import sys - try: from semapact.quality.ge_exporter import GreatExpectationsExporter except ImportError as exc: if "great_expectations" in str(exc) or "great-expectations" in str(exc): - sys.exit( - "Error: The 'great_expectations' library is required to export GE suites.\n" + raise ImportError( + "The 'great_expectations' library is required to export GE suites.\n" "Please install it using: pip install \"semapact[quality]\" or pip install great_expectations" - ) + ) from exc raise - try: - output_path = GreatExpectationsExporter().export_to_path( - args.contract, - args.output, - schema_name=args.schema_name, - suite_name=args.suite_name, - engine=args.engine, - ) - return str(output_path) - except (RuntimeError, ImportError) as exc: - if "requires pyspark to be installed" in str(exc) or "pyspark" in str(exc): - sys.exit(str(exc)) - raise + output_path = GreatExpectationsExporter().export_to_path( + args.contract, + args.output, + schema_name=args.schema_name, + suite_name=args.suite_name, + engine=args.engine, + ) + return str(output_path) + diff --git a/semapact/interfaces/commands/plan_cmd.py b/semapact/interfaces/commands/plan_cmd.py index f67a23b..6c7ee59 100644 --- a/semapact/interfaces/commands/plan_cmd.py +++ b/semapact/interfaces/commands/plan_cmd.py @@ -19,83 +19,72 @@ def run_plan(args: argparse.Namespace) -> None: print(f"\nšŸ” Contract Analysis: {args.base}") - try: - import_args: dict[str, Any] = {} - if args.type in {"delta", "delta-table"}: - oauth_token = None - if args.source.startswith("abfss://") or "dfs.core.windows.net" in args.source: - oauth_token = _resolve_adls_oauth_token_from_config() - - table_uris = _parse_table_uris(args.tables) - if not table_uris: - from semapact.utils.storage_adapter import StorageAdapterFactory - adapter = StorageAdapterFactory.get_adapter(args.source) - try: - table_uris = adapter.discover_delta_tables(args.source, credential=oauth_token) - except Exception as e: - import logging - logging.getLogger("semapact").warning(f"Failed to auto-discover delta tables: {e}") - table_uris = [] + import_args: dict[str, Any] = {} + if args.type in {"delta", "delta-table"}: + oauth_token = None + if args.source.startswith("abfss://") or "dfs.core.windows.net" in args.source: + oauth_token = _resolve_adls_oauth_token_from_config() - if oauth_token: - import_args["oauth_bearer_token"] = oauth_token - if table_uris: - import_args["table_uris"] = table_uris - elif args.tables: - import_args["tables"] = args.tables + table_uris = _parse_table_uris(args.tables) + if not table_uris: + from semapact.utils.storage_adapter import StorageAdapterFactory + adapter = StorageAdapterFactory.get_adapter(args.source) + try: + table_uris = adapter.discover_delta_tables(args.source, credential=oauth_token) + except Exception as e: + import logging + logging.getLogger("semapact").warning(f"Failed to auto-discover delta tables: {e}") + table_uris = [] + + if oauth_token: + import_args["oauth_bearer_token"] = oauth_token + if table_uris: + import_args["table_uris"] = table_uris + elif args.tables: + import_args["tables"] = args.tables - # Import temporary contract from source - imported = pipeline.import_schema( - source_type=args.type, - source=args.source, - uc_workspace_url=args.workspace_url, - uc_token=args.token, - import_args=import_args if import_args else None, - ) + # Import temporary contract from source + imported = pipeline.import_schema( + source_type=args.type, + source=args.source, + uc_workspace_url=args.workspace_url, + uc_token=args.token, + import_args=import_args if import_args else None, + ) - # Load base contract (governed target) - base_contract = pipeline.loader.load(args.base) + # Load base contract (governed target) + base_contract = pipeline.loader.load(args.base) - # Merge them (to normalize and evaluate breaks) - merge_result = pipeline.merge_contract_updates( - imported, - base_contract, - context=change_context, - fail_on_conflict=False, - ) - merged = merge_result.contract + # Merge them (to normalize and evaluate breaks) + merge_result = pipeline.merge_contract_updates( + imported, + base_contract, + context=change_context, + fail_on_conflict=False, + ) + merged = merge_result.contract - # Evaluate decision & ANALYZE operation gate (always allowed for analysis) - decision = evaluate_governance_decision( - base_contract, - merged, - context=change_context, - merge_conflicts=merge_result.conflicts, - ) - gate_res = evaluate_governance_gate(decision, GovernanceOperation.ANALYZE) + # Evaluate decision & ANALYZE operation gate (always allowed for analysis) + decision = evaluate_governance_decision( + base_contract, + merged, + context=change_context, + merge_conflicts=merge_result.conflicts, + ) + gate_res = evaluate_governance_gate(decision, GovernanceOperation.ANALYZE) - if not decision.evidence.has_changes: - print("šŸŸ¢ No changes detected.") - else: - print(f"šŸ“Š Governance Decision: {decision.decision.value} (Gate: {gate_res.reason})") - for reason in decision.reasons: - print(f" ā€¢ [{reason.code}] {reason.path or 'root'}: {reason.message}") + if not decision.evidence.has_changes: + print("šŸŸ¢ No changes detected.") + else: + print(f"šŸ“Š Governance Decision: {decision.decision.value} (Gate: {gate_res.reason})") + for reason in decision.reasons: + print(f" ā€¢ [{reason.code}] {reason.path or 'root'}: {reason.message}") - bump = decision.required_version_bump.upper() - if bump == "NONE": - print("\nāœ… Action Required: No version bump needed.") - elif bump == "MINOR": - print(f"\nāš ļø Action Required: Additive changes require version bump {bump}.") - elif bump == "MAJOR": - print(f"\nāš ļø Action Required: Breaking changes require version bump {bump}.") + bump = decision.required_version_bump.upper() + if bump == "NONE": + print("\nāœ… Action Required: No version bump needed.") + elif bump == "MINOR": + print(f"\nāš ļø Action Required: Additive changes require version bump {bump}.") + elif bump == "MAJOR": + print(f"\nāš ļø Action Required: Breaking changes require version bump {bump}.") - except Exception as e: - from semapact.exceptions import SemaPactError - import logging - - if isinstance(e, SemaPactError): - logging.getLogger("semapact").error("Plan failed: %s", e) - else: - logging.getLogger("semapact").error("Plan failed: %s", e, exc_info=True) - print(f"āŒ Error during plan: {e}") - raise SystemExit(1) diff --git a/semapact/interfaces/outcomes.py b/semapact/interfaces/outcomes.py new file mode 100644 index 0000000..6f0b645 --- /dev/null +++ b/semapact/interfaces/outcomes.py @@ -0,0 +1,106 @@ +"""Standardized process outcome vocabulary and CLI exit code mapping. + +This module defines the authoritative semantic outcome vocabulary and shell exit +code adapter for SemaPact CLI and CI automation layers. +""" + +from __future__ import annotations + +from enum import Enum, IntEnum +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from semapact.governance.gate import GovernanceGateResult + + +logger = logging.getLogger("semapact") + + +class ProcessOutcome(str, Enum): + """Authoritative semantic process outcome vocabulary.""" + + SUCCESS = "SUCCESS" + VALIDATION_FAILED = "VALIDATION_FAILED" + GOVERNANCE_BLOCKED = "GOVERNANCE_BLOCKED" + REVIEW_REQUIRED = "REVIEW_REQUIRED" + RUNTIME_ERROR = "RUNTIME_ERROR" + + +class CliExitCode(IntEnum): + """Standardized shell process exit codes for SemaPact.""" + + SUCCESS = 0 + VALIDATION_FAILED = 2 + GOVERNANCE_BLOCKED = 3 + REVIEW_REQUIRED = 4 + RUNTIME_ERROR = 5 + + +_OUTCOME_TO_EXIT_CODE: dict[ProcessOutcome, CliExitCode] = { + ProcessOutcome.SUCCESS: CliExitCode.SUCCESS, + ProcessOutcome.VALIDATION_FAILED: CliExitCode.VALIDATION_FAILED, + ProcessOutcome.GOVERNANCE_BLOCKED: CliExitCode.GOVERNANCE_BLOCKED, + ProcessOutcome.REVIEW_REQUIRED: CliExitCode.REVIEW_REQUIRED, + ProcessOutcome.RUNTIME_ERROR: CliExitCode.RUNTIME_ERROR, +} + + +def exit_code_from_outcome(outcome: ProcessOutcome) -> CliExitCode: + """Map a semantic ProcessOutcome to its corresponding CliExitCode.""" + try: + return _OUTCOME_TO_EXIT_CODE[outcome] + except KeyError: + raise ValueError(f"Unsupported ProcessOutcome: {outcome!r}") from None + + +def outcome_from_gate_result(gate_result: GovernanceGateResult) -> ProcessOutcome: + """Map a GovernanceGateResult directly to its corresponding ProcessOutcome.""" + if gate_result.allowed: + return ProcessOutcome.SUCCESS + if gate_result.reason == "blocked": + return ProcessOutcome.GOVERNANCE_BLOCKED + if gate_result.reason == "review_required": + return ProcessOutcome.REVIEW_REQUIRED + raise ValueError(f"Unsupported GovernanceGateResult reason: {gate_result.reason!r}") + + +def outcome_from_exception(exc: BaseException) -> ProcessOutcome: + """Determine the semantic ProcessOutcome for a given exception.""" + from semapact.exceptions import ( + GovernanceBlockedError, + GovernanceReviewRequiredError, + ValidationError, + ) + + if isinstance(exc, GovernanceBlockedError): + return ProcessOutcome.GOVERNANCE_BLOCKED + if isinstance(exc, GovernanceReviewRequiredError): + return ProcessOutcome.REVIEW_REQUIRED + if isinstance(exc, ValidationError): + return ProcessOutcome.VALIDATION_FAILED + + # Check for Pydantic / jsonschema validation errors + try: + from pydantic import ValidationError as PydanticValidationError + if isinstance(exc, PydanticValidationError): + return ProcessOutcome.VALIDATION_FAILED + except ImportError: + pass + + return ProcessOutcome.RUNTIME_ERROR + + +def exit_code_from_exception(exc: BaseException) -> int: + """Resolve an exit code from an exception. + + Preserves explicit SystemExit and KeyboardInterrupt codes, while mapping all other + domain, validation, governance, or runtime exceptions to standardized CliExitCodes. + """ + if isinstance(exc, KeyboardInterrupt): + return 130 + if isinstance(exc, SystemExit): + return exc.code if isinstance(exc.code, int) else CliExitCode.RUNTIME_ERROR + + outcome = outcome_from_exception(exc) + return int(exit_code_from_outcome(outcome)) diff --git a/tests/interfaces/test_cli_outcomes.py b/tests/interfaces/test_cli_outcomes.py new file mode 100644 index 0000000..7196b94 --- /dev/null +++ b/tests/interfaces/test_cli_outcomes.py @@ -0,0 +1,175 @@ +"""Unit and contract-level tests for standardized process outcomes and exit codes.""" + +from __future__ import annotations + +import pytest +from pydantic import BaseModel, Field + +from semapact.change_context import ChangeContext +from semapact.core.release import RequiredBump +from semapact.exceptions import ( + GovernanceBlockedError, + GovernanceReviewRequiredError, + StorageError, + ValidationError, +) +from semapact.governance import ( + DecisionResult, + GovernanceDecision, + GovernanceGateResult, + GovernanceOperation, + PolicyOutcome, + ValidationOutcome, + evaluate_governance_gate, +) +from semapact.governance.models import ChangeEvidence +from semapact.interfaces.outcomes import ( + CliExitCode, + ProcessOutcome, + exit_code_from_exception, + exit_code_from_outcome, + outcome_from_exception, + outcome_from_gate_result, +) + + +def _make_dummy_decision( + decision: DecisionResult, + bump: RequiredBump = "none", + breaking: bool = False, +) -> GovernanceDecision: + return GovernanceDecision( + decision_id="test-dec-1", + decision=decision, + contract_id="test-contract", + context=ChangeContext(effective_date="2026-08-29"), + breaking=breaking, + required_version_bump=bump, + validation=ValidationOutcome(valid=True), + policy=PolicyOutcome(valid=True), + evidence=ChangeEvidence(has_changes=True), + ) + + +def test_process_outcome_and_exit_code_mappings(): + """Verify exact 1-to-1 semantic vocabulary to shell exit code mapping.""" + expected_mappings = { + ProcessOutcome.SUCCESS: CliExitCode.SUCCESS, + ProcessOutcome.VALIDATION_FAILED: CliExitCode.VALIDATION_FAILED, + ProcessOutcome.GOVERNANCE_BLOCKED: CliExitCode.GOVERNANCE_BLOCKED, + ProcessOutcome.REVIEW_REQUIRED: CliExitCode.REVIEW_REQUIRED, + ProcessOutcome.RUNTIME_ERROR: CliExitCode.RUNTIME_ERROR, + } + + for outcome, expected_code in expected_mappings.items(): + assert exit_code_from_outcome(outcome) == expected_code + assert int(expected_code) in (0, 2, 3, 4, 5) + + assert int(CliExitCode.SUCCESS) == 0 + assert int(CliExitCode.VALIDATION_FAILED) == 2 + assert int(CliExitCode.GOVERNANCE_BLOCKED) == 3 + assert int(CliExitCode.REVIEW_REQUIRED) == 4 + assert int(CliExitCode.RUNTIME_ERROR) == 5 + + +def test_outcome_from_gate_result(): + """Verify GovernanceGateResult maps directly to the appropriate ProcessOutcome.""" + allowed_res = GovernanceGateResult( + allowed=True, reason="allowed", decision_id="d1" + ) + assert outcome_from_gate_result(allowed_res) == ProcessOutcome.SUCCESS + + blocked_res = GovernanceGateResult( + allowed=False, reason="blocked", decision_id="d2" + ) + assert outcome_from_gate_result(blocked_res) == ProcessOutcome.GOVERNANCE_BLOCKED + + review_res = GovernanceGateResult( + allowed=False, reason="review_required", decision_id="d3" + ) + assert outcome_from_gate_result(review_res) == ProcessOutcome.REVIEW_REQUIRED + + +def test_outcome_from_exception_and_exit_code(): + """Verify domain, validation, and runtime exceptions map to standardized outcomes and exit codes.""" + blocked_exc = GovernanceBlockedError("Change is blocked") + assert outcome_from_exception(blocked_exc) == ProcessOutcome.GOVERNANCE_BLOCKED + assert exit_code_from_exception(blocked_exc) == 3 + + review_exc = GovernanceReviewRequiredError("Manual review needed") + assert outcome_from_exception(review_exc) == ProcessOutcome.REVIEW_REQUIRED + assert exit_code_from_exception(review_exc) == 4 + + val_exc = ValidationError("Invalid ODCS syntax") + assert outcome_from_exception(val_exc) == ProcessOutcome.VALIDATION_FAILED + assert exit_code_from_exception(val_exc) == 2 + + # Pydantic validation error + class DummyModel(BaseModel): + num: int = Field(strict=True) + + try: + DummyModel.model_validate({"num": "not_an_int"}) + except Exception as pydantic_exc: + assert outcome_from_exception(pydantic_exc) == ProcessOutcome.VALIDATION_FAILED + assert exit_code_from_exception(pydantic_exc) == 2 + + # Runtime and infrastructure exceptions + runtime_exc = RuntimeError("Database unreachable") + assert outcome_from_exception(runtime_exc) == ProcessOutcome.RUNTIME_ERROR + assert exit_code_from_exception(runtime_exc) == 5 + + storage_exc = StorageError("ADLS token expired") + assert outcome_from_exception(storage_exc) == ProcessOutcome.RUNTIME_ERROR + assert exit_code_from_exception(storage_exc) == 5 + + # SystemExit and KeyboardInterrupt + assert exit_code_from_exception(KeyboardInterrupt()) == 130 + assert exit_code_from_exception(SystemExit(0)) == 0 + assert exit_code_from_exception(SystemExit(2)) == 2 + assert exit_code_from_exception(SystemExit("string error")) == 5 + + +@pytest.mark.parametrize( + "operation,decision_result,expected_outcome,expected_exit_code", + [ + # ANALYZE: Never fails because of governance decision (reports decision transparently) + (GovernanceOperation.ANALYZE, DecisionResult.ALLOW, ProcessOutcome.SUCCESS, 0), + (GovernanceOperation.ANALYZE, DecisionResult.REVIEW, ProcessOutcome.SUCCESS, 0), + (GovernanceOperation.ANALYZE, DecisionResult.BLOCK, ProcessOutcome.SUCCESS, 0), + # PROPOSE: ALLOW and REVIEW pass; BLOCK is blocked + (GovernanceOperation.PROPOSE, DecisionResult.ALLOW, ProcessOutcome.SUCCESS, 0), + (GovernanceOperation.PROPOSE, DecisionResult.REVIEW, ProcessOutcome.SUCCESS, 0), + (GovernanceOperation.PROPOSE, DecisionResult.BLOCK, ProcessOutcome.GOVERNANCE_BLOCKED, 3), + # APPLY: ALLOW passes; REVIEW requires review; BLOCK is blocked + (GovernanceOperation.APPLY, DecisionResult.ALLOW, ProcessOutcome.SUCCESS, 0), + (GovernanceOperation.APPLY, DecisionResult.REVIEW, ProcessOutcome.REVIEW_REQUIRED, 4), + (GovernanceOperation.APPLY, DecisionResult.BLOCK, ProcessOutcome.GOVERNANCE_BLOCKED, 3), + # PUBLISH: Same as APPLY + (GovernanceOperation.PUBLISH, DecisionResult.ALLOW, ProcessOutcome.SUCCESS, 0), + (GovernanceOperation.PUBLISH, DecisionResult.REVIEW, ProcessOutcome.REVIEW_REQUIRED, 4), + (GovernanceOperation.PUBLISH, DecisionResult.BLOCK, ProcessOutcome.GOVERNANCE_BLOCKED, 3), + # CI: Same as APPLY + (GovernanceOperation.CI, DecisionResult.ALLOW, ProcessOutcome.SUCCESS, 0), + (GovernanceOperation.CI, DecisionResult.REVIEW, ProcessOutcome.REVIEW_REQUIRED, 4), + (GovernanceOperation.CI, DecisionResult.BLOCK, ProcessOutcome.GOVERNANCE_BLOCKED, 3), + ], +) +def test_governance_gate_to_process_outcome_matrix( + operation: GovernanceOperation, + decision_result: DecisionResult, + expected_outcome: ProcessOutcome, + expected_exit_code: int, +): + """Verify that all governance operations and decisions map predictably through the gate.""" + decision = _make_dummy_decision( + decision_result, + bump="minor" if decision_result == DecisionResult.REVIEW else ("major" if decision_result == DecisionResult.BLOCK else "none"), + breaking=(decision_result == DecisionResult.BLOCK), + ) + gate_res = evaluate_governance_gate(decision, operation) + outcome = outcome_from_gate_result(gate_res) + exit_code = exit_code_from_outcome(outcome) + + assert outcome == expected_outcome + assert int(exit_code) == expected_exit_code diff --git a/tests/interfaces/test_cli_subprocess_outcomes.py b/tests/interfaces/test_cli_subprocess_outcomes.py new file mode 100644 index 0000000..be9df05 --- /dev/null +++ b/tests/interfaces/test_cli_subprocess_outcomes.py @@ -0,0 +1,388 @@ +"""Subprocess-level acceptance tests for standardized CLI process exit outcomes.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest +import yaml + + +def _run_cli(*args: str) -> subprocess.CompletedProcess[str]: + """Execute the SemaPact CLI in a subprocess.""" + cmd = [sys.executable, "-m", "semapact.interfaces.cli", *args] + return subprocess.run( + cmd, + capture_output=True, + text=True, + check=False, + ) + + +@pytest.fixture +def active_contract_yaml(tmp_path: Path) -> Path: + contract = { + "apiVersion": "v3.1.0", + "kind": "DataContract", + "id": "urn:datacontract:orders", + "name": "orders", + "version": "1.0.0", + "status": "active", + "schema": [ + { + "name": "orders", + "properties": [ + { + "name": "order_id", + "type": "string", + }, + { + "name": "amount", + "type": "number", + }, + ], + } + ], + } + path = tmp_path / "active_contract.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + return path + + +@pytest.fixture +def review_candidate_yaml(tmp_path: Path) -> Path: + """Additive change (new column) requiring MINOR bump.""" + contract = { + "apiVersion": "v3.1.0", + "kind": "DataContract", + "id": "urn:datacontract:orders", + "name": "orders", + "version": "1.0.0", + "status": "active", + "schema": [ + { + "name": "orders", + "properties": [ + { + "name": "order_id", + "type": "string", + }, + { + "name": "amount", + "type": "number", + }, + { + "name": "customer_id", + "type": "string", + }, + ], + } + ], + } + path = tmp_path / "review_candidate.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + return path + + +@pytest.fixture +def retired_contract_yaml(tmp_path: Path) -> Path: + contract = { + "apiVersion": "v3.1.0", + "kind": "DataContract", + "id": "urn:datacontract:orders", + "name": "orders", + "version": "1.0.0", + "status": "retired", + "schema": [ + { + "name": "orders", + "properties": [ + { + "name": "order_id", + "type": "string", + } + ], + } + ], + } + path = tmp_path / "retired_contract.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + return path + + +@pytest.fixture +def invalid_contract_yaml(tmp_path: Path) -> Path: + """Contract that violates ODCS schema structure (raises Pydantic ValidationError).""" + contract = { + "apiVersion": "v3.1.0", + "kind": "DataContract", + "id": "urn:datacontract:orders", + "name": "orders", + "version": "1.0.0", + "status": "active", + "schema": "not_a_list_violates_odcs_model", + } + path = tmp_path / "invalid_contract.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + return path + + +# ============================================================================== +# 1. ANALYZE Commands: Never fail because of governance decision +# ============================================================================== + +def test_subprocess_analyze_release_classify_allow(active_contract_yaml: Path): + res = _run_cli( + "release", + "classify", + "--base", + str(active_contract_yaml), + "--candidate", + str(active_contract_yaml), + "--effective-date", + "2026-08-29", + ) + assert res.returncode == 0 + payload = json.loads(res.stdout) + assert payload["hasChanges"] is False + assert payload["requiredBump"] == "none" + assert "exitCode" not in payload + assert "exitCode" not in payload.get("governanceDecision", {}) + + +def test_subprocess_analyze_release_classify_review( + active_contract_yaml: Path, review_candidate_yaml: Path +): + res = _run_cli( + "release", + "classify", + "--base", + str(active_contract_yaml), + "--candidate", + str(review_candidate_yaml), + "--effective-date", + "2026-08-29", + ) + assert res.returncode == 0 + payload = json.loads(res.stdout) + assert payload["hasChanges"] is True + assert payload["requiredBump"] == "minor" + assert "exitCode" not in payload + + +def test_subprocess_analyze_release_classify_block( + retired_contract_yaml: Path, review_candidate_yaml: Path +): + # ANALYZE on retired base contract still exits with code 0 and reports BLOCK + res = _run_cli( + "release", + "classify", + "--base", + str(retired_contract_yaml), + "--candidate", + str(review_candidate_yaml), + "--effective-date", + "2026-08-29", + ) + assert res.returncode == 0 + payload = json.loads(res.stdout) + assert payload["hasChanges"] is True + assert payload["governanceDecision"]["decision"] == "BLOCK" + assert "exitCode" not in payload + + +# ============================================================================== +# 2. PROPOSE Commands: ALLOW & REVIEW -> 0, BLOCK -> 3 (GOVERNANCE_BLOCKED) +# ============================================================================== + +def test_subprocess_propose_release_prepare_review( + active_contract_yaml: Path, review_candidate_yaml: Path, tmp_path: Path +): + out_yaml = tmp_path / "prepared.yaml" + res = _run_cli( + "release", + "prepare", + "--base", + str(active_contract_yaml), + "--candidate", + str(review_candidate_yaml), + "--release-tag", + "v1.1.0", + "--output", + str(out_yaml), + "--effective-date", + "2026-08-29", + ) + assert res.returncode == 0 + payload = json.loads(res.stdout) + assert payload["actualBump"] == "minor" + assert "exitCode" not in payload + + +def test_subprocess_propose_release_prepare_no_bump_runtime_error( + active_contract_yaml: Path, tmp_path: Path +): + out_yaml = tmp_path / "prepared.yaml" + res = _run_cli( + "release", + "prepare", + "--base", + str(active_contract_yaml), + "--candidate", + str(active_contract_yaml), + "--release-tag", + "v1.0.0", + "--output", + str(out_yaml), + "--effective-date", + "2026-08-29", + ) + # Attempting to prepare a release candidate when no bump is required returns RUNTIME_ERROR (5) + assert res.returncode == 5 + assert "Contract changes do not require a release version bump" in res.stderr + + +def test_subprocess_propose_release_prepare_block( + retired_contract_yaml: Path, review_candidate_yaml: Path, tmp_path: Path +): + out_yaml = tmp_path / "prepared.yaml" + res = _run_cli( + "release", + "prepare", + "--base", + str(retired_contract_yaml), + "--candidate", + str(review_candidate_yaml), + "--release-tag", + "v2.0.0", + "--output", + str(out_yaml), + "--effective-date", + "2026-08-29", + ) + # PROPOSE with BLOCK decision must exit with 3 (GOVERNANCE_BLOCKED) + assert res.returncode == 3 + assert "Governance decision BLOCKED" in res.stderr + assert "Traceback (most recent call last)" not in res.stderr + + +def test_subprocess_propose_merge_block( + retired_contract_yaml: Path, review_candidate_yaml: Path, tmp_path: Path +): + out_yaml = tmp_path / "merged.yaml" + res = _run_cli( + "merge", + "--base", + str(review_candidate_yaml), + "--business", + str(retired_contract_yaml), + "--output", + str(out_yaml), + "--effective-date", + "2026-08-29", + ) + assert res.returncode == 3 + assert "Governance decision BLOCKED" in res.stderr + + +# ============================================================================== +# 3. APPLY Commands: ALLOW -> 0, REVIEW -> 4 (REVIEW_REQUIRED), BLOCK -> 3 +# ============================================================================== + +def test_subprocess_apply_lifecycle_promote_review_required( + active_contract_yaml: Path, tmp_path: Path +): + out_yaml = tmp_path / "promoted.yaml" + res = _run_cli( + "lifecycle", + "promote", + "--contract", + str(active_contract_yaml), + "--schema", + "orders", + "--property", + "order_id", + "--output", + str(out_yaml), + "--effective-date", + "2026-08-29", + ) + # Applying lifecycle mutation produces a REVIEW decision on additive changes, which requires review (4) + assert res.returncode == 4 + assert "Governance decision REVIEW required" in res.stderr + assert "Traceback (most recent call last)" not in res.stderr + + +def test_subprocess_apply_lifecycle_deprecate_block_on_retired( + retired_contract_yaml: Path, tmp_path: Path +): + out_yaml = tmp_path / "deprecated.yaml" + res = _run_cli( + "lifecycle", + "deprecate", + "--contract", + str(retired_contract_yaml), + "--schema", + "orders", + "--property", + "order_id", + "--output", + str(out_yaml), + "--effective-date", + "2026-08-29", + ) + # Mutating retired contract is blocked (3) + assert res.returncode == 3 + assert "Governance decision BLOCKED" in res.stderr + + +# ============================================================================== +# 4. VALIDATION_FAILED (Exit code 2) +# ============================================================================== + +def test_subprocess_validation_failed_on_invalid_arguments(): + res = _run_cli("unknown-command") + assert res.returncode == 2 + + res_missing_args = _run_cli("release", "classify") + assert res_missing_args.returncode == 2 + + +def test_subprocess_validation_failed_on_invalid_contract( + invalid_contract_yaml: Path, active_contract_yaml: Path +): + res = _run_cli( + "release", + "classify", + "--base", + str(invalid_contract_yaml), + "--candidate", + str(active_contract_yaml), + "--effective-date", + "2026-08-29", + ) + assert res.returncode == 2 + assert "āŒ" in res.stderr + + + +# ============================================================================== +# 5. RUNTIME_ERROR (Exit code 5) +# ============================================================================== + +def test_subprocess_runtime_error_on_missing_file(active_contract_yaml: Path): + res = _run_cli( + "release", + "classify", + "--base", + "/nonexistent/path/contract.yaml", + "--candidate", + str(active_contract_yaml), + "--effective-date", + "2026-08-29", + ) + # File not found error is a runtime execution failure + assert res.returncode in (2, 5) diff --git a/tests/test_semapact_cli.py b/tests/test_semapact_cli.py index f47cd5f..4ffa9e5 100644 --- a/tests/test_semapact_cli.py +++ b/tests/test_semapact_cli.py @@ -23,9 +23,10 @@ def export_to_path(self, *args, **kwargs): "out.json", ] with patch("sys.argv", test_args): - # The new CLI top-level error handler catches RuntimeErrors and returns 1 + # The standardized CLI top-level error handler catches RuntimeErrors and returns 5 (RUNTIME_ERROR) exit_code = main() - assert exit_code == 1 + assert exit_code == 5 + def test_cli_export_sql_databricks_location(tmp_path): From b2b760e3e1bee5fd5dbb63c9ac808bb1039a2a0d Mon Sep 17 00:00:00 2001 From: ElliotSun Date: Sat, 29 Aug 2026 09:28:04 +1000 Subject: [PATCH 2/3] fix(cli): refine exception taxonomy for release validation and missing files (#30) - Add ReleaseValidationError(ValidationError) for domain release precondition failures - Raise ReleaseValidationError in apply_release_candidate when no bump or insufficient bump (exit code 2) - Wrap raw Pydantic and ODCS structure errors in loader.py as SemaPact ValidationError - Keep outcome_from_exception strictly focused on SemaPact exception taxonomy - Enforce exact exit code 5 (RUNTIME_ERROR) assertion for missing files --- semapact/core/loader.py | 17 +++++++++++++---- semapact/core/release.py | 9 +++++++-- semapact/exceptions.py | 10 +++++++++- semapact/interfaces/outcomes.py | 9 +-------- tests/interfaces/test_cli_outcomes.py | 15 ++++++--------- .../interfaces/test_cli_subprocess_outcomes.py | 10 ++++++---- 6 files changed, 42 insertions(+), 28 deletions(-) diff --git a/semapact/core/loader.py b/semapact/core/loader.py index f0656fd..795db1e 100644 --- a/semapact/core/loader.py +++ b/semapact/core/loader.py @@ -16,7 +16,8 @@ from open_data_contract_standard.model import OpenDataContractStandard from semapact.core.config import config_manager -from semapact.exceptions import StorageError +from semapact.exceptions import StorageError, ValidationError + LOGGER = logging.getLogger(__name__) DEFAULT_SPARKUTILS_MAX_BYTES = int( @@ -71,13 +72,21 @@ def load_contract( ) payload = yaml.safe_load(contract_text) if not isinstance(payload, dict): - raise ValueError("Contract YAML must deserialize into a mapping object") + raise ValidationError("Contract YAML must deserialize into a mapping object") spec_version = payload.get("dataContractSpecification") or payload.get("apiVersion", "") if not (str(spec_version).startswith("3.") or str(spec_version).startswith("v3.")): - raise ValueError(f"SemaPact requires ODCS version 3.x, found: {spec_version}") + raise ValidationError(f"SemaPact requires ODCS version 3.x, found: {spec_version}") - return OpenDataContractStandard.model_validate(payload) + try: + return OpenDataContractStandard.model_validate(payload) + except Exception as exc: + from pydantic import ValidationError as PydanticValidationError + + if isinstance(exc, PydanticValidationError): + raise ValidationError(f"ODCS contract validation failed: {exc}") from exc + raise + def read_contract_text( diff --git a/semapact/core/release.py b/semapact/core/release.py index 1127eb3..7d66a19 100644 --- a/semapact/core/release.py +++ b/semapact/core/release.py @@ -217,16 +217,21 @@ def apply_release_candidate( ) if required_bump == "none": - raise ValueError("Contract changes do not require a release version bump") + from semapact.exceptions import ReleaseValidationError + + raise ReleaseValidationError("Contract changes do not require a release version bump") target_version = parse_release_tag_version(release_tag) actual_bump = classify_version_bump(str(base_model.version or ""), target_version) if VERSION_RANK[actual_bump] < VERSION_RANK[required_bump]: - raise ValueError( + from semapact.exceptions import ReleaseValidationError + + raise ReleaseValidationError( f"Release tag '{release_tag}' applies a {actual_bump} bump, but contract requires at least a " f"{required_bump} bump" ) + promoted = candidate_model.model_copy(deep=True) promoted.version = target_version return PromotionResult( diff --git a/semapact/exceptions.py b/semapact/exceptions.py index c9f1884..5b80f18 100644 --- a/semapact/exceptions.py +++ b/semapact/exceptions.py @@ -20,12 +20,20 @@ class SemaPactError(Exception): pass -class ValidationError(SemaPactError): +class ValidationError(SemaPactError, ValueError): """Raised when a contract fails governance or structure validation.""" pass + +class ReleaseValidationError(ValidationError): + """Raised when release candidate validation fails (e.g. insufficient version bump or no changes to release).""" + + pass + + + class MergeConflictError(SemaPactError): """Raised by the merge engine when business and technical metadata fatally conflict.""" diff --git a/semapact/interfaces/outcomes.py b/semapact/interfaces/outcomes.py index 6f0b645..0959fca 100644 --- a/semapact/interfaces/outcomes.py +++ b/semapact/interfaces/outcomes.py @@ -80,17 +80,10 @@ def outcome_from_exception(exc: BaseException) -> ProcessOutcome: if isinstance(exc, ValidationError): return ProcessOutcome.VALIDATION_FAILED - # Check for Pydantic / jsonschema validation errors - try: - from pydantic import ValidationError as PydanticValidationError - if isinstance(exc, PydanticValidationError): - return ProcessOutcome.VALIDATION_FAILED - except ImportError: - pass - return ProcessOutcome.RUNTIME_ERROR + def exit_code_from_exception(exc: BaseException) -> int: """Resolve an exit code from an exception. diff --git a/tests/interfaces/test_cli_outcomes.py b/tests/interfaces/test_cli_outcomes.py index 7196b94..2c0eb5a 100644 --- a/tests/interfaces/test_cli_outcomes.py +++ b/tests/interfaces/test_cli_outcomes.py @@ -3,7 +3,7 @@ from __future__ import annotations import pytest -from pydantic import BaseModel, Field + from semapact.change_context import ChangeContext from semapact.core.release import RequiredBump @@ -104,15 +104,12 @@ def test_outcome_from_exception_and_exit_code(): assert outcome_from_exception(val_exc) == ProcessOutcome.VALIDATION_FAILED assert exit_code_from_exception(val_exc) == 2 - # Pydantic validation error - class DummyModel(BaseModel): - num: int = Field(strict=True) + from semapact.exceptions import ReleaseValidationError + + release_val_exc = ReleaseValidationError("No version bump required") + assert outcome_from_exception(release_val_exc) == ProcessOutcome.VALIDATION_FAILED + assert exit_code_from_exception(release_val_exc) == 2 - try: - DummyModel.model_validate({"num": "not_an_int"}) - except Exception as pydantic_exc: - assert outcome_from_exception(pydantic_exc) == ProcessOutcome.VALIDATION_FAILED - assert exit_code_from_exception(pydantic_exc) == 2 # Runtime and infrastructure exceptions runtime_exc = RuntimeError("Database unreachable") diff --git a/tests/interfaces/test_cli_subprocess_outcomes.py b/tests/interfaces/test_cli_subprocess_outcomes.py index be9df05..48ebce1 100644 --- a/tests/interfaces/test_cli_subprocess_outcomes.py +++ b/tests/interfaces/test_cli_subprocess_outcomes.py @@ -222,7 +222,7 @@ def test_subprocess_propose_release_prepare_review( assert "exitCode" not in payload -def test_subprocess_propose_release_prepare_no_bump_runtime_error( +def test_subprocess_propose_release_prepare_no_bump_validation_failed( active_contract_yaml: Path, tmp_path: Path ): out_yaml = tmp_path / "prepared.yaml" @@ -240,11 +240,12 @@ def test_subprocess_propose_release_prepare_no_bump_runtime_error( "--effective-date", "2026-08-29", ) - # Attempting to prepare a release candidate when no bump is required returns RUNTIME_ERROR (5) - assert res.returncode == 5 + # Attempting to prepare a release candidate when no bump is required returns VALIDATION_FAILED (2) + assert res.returncode == 2 assert "Contract changes do not require a release version bump" in res.stderr + def test_subprocess_propose_release_prepare_block( retired_contract_yaml: Path, review_candidate_yaml: Path, tmp_path: Path ): @@ -385,4 +386,5 @@ def test_subprocess_runtime_error_on_missing_file(active_contract_yaml: Path): "2026-08-29", ) # File not found error is a runtime execution failure - assert res.returncode in (2, 5) + assert res.returncode == 5 + From 4393eae30b47f484362784dbf974c2e2d0c10b61 Mon Sep 17 00:00:00 2001 From: Elliot Sun Date: Sat, 29 Aug 2026 09:58:19 +1000 Subject: [PATCH 3/3] Potential fix for pull request finding 'CodeQL / Incomplete URL substring sanitization' Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- semapact/interfaces/commands/plan_cmd.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/semapact/interfaces/commands/plan_cmd.py b/semapact/interfaces/commands/plan_cmd.py index 6c7ee59..616010e 100644 --- a/semapact/interfaces/commands/plan_cmd.py +++ b/semapact/interfaces/commands/plan_cmd.py @@ -1,4 +1,5 @@ import argparse +from urllib.parse import urlparse from typing import Any from semapact.governance import ( GovernanceOperation, @@ -22,9 +23,16 @@ def run_plan(args: argparse.Namespace) -> None: import_args: dict[str, Any] = {} if args.type in {"delta", "delta-table"}: oauth_token = None - if args.source.startswith("abfss://") or "dfs.core.windows.net" in args.source: + source_is_abfss = args.source.startswith("abfss://") + parsed_source = urlparse(args.source) + source_host = parsed_source.hostname or "" + source_is_adls_host = ( + source_host == "dfs.core.windows.net" + or source_host.endswith(".dfs.core.windows.net") + ) + if source_is_abfss or source_is_adls_host: oauth_token = _resolve_adls_oauth_token_from_config() - + table_uris = _parse_table_uris(args.tables) if not table_uris: from semapact.utils.storage_adapter import StorageAdapterFactory