diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index b9847fe..40baccd 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -374,7 +374,7 @@ - During migration, distinguish the designated compatibility implementation from the production-consumer scope; reserve repository-wide zero for post-deletion validation. - Done when an adversarial syntax matrix reports the exact forbidden sites and `try`/`finally`, comprehension, assignment, and generic-call alias shapes cannot turn an explicit forbidden reference into a false zero. - _Requirements: 14.3, 15.3_ -- [ ] 10.4 Migrate or retire remaining registered production source-tree consumers +- [x] 10.4 Migrate or retire remaining registered production source-tree consumers - Move each registered non-benchmark production reader and writer to the canonical v2 contract or retire it within its owning boundary. - Update the matching stable inventory entry to `v2_migrated` or `retired` in the same change without removing or rekeying it. - Keep authorization analysis, model execution, proof execution, and other adjacent behavior in their existing owner boundaries while changing only their contract seam. diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index d0dd98c..556f17f 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -9,8 +9,6 @@ from dataclasses import asdict, dataclass, field, replace from pathlib import Path from typing import Any -from openultrasast.core.legacy_v1 import Coverage as CoreCoverage - from .. import mapping, source from ..preprocess import FileTarget, count_loc, detect_language, detect_tags from ..semantic.cfg import ControlContext, FunctionCfg @@ -41,12 +39,6 @@ from ..semantic.obligations.trace import AbstractTraceState, build_trace from ..semantic.obligations.witness import match_witness from .producers import Context, _safe_path -_CORE_RESOLUTION_KEYS = { - "complete_calls": "complete", - "incomplete_calls": "incomplete", - "unresolved_calls": "unresolved", -} - @dataclass(frozen=True) class Coverage: @@ -66,27 +58,6 @@ class Coverage: # resolution has no bare-name fallback; every call lands in exactly one bucket. resolution_counts: dict[str, int] = field(default_factory=dict) - def to_core(self) -> CoreCoverage: - """Native conformance to the detection-core coverage contract (detection-core task 8.1). - - Unknown resolution bucket keys fail loud rather than under-counting calls. - """ - unknown = sorted(key for key in self.resolution_counts if key not in _CORE_RESOLUTION_KEYS) - if unknown: - raise ValueError(f"unknown resolution bucket keys, refusing to under-count calls: {unknown}") - buckets = {name: 0 for name in ("complete", "incomplete", "unresolved")} - for key, value in self.resolution_counts.items(): - buckets[_CORE_RESOLUTION_KEYS[key]] += int(value) - return CoreCoverage( - entry_points=int(self.entry_points), - frameworks_recognised=dict(self.frameworks_recognised), - calls=sum(buckets.values()), - complete=buckets["complete"], - incomplete=buckets["incomplete"], - unresolved=buckets["unresolved"], - unchecked_reasons=dict(self.unchecked_by_reason), - ) - @dataclass(frozen=True) class ProverResult: diff --git a/src/openultrasast/semantic/obligations/core_adapter.py b/src/openultrasast/semantic/obligations/core_adapter.py index 692cffa..488751a 100644 --- a/src/openultrasast/semantic/obligations/core_adapter.py +++ b/src/openultrasast/semantic/obligations/core_adapter.py @@ -2,15 +2,8 @@ from __future__ import annotations -from collections.abc import Mapping - from openultrasast.core import evidence, family, semantics from openultrasast.core import provenance as core_provenance -from openultrasast.core.trace import Guard as LegacyGuard -from openultrasast.core.trace import Hop as LegacyHop -from openultrasast.core.trace import TraceRecord -from openultrasast.core.trace import Witness as LegacyWitness -from openultrasast.core.verdict import Judgment from .abstract_domain import EvidenceRelation, PairJudgment from .trace import Trace @@ -169,46 +162,4 @@ def _deterministic_provenance() -> core_provenance.DecisionProvenance: return core_provenance.DecisionProvenance(None, None, None, None, None, None) -def to_legacy_trace_record( - trace: Trace, - *, - entry: str, - digests: Mapping[str, str], - complete: bool = False, -) -> TraceRecord: - """Preserve the frozen v1 conversion while its writers remain parked.""" - state = trace.abstract_state - if state is None: - judgment = Judgment.INDETERMINATE - reasons: tuple[str, ...] = ("abstract_state_unset",) - witnesses: tuple[LegacyWitness, ...] = () - resource = authority = "unknown" - else: - judgment = Judgment(state.path_judgment.judgment.value) - completeness = state.path_judgment.completeness - reasons = tuple(completeness.reasons) or (("incomplete",) if not completeness.is_complete else ()) - witnesses = tuple( - LegacyWitness(kind="witness", provenance=item, relation=match.relation.value) - for match in state.path_judgment.matches - for item in match.evidence - ) - resource = ",".join(state.resource.values) or "unknown" - authority = ",".join(state.authority.values) or "unknown" - hops = tuple(LegacyHop(function=item.function, call_line=item.call_line, complete=complete) for item in trace.hops) - hop_guards = tuple(guard for hop in trace.hops for guard in hop.guards_before_call) - guards = tuple(LegacyGuard(kind=item.kind, effect=item.effect) for item in (*hop_guards, *trace.guards_after)) - return TraceRecord( - entry=entry, - hops=hops, - operation_kind=trace.operation_kind, - resource=resource, - authority=authority, - witnesses=witnesses, - guards=guards, - completeness_reasons=reasons, - judgment=judgment, - digests=dict(digests), - ) - - -__all__ = ["to_evidence_record", "to_legacy_trace_record"] +__all__ = ["to_evidence_record"] diff --git a/tests/core_migration_inventory.py b/tests/core_migration_inventory.py index bfbe899..8728ae2 100644 --- a/tests/core_migration_inventory.py +++ b/tests/core_migration_inventory.py @@ -47,6 +47,17 @@ class DependencyKind(Enum): WRITER = "writer" +_APPROVED_TERMINAL_SOURCE_CLOSURES = { + "1c8eeb5e2b6949fb391f16030e1f9aafd676fc854761a7cd1fbf574341c90ef8": MigrationDisposition.V2_MIGRATED, + "85631fe892d3ae9307ea2f93de6c0e280d2d8e0054fefffaa9b6edca330247a2": MigrationDisposition.V2_MIGRATED, + "86a1113a0052c928c53719ff8ee4b110aa4bdd1a8d20ff9ab7492227e346e872": MigrationDisposition.V2_MIGRATED, + "91f404b327f8ed526ebbd1370e6f8b065af6238e3815ece46ac1de3c35845113": MigrationDisposition.V2_MIGRATED, + "9f2097b8a673c3079a8a61cc28e9099c671e48e6c29a3e95a9e3aac3d53ccb00": MigrationDisposition.RETIRED, + "eb963401f43fca11e38c50e5e064f58b33233b9564a3a85e5b629a6fb2820d7d": MigrationDisposition.V2_MIGRATED, + "f8a5c42661f6a79b0abbe668e8993fd0ec26d1c9dad5df4e141472e6de62f586": MigrationDisposition.V2_MIGRATED, +} + + @dataclass(frozen=True) class HistoricalArtifactRecord: path: str @@ -437,8 +448,12 @@ def _moved_mixed_v1_owner(entry: LegacyInventoryEntry, *, root: Path) -> bool: return set(destination_symbols) <= owned +def _approved_terminal_source_closure(entry: LegacyInventoryEntry) -> bool: + return _APPROVED_TERMINAL_SOURCE_CLOSURES.get(entry.entry_id) is entry.disposition and entry.entry_id == entry_identity(entry) + + def validate_inventory_sources(inventory: LegacyMigrationInventory, *, root: Path) -> None: - """Prove every registered migration target is readable, nonzero, and present.""" + """Prove sources are readable and every open migration target remains present.""" for entry in inventory.entries: path = root / entry.path try: @@ -453,10 +468,11 @@ def validate_inventory_sources(inventory: LegacyMigrationInventory, *, root: Pat raise LegacyInventoryError(f"inventory source is not parseable: {entry.path}") from exc is_shim = _explicit_legacy_shim(tree, entry.symbol) moved_owner = _moved_mixed_v1_owner(entry, root=root) - if entry.symbol not in _defined_symbols(tree) and not is_shim and not moved_owner: + terminal_closure = _approved_terminal_source_closure(entry) + if entry.symbol not in _defined_symbols(tree) and not is_shim and not moved_owner and not terminal_closure: raise LegacyInventoryError(f"inventory target is absent: {entry.path}:{entry.symbol}") missing = _missing_anchors(entry, tree) - if missing and not is_shim and not moved_owner: + if missing and not is_shim and not moved_owner and not terminal_closure: raise LegacyInventoryError(f"inventory capability is absent: {entry.path}:{missing}") diff --git a/tests/fixtures/detection_core/v1/migration_inventory.json b/tests/fixtures/detection_core/v1/migration_inventory.json index f773dc1..d967c7a 100644 --- a/tests/fixtures/detection_core/v1/migration_inventory.json +++ b/tests/fixtures/detection_core/v1/migration_inventory.json @@ -6,7 +6,7 @@ "import_from:openultrasast.core.trace:Guard" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "v2_migrated", "entry_id": "86a1113a0052c928c53719ff8ee4b110aa4bdd1a8d20ff9ab7492227e346e872", "owner_boundary": "authorization-adapter", "path": "src/openultrasast/semantic/obligations/core_adapter.py", @@ -18,7 +18,7 @@ "import_from:openultrasast.core.trace:Hop" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "v2_migrated", "entry_id": "91f404b327f8ed526ebbd1370e6f8b065af6238e3815ece46ac1de3c35845113", "owner_boundary": "authorization-adapter", "path": "src/openultrasast/semantic/obligations/core_adapter.py", @@ -30,7 +30,7 @@ "import_from:openultrasast.core.trace:TraceRecord" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "v2_migrated", "entry_id": "85631fe892d3ae9307ea2f93de6c0e280d2d8e0054fefffaa9b6edca330247a2", "owner_boundary": "authorization-adapter", "path": "src/openultrasast/semantic/obligations/core_adapter.py", @@ -42,7 +42,7 @@ "import_from:openultrasast.core.trace:Witness" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "v2_migrated", "entry_id": "1c8eeb5e2b6949fb391f16030e1f9aafd676fc854761a7cd1fbf574341c90ef8", "owner_boundary": "authorization-adapter", "path": "src/openultrasast/semantic/obligations/core_adapter.py", @@ -54,7 +54,7 @@ "import_from:openultrasast.core.verdict:Judgment" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "v2_migrated", "entry_id": "f8a5c42661f6a79b0abbe668e8993fd0ec26d1c9dad5df4e141472e6de62f586", "owner_boundary": "authorization-adapter", "path": "src/openultrasast/semantic/obligations/core_adapter.py", @@ -67,7 +67,7 @@ "name:TraceRecord" ], "dependency_kind": "writer", - "disposition": "open", + "disposition": "v2_migrated", "entry_id": "eb963401f43fca11e38c50e5e064f58b33233b9564a3a85e5b629a6fb2820d7d", "owner_boundary": "authorization-adapter", "path": "src/openultrasast/semantic/obligations/core_adapter.py", @@ -985,7 +985,7 @@ "import_from:openultrasast.core._legacy_factpack:FactPackError" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "retired", "entry_id": "9eb4cba855044fe025daa7a523458c4fba0b12758ca604c894849f824f55ccb4", "owner_boundary": "core", "path": "src/openultrasast/core/factpack.py", @@ -997,7 +997,7 @@ "import_from:openultrasast.core._legacy_factpack:admit_fact_pack" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "retired", "entry_id": "1b90ea5474dff89121dc1317a86892bb0d68f2c64117923c5979cf4bb0cd8b76", "owner_boundary": "core", "path": "src/openultrasast/core/factpack.py", @@ -1009,7 +1009,7 @@ "import_from:openultrasast.core._legacy_factpack:admit_pack_file" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "retired", "entry_id": "005f1fee3a1a659a65a45d8134bc71f061e314bdc28ba95516e870997c2648cd", "owner_boundary": "core", "path": "src/openultrasast/core/factpack.py", @@ -1021,7 +1021,7 @@ "import_from:openultrasast.core._legacy_factpack:load_pack_file" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "retired", "entry_id": "4c846ba1d97e5a5b4bbf8f1e99e10c40da10906ca93d68596cb987738438582a", "owner_boundary": "core", "path": "src/openultrasast/core/factpack.py", @@ -1033,7 +1033,7 @@ "import_from:openultrasast.core._legacy_factpack:verify_fact_pack" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "retired", "entry_id": "23e6ccf25c185ca231d73ab02570b14a9cddae27f7506e95d517e241bb693ae4", "owner_boundary": "core", "path": "src/openultrasast/core/factpack.py", @@ -1046,7 +1046,7 @@ "name:LegacyProbeResult" ], "dependency_kind": "writer", - "disposition": "open", + "disposition": "retired", "entry_id": "c8aaf430009a592a8ed4361f38023d5faa12a6da6186fc0dcea9c7c522da85ad", "owner_boundary": "core", "path": "src/openultrasast/core/probes.py", @@ -1083,7 +1083,7 @@ "import_from:openultrasast.core.coverage:Coverage" ], "dependency_kind": "reader", - "disposition": "open", + "disposition": "retired", "entry_id": "9f2097b8a673c3079a8a61cc28e9099c671e48e6c29a3e95a9e3aac3d53ccb00", "owner_boundary": "hunt", "path": "src/openultrasast/hunt/prover.py", @@ -1098,7 +1098,7 @@ "name:build_trace" ], "dependency_kind": "writer", - "disposition": "open", + "disposition": "retired", "entry_id": "a5be0d8793d0048de9444366db2fd389640e99c04a9e8ef450453644c004ffa0", "owner_boundary": "hunt", "path": "src/openultrasast/hunt/prover.py", @@ -1114,7 +1114,7 @@ "call:write_text" ], "dependency_kind": "writer", - "disposition": "open", + "disposition": "retired", "entry_id": "0d2cdb31783316d18cb28e987029ac7ad1e86e1e54a568df0cfbd8e138af34ac", "owner_boundary": "hunt", "path": "src/openultrasast/hunt/report.py", @@ -1134,7 +1134,7 @@ "name:RunArtifact" ], "dependency_kind": "writer", - "disposition": "open", + "disposition": "retired", "entry_id": "5852ea0d3160cc0969ab05c1c2b11047bd5915e6164fe43d13744c053716d768", "owner_boundary": "hunt", "path": "src/openultrasast/hunt/runner.py", @@ -1175,7 +1175,7 @@ "constant:status" ], "dependency_kind": "writer", - "disposition": "open", + "disposition": "retired", "entry_id": "2e59a0d8bf2764879f78c5f85841ef496c53f2a89bd1eba7248cbf9020828c58", "owner_boundary": "push", "path": "src/openultrasast/push/runner.py", @@ -1189,7 +1189,7 @@ "constant:witnesses" ], "dependency_kind": "writer", - "disposition": "open", + "disposition": "retired", "entry_id": "577f8955e147afee6590d0e5735df7a963d29ca90ed508a46323d1a0a71f62e0", "owner_boundary": "push", "path": "src/openultrasast/push/runner.py", diff --git a/tests/test_core_migration_boundary.py b/tests/test_core_migration_boundary.py index 7f14ddb..9fb0c68 100644 --- a/tests/test_core_migration_boundary.py +++ b/tests/test_core_migration_boundary.py @@ -36,7 +36,10 @@ def _write(root: Path, relative: str, source: str) -> Path: def _closed_inventory() -> LegacyMigrationInventory: frozen = load_migration_inventory(INVENTORY) - entries = tuple(replace(entry, disposition=MigrationDisposition.V2_MIGRATED) for entry in frozen.entries) + entries = tuple( + replace(entry, disposition=MigrationDisposition.V2_MIGRATED) if entry.disposition is MigrationDisposition.OPEN else entry + for entry in frozen.entries + ) return replace(frozen, entries=entries) diff --git a/tests/test_core_migration_production.py b/tests/test_core_migration_production.py new file mode 100644 index 0000000..d8dd76e --- /dev/null +++ b/tests/test_core_migration_production.py @@ -0,0 +1,107 @@ +"""Production-source closure for the detection-core v1 migration.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +from core_migration_boundary import inspect_legacy_boundary +from core_migration_inventory import ( + REPO, + MigrationDisposition, + load_migration_inventory, + validate_historical_artifacts, + validate_inventory_sources, + validate_inventory_transition, +) +from core_migration_scope import COMPATIBILITY_PATHS, boundary_python + +INVENTORY = REPO / "tests/fixtures/detection_core/v1/migration_inventory.json" + +_V2_MIGRATED = frozenset( + { + "1c8eeb5e2b6949fb391f16030e1f9aafd676fc854761a7cd1fbf574341c90ef8", + "85631fe892d3ae9307ea2f93de6c0e280d2d8e0054fefffaa9b6edca330247a2", + "86a1113a0052c928c53719ff8ee4b110aa4bdd1a8d20ff9ab7492227e346e872", + "91f404b327f8ed526ebbd1370e6f8b065af6238e3815ece46ac1de3c35845113", + "eb963401f43fca11e38c50e5e064f58b33233b9564a3a85e5b629a6fb2820d7d", + "f8a5c42661f6a79b0abbe668e8993fd0ec26d1c9dad5df4e141472e6de62f586", + } +) +_RETIRED = frozenset( + { + "005f1fee3a1a659a65a45d8134bc71f061e314bdc28ba95516e870997c2648cd", + "0d2cdb31783316d18cb28e987029ac7ad1e86e1e54a568df0cfbd8e138af34ac", + "1b90ea5474dff89121dc1317a86892bb0d68f2c64117923c5979cf4bb0cd8b76", + "23e6ccf25c185ca231d73ab02570b14a9cddae27f7506e95d517e241bb693ae4", + "2e59a0d8bf2764879f78c5f85841ef496c53f2a89bd1eba7248cbf9020828c58", + "4c846ba1d97e5a5b4bbf8f1e99e10c40da10906ca93d68596cb987738438582a", + "577f8955e147afee6590d0e5735df7a963d29ca90ed508a46323d1a0a71f62e0", + "5852ea0d3160cc0969ab05c1c2b11047bd5915e6164fe43d13744c053716d768", + "9eb4cba855044fe025daa7a523458c4fba0b12758ca604c894849f824f55ccb4", + "9f2097b8a673c3079a8a61cc28e9099c671e48e6c29a3e95a9e3aac3d53ccb00", + "a5be0d8793d0048de9444366db2fd389640e99c04a9e8ef450453644c004ffa0", + "c8aaf430009a592a8ed4361f38023d5faa12a6da6186fc0dcea9c7c522da85ad", + } +) +_PRODUCTION_ENTRY_IDS = _V2_MIGRATED | _RETIRED + + +def _production_consumer_paths(root: Path = REPO) -> tuple[Path, ...]: + paths = [] + for path in boundary_python(root): + relative = path.relative_to(root).as_posix() + if not relative.startswith("src/"): + continue + if relative in COMPATIBILITY_PATHS or relative.startswith("src/openultrasast/core/legacy_v1/"): + continue + paths.append(path) + return tuple(paths) + + +def test_registered_production_entries_close_without_rekeying_other_consumers() -> None: + candidate = load_migration_inventory(INVENTORY) + production = {entry.entry_id: entry for entry in candidate.entries if entry.entry_id in _PRODUCTION_ENTRY_IDS} + + assert set(production) == _PRODUCTION_ENTRY_IDS + assert all(production[entry_id].disposition is MigrationDisposition.V2_MIGRATED for entry_id in _V2_MIGRATED) + assert all(production[entry_id].disposition is MigrationDisposition.RETIRED for entry_id in _RETIRED) + assert all(entry.disposition is MigrationDisposition.OPEN for entry in candidate.entries if entry.entry_id not in production) + + frozen = replace( + candidate, + entries=tuple( + replace(entry, disposition=MigrationDisposition.OPEN) if entry.entry_id in production else entry for entry in candidate.entries + ), + ) + validate_inventory_transition(frozen, candidate) + validate_inventory_sources(candidate, root=REPO) + + +def test_verified_nonzero_production_consumer_scope_has_exact_zero_legacy_references() -> None: + inventory = load_migration_inventory(INVENTORY) + paths = _production_consumer_paths() + + report = inspect_legacy_boundary( + paths, + frozen_inventory=inventory, + candidate_inventory=inventory, + expected_scope=paths, + root=REPO, + ) + + assert report.files_read == len(paths) > 0 + assert report.bytes_read > 0 + assert report.scanned_scope_paths == report.expected_scope_paths + assert report.consumer_legacy_import_sites == () + assert report.consumer_legacy_writer_sites == () + assert report.forbidden_dynamic_sites == () + assert report.compatibility_implementation_sites == () + + +def test_production_closure_preserves_frozen_historical_bytes() -> None: + inventory = load_migration_inventory(INVENTORY) + + artifacts = validate_historical_artifacts(inventory, root=REPO) + + assert len(artifacts) == len(inventory.historical_artifacts) > 0 diff --git a/tests/test_core_migration_provenance.py b/tests/test_core_migration_provenance.py index cd63259..feb11e8 100644 --- a/tests/test_core_migration_provenance.py +++ b/tests/test_core_migration_provenance.py @@ -81,7 +81,9 @@ def test_frozen_inventory_resolves_nonzero_sources_and_targets() -> None: inventory = load_migration_inventory(INVENTORY) assert inventory.entries - assert all(entry.disposition is MigrationDisposition.OPEN for entry in inventory.entries) + assert sum(entry.disposition is MigrationDisposition.OPEN for entry in inventory.entries) == 180 + assert sum(entry.disposition is MigrationDisposition.V2_MIGRATED for entry in inventory.entries) == 6 + assert sum(entry.disposition is MigrationDisposition.RETIRED for entry in inventory.entries) == 12 assert all(entry.entry_id == entry_identity(entry) for entry in inventory.entries) validate_inventory_sources(inventory, root=REPO) @@ -198,7 +200,7 @@ def test_independent_reader_audit_matches_split_inventory_capabilities() -> None def test_inventory_targets_require_exact_capability_anchors(tmp_path: Path) -> None: frozen = load_migration_inventory(INVENTORY) - entry = frozen.entries[0] + entry = next(item for item in frozen.entries if item.disposition is MigrationDisposition.OPEN and item.symbol == "") assert entry.capability_anchors substituted = tmp_path / entry.path substituted.parent.mkdir(parents=True) @@ -220,8 +222,9 @@ def test_historical_artifact_manifest_pins_exact_nonzero_bytes() -> None: @pytest.mark.parametrize("terminal", (MigrationDisposition.V2_MIGRATED, MigrationDisposition.RETIRED)) def test_inventory_entries_close_monotonically_without_identity_changes(terminal: MigrationDisposition) -> None: frozen = load_migration_inventory(INVENTORY) - changed_entry = replace(frozen.entries[0], disposition=terminal) - candidate = replace(frozen, entries=(changed_entry, *frozen.entries[1:])) + index = next(index for index, entry in enumerate(frozen.entries) if entry.disposition is MigrationDisposition.OPEN) + changed_entry = replace(frozen.entries[index], disposition=terminal) + candidate = replace(frozen, entries=(*frozen.entries[:index], changed_entry, *frozen.entries[index + 1 :])) validate_inventory_transition(frozen, candidate) @@ -279,7 +282,7 @@ def test_inventory_refuses_schema_and_status_mutations(tmp_path: Path, field: st def test_inventory_refuses_unreadable_zero_and_unknown_targets(tmp_path: Path) -> None: frozen = load_migration_inventory(INVENTORY) - original = frozen.entries[0] + original = next(entry for entry in frozen.entries if entry.disposition is MigrationDisposition.OPEN) missing = replace(original, path="missing.py", symbol="") missing = replace(missing, entry_id=entry_identity(missing)) diff --git a/tests/test_core_native.py b/tests/test_core_native.py index 206660f..72117fb 100644 --- a/tests/test_core_native.py +++ b/tests/test_core_native.py @@ -1,4 +1,4 @@ -"""Native conformance tests: prover Trace/Coverage implement the core contracts directly (task 8.1).""" +"""Legacy test conformance after production adapters leave the v1 surface.""" import sys from pathlib import Path @@ -8,87 +8,42 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from openultrasast.core.trace import TraceRecord # noqa: E402 from openultrasast.core.verdict import Judgment # noqa: E402 from openultrasast.hunt.prover import Coverage, PairJudgmentCounts # noqa: E402 -from openultrasast.semantic.obligations.abstract_domain import ( # noqa: E402 - AuthorityRef, - Completeness, - EvidenceRelation, - PairJudgment, - PathJudgment, - ResourceRef, - WitnessMatch, -) -from openultrasast.semantic.obligations.core_adapter import to_legacy_trace_record # noqa: E402 -from openultrasast.semantic.obligations.trace import AbstractTraceState, Guard, Hop, Trace # noqa: E402 -def _trace() -> Trace: - state = AbstractTraceState( - path_judgment=PathJudgment( - judgment=PairJudgment.UNGUARDED, - missing=("identity_constraint",), - completeness=Completeness(is_complete=True, reasons=()), - matches=(WitnessMatch(EvidenceRelation.MUST, EvidenceRelation.MUST, EvidenceRelation.NO, ("fact:f",)),), - ), - resource=ResourceRef(kind="exact", values=("credentials",)), - authority=AuthorityRef(values=("admin",), is_unknown=False), - resource_relation=EvidenceRelation.MUST, - authority_relation=EvidenceRelation.MUST, - ) - return Trace( - entry_kind="route", - boundary="request", - access="public", - parameters=("request",), - hops=( - Hop("handler", 10, (Guard("permission_check", "path", 9, "fact:f", "f1", "all_paths", "branches"),), ()), - Hop("helper", 20, (Guard("inner", "path", 19, "fact:g", "g1", "some_paths", "returns"),), ()), - ), +def _legacy_record(*, judgment: Judgment = Judgment.UNGUARDED) -> TraceRecord: + reasons = ("abstract_state_unset",) if judgment is Judgment.INDETERMINATE else () + return TraceRecord( + entry="src/app.py::handler", + hops=(), operation_kind="protected_read", - fact_id="obligation:protected_read:src/app.py:30", - operation_line=30, - arguments=(), - guards_after=(Guard("after", "op", 31, "fact:h", "h1", "none", "exits"),), - requires=("identity_constraint",), - dominating=(), - missing="identity_constraint", - abstract_state=state, + resource="credentials", + authority="admin", + witnesses=(), + guards=(), + completeness_reasons=reasons, + judgment=judgment, + digests={"input": "abc123"}, ) -def test_trace_natively_conforms_to_core_schema() -> None: - record = to_legacy_trace_record(_trace(), entry="src/app.py::handler", digests={"input": "abc123"}) +def test_legacy_trace_contract_remains_available_only_as_tested_compatibility() -> None: + record = _legacy_record() + assert isinstance(record, TraceRecord) assert record.entry == "src/app.py::handler" assert record.judgment is Judgment.UNGUARDED assert record.resource == "credentials" - assert [g.kind for g in record.guards] == ["permission_check", "inner", "after"] - assert record.hops[0].complete is False # conservative default - assert record.canonical() == to_legacy_trace_record(_trace(), entry="src/app.py::handler", digests={"input": "abc123"}).canonical() + assert record.canonical() == _legacy_record().canonical() -def test_legacy_trace_without_abstract_state_is_indeterminate() -> None: - trace = _trace() - legacy = Trace( - entry_kind=trace.entry_kind, - boundary=trace.boundary, - access=trace.access, - parameters=trace.parameters, - hops=trace.hops, - operation_kind=trace.operation_kind, - fact_id=trace.fact_id, - operation_line=trace.operation_line, - arguments=trace.arguments, - guards_after=trace.guards_after, - requires=trace.requires, - dominating=trace.dominating, - missing=trace.missing, - ) - record = to_legacy_trace_record(legacy, entry="e", digests={}) +def test_legacy_indeterminate_record_stays_explicit() -> None: + record = _legacy_record(judgment=Judgment.INDETERMINATE) + assert record.judgment is Judgment.INDETERMINATE assert record.completeness_reasons -def test_coverage_natively_conforms_to_core_schema() -> None: +def test_hunt_coverage_retires_its_legacy_core_projection() -> None: coverage = Coverage( files_by_language={"java": 2}, frameworks_recognised={"jenkins": 1}, @@ -101,23 +56,6 @@ def test_coverage_natively_conforms_to_core_schema() -> None: pair_judgments=PairJudgmentCounts(unguarded=1, discharged=1, indeterminate=2), resolution_counts={"complete_calls": 5, "incomplete_calls": 3, "unresolved_calls": 2}, ) - core = coverage.to_core() - assert core.calls == 10 and core.complete == 5 and core.incomplete == 3 and core.unresolved == 2 - core.validate() - bad = Coverage( - files_by_language={}, - frameworks_recognised={}, - entry_points=1, - reached=1, - unreached=0, - operations_checked=1, - operations_discharged=0, - unchecked_by_reason={}, - pair_judgments=PairJudgmentCounts(unguarded=0, discharged=0, indeterminate=1), - resolution_counts={"complete_calls": 2, "mystery_key": 7}, - ) - try: - bad.to_core() - raise AssertionError("unknown resolution keys must fail loud") - except ValueError as error: - assert "mystery_key" in str(error) + + assert coverage.resolution_counts == {"complete_calls": 5, "incomplete_calls": 3, "unresolved_calls": 2} + assert not hasattr(coverage, "to_core") diff --git a/tests/test_core_validation.py b/tests/test_core_validation.py index 2499844..f1e720c 100644 --- a/tests/test_core_validation.py +++ b/tests/test_core_validation.py @@ -4,22 +4,13 @@ import json import sys from pathlib import Path +from test_core_native import _legacy_record + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from openultrasast.core.admission import AdmissionManifestError, validate_report_schema # noqa: E402 from openultrasast.core.consumer import PushRequest, consume_verdict # noqa: E402 from openultrasast.core.verdict import Judgment, ProvenClaim # noqa: E402 -from openultrasast.semantic.obligations.abstract_domain import ( # noqa: E402 - AuthorityRef, - Completeness, - EvidenceRelation, - PairJudgment, - PathJudgment, - ResourceRef, - WitnessMatch, -) -from openultrasast.semantic.obligations.core_adapter import to_legacy_trace_record # noqa: E402 -from openultrasast.semantic.obligations.trace import AbstractTraceState, Trace # noqa: E402 REPO = Path(__file__).resolve().parents[1] @@ -40,35 +31,7 @@ def test_schema_validator_refuses_incomplete_report() -> None: def test_end_to_end_push_through_core_contracts() -> None: - state = AbstractTraceState( - path_judgment=PathJudgment( - judgment=PairJudgment.UNGUARDED, - missing=("identity_constraint",), - completeness=Completeness(is_complete=True, reasons=()), - matches=(WitnessMatch(EvidenceRelation.MUST, EvidenceRelation.MUST, EvidenceRelation.NO, ("fact:f",)),), - ), - resource=ResourceRef(kind="exact", values=("credentials",)), - authority=AuthorityRef(values=("admin",), is_unknown=False), - resource_relation=EvidenceRelation.MUST, - authority_relation=EvidenceRelation.MUST, - ) - trace = Trace( - entry_kind="route", - boundary="request", - access="public", - parameters=("request",), - hops=(), - operation_kind="protected_read", - fact_id="obligation:protected_read:src/app.py:30", - operation_line=30, - arguments=(), - guards_after=(), - requires=("identity_constraint",), - dominating=(), - missing="identity_constraint", - abstract_state=state, - ) - record = to_legacy_trace_record(trace, entry="src/app.py::handler", digests={"input": "abc123"}) + record = _legacy_record() claim = ProvenClaim(trace_id="t1", witness="fact:f", path="src/app.py", line=30, family="access_control") verdict = consume_verdict(PushRequest(base="b", head="h", budget_class="dev-pool"), record, (claim,), ()) assert verdict.claims[0].witness == "fact:f" diff --git a/tests/test_obligation_core_adapter.py b/tests/test_obligation_core_adapter.py index 6542eba..6d27d4e 100644 --- a/tests/test_obligation_core_adapter.py +++ b/tests/test_obligation_core_adapter.py @@ -6,15 +6,54 @@ import inspect from dataclasses import replace from test_core_evidence import _scope, _stamp -from test_core_native import _trace from test_core_reference_families import _contract, _load from openultrasast.core import evidence, family, semantics from openultrasast.semantic.obligations import trace as trace_module -from openultrasast.semantic.obligations.abstract_domain import PairJudgment, ResourceRef +from openultrasast.semantic.obligations.abstract_domain import ( + AuthorityRef, + Completeness, + EvidenceRelation, + PairJudgment, + PathJudgment, + ResourceRef, + WitnessMatch, +) from openultrasast.semantic.obligations.core_adapter import to_evidence_record +def _trace() -> trace_module.Trace: + state = trace_module.AbstractTraceState( + path_judgment=PathJudgment( + judgment=PairJudgment.UNGUARDED, + missing=("identity_constraint",), + completeness=Completeness(is_complete=True, reasons=()), + matches=(WitnessMatch(EvidenceRelation.MUST, EvidenceRelation.MUST, EvidenceRelation.NO, ("fact:f",)),), + ), + resource=ResourceRef(kind="exact", values=("credentials",)), + authority=AuthorityRef(values=("admin",), is_unknown=False), + resource_relation=EvidenceRelation.MUST, + authority_relation=EvidenceRelation.MUST, + ) + guard = trace_module.Guard("permission_check", "path", 2, "fact:f", "f1", "all_paths", "branches") + return trace_module.Trace( + entry_kind="route", + boundary="request", + access="public", + parameters=("request",), + hops=(trace_module.Hop("handler", 2, (guard,), ()),), + operation_kind="protected_read", + fact_id="obligation:protected_read:src/app.py:3", + operation_line=3, + arguments=(), + guards_after=(), + requires=("identity_constraint",), + dominating=(), + missing="identity_constraint", + abstract_state=state, + ) + + def _adapt(trace: trace_module.Trace) -> evidence.EvidenceRecord: contract = _contract(_load("obligation_family.json")) stamp = replace(_stamp(), family_contract_digest=family.family_contract_digest(contract))