From 7c0745247e1d7a58d7a4061b9517d1a79b9ff8c3 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:48:25 +0100 Subject: [PATCH] feat(access-control-prover): integrate reviewed entry facts --- .kiro/specs/access-control-prover/tasks.md | 2 +- .../mapping/entry_fact_integration.py | 152 ++++++++ src/openultrasast/mapping/legacy.py | 40 +- tests/test_entry_fact_integration.py | 362 ++++++++++++++++++ 4 files changed, 554 insertions(+), 2 deletions(-) create mode 100644 src/openultrasast/mapping/entry_fact_integration.py create mode 100644 tests/test_entry_fact_integration.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index e80b34f..e7051ff 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -389,7 +389,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 7.2, 7.4, 12.2, 12.3, 13.7_ - _Boundary: Downstream Adapter Conformance Handoff_ - _Depends: 14.7_ -- [ ] 14.9 Migrate the production entry caller and prove zero bypasses +- [x] 14.9 Migrate the production entry caller and prove zero bypasses - Wire the externally bootstrapped authority-bearing adapter into the target-aware `mapping/legacy.py` entry path while keeping `analyze_entry_points` stable for its callers. - Retain the legacy projection only for compatibility recognition; prohibit it from supplying v2 evidence, readiness or admission. - Run import/caller and artifact-writer censuses proving no production path bypasses authority or treats legacy projection output as v2 evidence. diff --git a/src/openultrasast/mapping/entry_fact_integration.py b/src/openultrasast/mapping/entry_fact_integration.py new file mode 100644 index 0000000..e294006 --- /dev/null +++ b/src/openultrasast/mapping/entry_fact_integration.py @@ -0,0 +1,152 @@ +"""Authority-bearing integration of reviewed entry facts at the mapping boundary.""" + +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from pathlib import Path + +from openultrasast.core import family +from openultrasast.core.factpack import FactPack, StructuralRequirementKind +from openultrasast.mapping.entry_fact_matcher import EntryFactMatchResult, match_entry_fact +from openultrasast.mapping.entry_fact_structure import candidates, read_source_structure +from openultrasast.mapping.entry_packs import load_installed_entry_fact_authority +from openultrasast.mapping.fact_authority import EntryFactAuthorityBundle, EntryFactAuthorityContext +from openultrasast.mapping.fact_match_store import FactMatchArtifactWriter +from openultrasast.mapping.installed_fact_authority import DigestDirectoryResolver, fact_pack_bytes, installed_authority_bytes +from openultrasast.preprocess import FileTarget + +_DIGEST = re.compile(r"[0-9a-f]{64}") +_LANGUAGES = frozenset(("javascript", "typescript")) + + +class EntryFactIntegrationError(ValueError): + """The production mapping boundary cannot run reviewed fact matching safely.""" + + +def _validate_contract_and_packs(supplied: EntryFactAuthorityBundle, installed: EntryFactAuthorityBundle) -> None: + if type(supplied.contract) is not family.FamilyContract: + raise TypeError("authority contract must use the exact FamilyContract type") + try: + supplied_contract = family.family_contract_digest(supplied.contract) + installed_contract = family.family_contract_digest(installed.contract) + except (family.FamilyContractError, TypeError, ValueError) as exc: + raise EntryFactIntegrationError("authority contract cannot be validated") from exc + if supplied_contract != installed_contract: + raise EntryFactIntegrationError("authority contract differs from the installed reviewed contract") + if type(supplied.packs) is not tuple or len(supplied.packs) != len(installed.packs): + raise EntryFactIntegrationError("authority pack tuple differs from the installed reviewed packs") + for supplied_pack, installed_pack in zip(supplied.packs, installed.packs, strict=True): + if type(supplied_pack) is not FactPack or type(supplied_pack.language) is not str: + raise TypeError("authority packs must use exact FactPack and language types") + if supplied_pack.language != installed_pack.language or fact_pack_bytes(supplied_pack) != fact_pack_bytes(installed_pack): + raise EntryFactIntegrationError("authority pack body differs from the installed reviewed pack") + + +def _validate_inputs( + authority: EntryFactAuthorityContext, + writer: FactMatchArtifactWriter, + evaluation_population_digest: str, +) -> EntryFactAuthorityContext: + if type(authority) is not EntryFactAuthorityContext: + raise TypeError("authority must be an authenticated EntryFactAuthorityContext") + if type(authority.bundle) is not EntryFactAuthorityBundle: + raise TypeError("authority bundle must use the exact reviewed bundle type") + if type(writer) is not FactMatchArtifactWriter: + raise TypeError("writer must be a FactMatchArtifactWriter") + if writer.store_identity not in authority.bundle.review_policy.trusted_store_identities: + raise EntryFactIntegrationError("fact-match writer does not use a trusted store") + if not isinstance(evaluation_population_digest, str) or _DIGEST.fullmatch(evaluation_population_digest) is None: + raise EntryFactIntegrationError("evaluation population digest is invalid") + installed = load_installed_entry_fact_authority(authority.custody) + _validate_contract_and_packs(authority.bundle, installed.bundle) + if installed_authority_bytes(authority.bundle) != installed_authority_bytes(installed.bundle): + raise EntryFactIntegrationError("authority context does not equal externally authenticated installed authority") + supplied_resolver, installed_resolver = authority.resolver, installed.resolver + if type(supplied_resolver) is not DigestDirectoryResolver or type(installed_resolver) is not DigestDirectoryResolver: + raise EntryFactIntegrationError("authority resolver is not the exact installed resolver type") + if ( + type(supplied_resolver.root) is not type(installed_resolver.root) + or supplied_resolver.root.parts != installed_resolver.root.parts + or type(supplied_resolver.store_identity) is not str + or supplied_resolver.store_identity != installed_resolver.store_identity + ): + raise EntryFactIntegrationError("authority resolver does not equal the installed resolver identity") + return installed + + +def _pack(authority: EntryFactAuthorityContext, language: str) -> FactPack: + packs = tuple(item for item in authority.bundle.packs if item.language == language) + if len(packs) != 1: + raise EntryFactIntegrationError("reviewed language pack is missing or ambiguous") + return packs[0] + + +def _target_for_fact(fact: object) -> str: + requirements = getattr(fact, "structural_requirements", ()) + values = tuple( + item.expected for item in requirements if item.kind is StructuralRequirementKind.QUALIFIED_NAME and isinstance(item.expected, str) + ) + if len(values) != 1: + raise EntryFactIntegrationError("reviewed fact does not declare one qualified target") + return values[0] + + +def _candidate_fact_ids(repository_root: Path, target: FileTarget, pack: FactPack) -> tuple[str, ...]: + source = read_source_structure(repository_root, target.path, target.language) + return tuple(fact.fact_id for fact in pack.facts if candidates(source, _target_for_fact(fact))) + + +def _validate_source_texts(repository_root: Path, targets: Sequence[FileTarget], source_texts: Mapping[str, str] | None) -> None: + if source_texts is None: + return + for target in targets: + if target.language not in _LANGUAGES: + continue + try: + installed = (repository_root / target.path).read_bytes().decode("utf-8") + except (OSError, UnicodeDecodeError) as exc: + raise EntryFactIntegrationError("reviewed matching requires readable UTF-8 product source") from exc + if source_texts.get(target.path) != installed: + raise EntryFactIntegrationError("cached source text differs from reviewed matcher input") + + +def match_reviewed_entry_targets( + repository_root: Path, + targets: Sequence[FileTarget], + authority: EntryFactAuthorityContext, + writer: FactMatchArtifactWriter, + evaluation_population_digest: str, + *, + source_texts: Mapping[str, str] | None = None, +) -> tuple[EntryFactMatchResult, ...]: + """Persist and validate every structurally present reviewed entry fact. + + Compatibility projection rows are deliberately absent from this interface. They + may recognize legacy entries, but only reviewed pack facts with typed structure + reach the matcher and its canonical artifact writer. + """ + authenticated = _validate_inputs(authority, writer, evaluation_population_digest) + root = Path(repository_root) + _validate_source_texts(root, targets, source_texts) + results: list[EntryFactMatchResult] = [] + for target in targets: + if target.language not in _LANGUAGES: + continue + pack = _pack(authenticated, target.language) + for fact_id in _candidate_fact_ids(root, target, pack): + results.extend( + match_entry_fact( + repository_root=root, + relative_path=target.path, + language=target.language, + fact_id=fact_id, + authority=authenticated, + writer=writer, + evaluation_population_digest=evaluation_population_digest, + ) + ) + return tuple(results) + + +__all__ = ["EntryFactIntegrationError", "match_reviewed_entry_targets"] diff --git a/src/openultrasast/mapping/legacy.py b/src/openultrasast/mapping/legacy.py index 269538a..5dc53c1 100644 --- a/src/openultrasast/mapping/legacy.py +++ b/src/openultrasast/mapping/legacy.py @@ -6,12 +6,17 @@ import re from collections.abc import Mapping, Sequence from dataclasses import asdict, dataclass, replace from pathlib import Path -from typing import Any, Literal, cast +from typing import TYPE_CHECKING, Any, Literal, cast from ..preprocess import FileTarget from .entry_packs import load_entry_pack_patterns from .registry import registry_entries +if TYPE_CHECKING: + from .entry_fact_matcher import EntryFactMatchResult + from .fact_authority import EntryFactAuthorityContext + from .fact_match_store import FactMatchArtifactWriter + MappingKind = Literal["semgrep", "codeql", "differential", "sharp_edge"] AccessLevel = Literal[ "public", @@ -76,6 +81,14 @@ class EntryPointRecord: registered_by: str = "" +@dataclass(frozen=True) +class EntryPointFactAnalysis: + """Byte-stable compatibility records plus explicit reviewed v2 matches.""" + + entry_points: tuple[EntryPointRecord, ...] + fact_matches: tuple[EntryFactMatchResult, ...] + + def ingest_sarif(path: Path) -> list[StaticHint]: payload = json.loads(path.read_text()) hints: list[StaticHint] = [] @@ -121,6 +134,31 @@ def analyze_entry_points( return sorted(records, key=lambda item: (item.path, item.line is None, item.line or 0, item.name)) +def analyze_entry_points_with_facts( + root: Path, + targets: list[FileTarget], + *, + authority: EntryFactAuthorityContext, + writer: FactMatchArtifactWriter, + evaluation_population_digest: str, + texts: Mapping[str, str] | None = None, + include_manifests: bool = True, +) -> EntryPointFactAnalysis: + """Run reviewed v2 matching beside unchanged compatibility recognition.""" + from .entry_fact_integration import match_reviewed_entry_targets + + entries = analyze_entry_points(root, targets, texts=texts, include_manifests=include_manifests) + matches = match_reviewed_entry_targets( + root, + targets, + authority, + writer, + evaluation_population_digest, + source_texts=texts, + ) + return EntryPointFactAnalysis(tuple(entries), matches) + + def php_hook_callbacks(root: Path, targets: list[FileTarget], register_calls: Sequence[str] | None = None) -> dict[str, tuple[str, ...]]: """WordPress's hook registry, read out of the source text because the graph does not carry it. diff --git a/tests/test_entry_fact_integration.py b/tests/test_entry_fact_integration.py new file mode 100644 index 0000000..ab2275f --- /dev/null +++ b/tests/test_entry_fact_integration.py @@ -0,0 +1,362 @@ +"""Production mapping integration for reviewed entry facts.""" + +from __future__ import annotations + +import ast +import json +import subprocess +import sys +from dataclasses import asdict, replace +from pathlib import Path + +import pytest + +from openultrasast.core import canonical, provenance +from openultrasast.core.semantics import RelationStatus +from openultrasast.mapping.entry_fact_integration import match_reviewed_entry_targets +from openultrasast.mapping.entry_packs import load_installed_entry_fact_authority +from openultrasast.mapping.fact_authority import EntryFactAuthorityContext +from openultrasast.mapping.fact_match_store import FactMatchArtifactWriter +from openultrasast.mapping.legacy import analyze_entry_points, analyze_entry_points_with_facts +from openultrasast.preprocess import FileTarget, detect_language + +SRC = Path(__file__).parents[1] / "src" / "openultrasast" + + +class _UnusedCustody: + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + raise AssertionError(f"authority load read custody: {population_digest}") + + +class _Custody: + def __init__(self, logs: dict[str, provenance.PopulationCustodyLog]) -> None: + self.logs = logs + + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + return self.logs[population_digest] + + +class _EqualitySpoofResolver: + def __eq__(self, other: object) -> bool: + return True + + def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent: + raise AssertionError(f"synthetic resolver used: {reference.artifact_id}") + + +class _ContextSubclass(EntryFactAuthorityContext): + pass + + +class _MemoryWriter(FactMatchArtifactWriter): + def persist_observation(self, value: object) -> canonical.ContentRef: + raise AssertionError(f"memory writer used: {value!r}") + + def persist_match(self, value: object) -> canonical.ContentRef: + raise AssertionError(f"memory writer used: {value!r}") + + +def _authority() -> tuple[EntryFactAuthorityContext, str]: + loaded = load_installed_entry_fact_authority(_UnusedCustody()) + bundle = loaded.bundle + custodian = bundle.custody_policy.reviewed_custodian_digests[0] + logs: dict[str, provenance.PopulationCustodyLog] = {} + for lineage in bundle.lineages: + for use in lineage.uses: + prior = logs.setdefault( + use.source_population_digest, + provenance.empty_population_custody_log(use.source_population_digest, custodian), + ) + request = provenance.PopulationUseRequest( + use.source_population_digest, + lineage.arm_id, + use.purpose, + use.labels_accessed, + use.influenced_artifact_digest, + prior.log_digest, + ) + receipt = provenance.record_population_use(request, prior, custodian, bundle.custody_policy) + assert receipt.receipt_digest == use.population_use_receipt_digest + logs[use.source_population_digest] = provenance.append_population_use(prior, receipt, bundle.custody_policy) + evaluation = canonical.content_digest("production-entry-integration-population") + logs[evaluation] = provenance.empty_population_custody_log(evaluation, custodian) + return replace(loaded, custody=_Custody(logs)), evaluation + + +def _target(root: Path, name: str, content: str) -> FileTarget: + path = root / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + assert path.stat().st_size > 0 + return FileTarget(name, str(path), detect_language(path), content.count("\n") + 1, [], False, [], []) + + +def _writer(root: Path, authority: EntryFactAuthorityContext) -> FactMatchArtifactWriter: + return FactMatchArtifactWriter( + root / "fact-matches", + authority.bundle.review_policy.trusted_store_identities[0], + canonical.content_digest("retain-production-entry-facts"), + ) + + +def _canonical_records(values: object) -> bytes: + return json.dumps([asdict(item) for item in values], sort_keys=True, separators=(",", ":")).encode() + + +def test_default_entry_mapping_remains_byte_stable_when_v2_is_not_requested(tmp_path: Path) -> None: + from openultrasast.mapping import legacy + + source = 'import express from "express";\nconst app = express();\napp.get("/x", handler);\nfunction handler() {}\n' + target = _target(tmp_path, "src/routes.js", source) + before = analyze_entry_points(tmp_path, [target], include_manifests=False) + assert not hasattr(legacy, "match_reviewed_entry_targets") + after = analyze_entry_points(tmp_path, [target], include_manifests=False) + + assert _canonical_records(before) == _canonical_records(after) + assert json.loads(_canonical_records(after)) == [ + { + "access_declared": False, + "access_evidence": ['app.get("/x", handler);'], + "access_level": "public", + "conditions": [], + "defines_handler": True, + "end_line": 4, + "function_name": "handler", + "kind": "route", + "line": 4, + "name": "http_handler", + "path": "src/routes.js", + "provenance": "entrypoint:route:public", + "rationale": "route surface classified as public at http_request", + "registered_by": "", + "trust_boundary": "http_request", + } + ] + + +def test_explicit_v2_path_persists_complete_matches_beside_legacy_output(tmp_path: Path) -> None: + authority, evaluation = _authority() + writer = _writer(tmp_path, authority) + target = _target( + tmp_path, + "src/routes.js", + 'import express from "express";\n' + "const app = express();\n" + 'app.get("/admin", adminHandler);\n' + "function adminHandler(req, res) { res.sendStatus(204); }\n", + ) + + analysis = analyze_entry_points_with_facts( + tmp_path, + [target], + authority=authority, + writer=writer, + evaluation_population_digest=evaluation, + include_manifests=False, + ) + + assert _canonical_records(analysis.entry_points) == _canonical_records( + analyze_entry_points(tmp_path, [target], include_manifests=False) + ) + route = next(item for item in analysis.fact_matches if item.match.fact_id == "entry:express:app-route:get") + assert route.decision.status is RelationStatus.COMPLETE + assert route.relations + assert writer.resolver.resolve(route.match_reference).content + assert all(writer.resolver.resolve(item.evidence).content for item in route.match.observations) + + +@pytest.mark.parametrize( + "name,source,reason,legacy_only", + ( + ( + "name-only.js", + 'app.get("/admin", adminHandler);\nfunction adminHandler(req, res) { res.sendStatus(204); }\n', + "missing_observation:receiver", + False, + ), + ( + "partial.js", + 'import express from "express";\nconst app = express();\napp.get("/admin");\n', + "missing_observation:arguments", + False, + ), + ( + "legacy-only.js", + '// Legacy recognition text only: app.get("/admin", adminHandler)\nexport const marker = "not structural";\n', + None, + True, + ), + ), +) +def test_projection_or_incomplete_structure_cannot_become_v2_evidence( + tmp_path: Path, name: str, source: str, reason: str | None, legacy_only: bool +) -> None: + authority, evaluation = _authority() + target = _target(tmp_path, name, source) + analysis = analyze_entry_points_with_facts( + tmp_path, + [target], + authority=authority, + writer=_writer(tmp_path, authority), + evaluation_population_digest=evaluation, + include_manifests=False, + ) + + route_matches = [item for item in analysis.fact_matches if item.match.fact_id == "entry:express:app-route:get"] + if legacy_only: + assert analysis.entry_points and route_matches == [] + else: + assert len(route_matches) == 1 + assert route_matches[0].decision.status is not RelationStatus.COMPLETE + assert reason in route_matches[0].decision.reasons + assert all(relation.status is not RelationStatus.COMPLETE for relation in route_matches[0].relations) + + +def test_explicit_v2_boundary_rejects_untrusted_writer_or_context(tmp_path: Path) -> None: + authority, evaluation = _authority() + target = _target(tmp_path, "src/routes.js", 'app.get("/x", handler);\nfunction handler() {}\n') + untrusted = FactMatchArtifactWriter( + tmp_path / "untrusted", + "unreviewed-store", + canonical.content_digest("retain-untrusted"), + ) + with pytest.raises(ValueError, match="trusted store"): + match_reviewed_entry_targets(tmp_path, [target], authority, untrusted, evaluation) + with pytest.raises(TypeError, match="authority"): + match_reviewed_entry_targets(tmp_path, [target], object(), _writer(tmp_path, authority), evaluation) # type: ignore[arg-type] + substituted = replace(authority, bundle=replace(authority.bundle, bundle_digest="0" * 64)) + with pytest.raises(ValueError, match="externally authenticated"): + match_reviewed_entry_targets(tmp_path, [target], substituted, _writer(tmp_path, authority), evaluation) + subclassed = _ContextSubclass(authority.bundle, authority.custody, authority.resolver) + with pytest.raises(TypeError, match="authority"): + match_reviewed_entry_targets(tmp_path, [target], subclassed, _writer(tmp_path, authority), evaluation) + + +def test_explicit_v2_rejects_equality_spoofed_resolver_before_matching(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.mapping import entry_fact_integration + + authority, evaluation = _authority() + target = _target(tmp_path, "src/routes.js", 'app.get("/x", handler);\nfunction handler() {}\n') + spoofed = replace(authority, resolver=_EqualitySpoofResolver()) + monkeypatch.setattr(entry_fact_integration, "match_entry_fact", lambda **kwargs: pytest.fail("matcher reached")) + + with pytest.raises(ValueError, match="resolver"): + match_reviewed_entry_targets(tmp_path, [target], spoofed, _writer(tmp_path, authority), evaluation) + + +def test_explicit_v2_rejects_writer_subclass_before_persistence(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.mapping import entry_fact_integration + + authority, evaluation = _authority() + target = _target(tmp_path, "src/routes.js", 'app.get("/x", handler);\nfunction handler() {}\n') + real = _writer(tmp_path, authority) + writer = _MemoryWriter(real.root, real.store_identity, real.retention_policy_digest) + monkeypatch.setattr(entry_fact_integration, "match_entry_fact", lambda **kwargs: pytest.fail("matcher reached")) + + with pytest.raises(TypeError, match="writer"): + match_reviewed_entry_targets(tmp_path, [target], authority, writer, evaluation) + + +def test_explicit_v2_rejects_same_type_pack_body_substitution_before_matching(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.mapping import entry_fact_integration + + authority, evaluation = _authority() + target = _target(tmp_path, "src/routes.js", 'app.get("/x", handler);\nfunction handler() {}\n') + pack = replace(authority.bundle.packs[0], normalized_content_digest="0" * 64) + bundle = replace(authority.bundle, packs=(pack, *authority.bundle.packs[1:])) + substituted = replace(authority, bundle=bundle) + monkeypatch.setattr(entry_fact_integration, "match_entry_fact", lambda **kwargs: pytest.fail("matcher reached")) + + with pytest.raises(ValueError, match="pack"): + match_reviewed_entry_targets(tmp_path, [target], substituted, _writer(tmp_path, authority), evaluation) + + +def test_explicit_v2_rejects_same_type_contract_body_substitution_before_matching(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.mapping import entry_fact_integration + + authority, evaluation = _authority() + target = _target(tmp_path, "src/routes.js", 'app.get("/x", handler);\nfunction handler() {}\n') + contract = replace(authority.bundle.contract, family_version="2.0.1") + substituted = replace(authority, bundle=replace(authority.bundle, contract=contract)) + monkeypatch.setattr(entry_fact_integration, "match_entry_fact", lambda **kwargs: pytest.fail("matcher reached")) + + with pytest.raises(ValueError, match="contract"): + match_reviewed_entry_targets(tmp_path, [target], substituted, _writer(tmp_path, authority), evaluation) + + +def test_default_legacy_import_does_not_eagerly_load_v2_matcher_modules() -> None: + script = """ +import sys +import openultrasast.mapping.legacy +forbidden = { + 'openultrasast.mapping.entry_fact_integration', + 'openultrasast.mapping.entry_fact_matcher', + 'openultrasast.mapping.entry_fact_structure', +} +assert forbidden.isdisjoint(sys.modules), sorted(forbidden.intersection(sys.modules)) +""" + completed = subprocess.run([sys.executable, "-I", "-c", script], capture_output=True, text=True, check=False) + assert completed.returncode == 0, completed.stderr + + +def test_explicit_v2_refuses_cached_text_that_differs_from_product_bytes(tmp_path: Path) -> None: + authority, evaluation = _authority() + source = 'app.get("/real", handler);\nfunction handler() {}\n' + target = _target(tmp_path, "src/routes.js", source) + with pytest.raises(ValueError, match="cached source text differs"): + analyze_entry_points_with_facts( + tmp_path, + [target], + authority=authority, + writer=_writer(tmp_path, authority), + evaluation_population_digest=evaluation, + texts={target.path: source.replace("/real", "/stale")}, + include_manifests=False, + ) + + +def test_production_censuses_find_no_projection_evidence_or_writer_bypass() -> None: + calls: dict[str, list[str]] = { + "load_entry_pack_patterns": [], + "match_entry_fact": [], + "persist_observation": [], + "persist_match": [], + "StructuralObservation": [], + "FactMatchEvidence": [], + "FactMatchArtifactWriter": [], + } + conformance_only = { + "mapping/_entry_fact_conformance_fs.py", + "mapping/_entry_fact_conformance_negative.py", + "mapping/_entry_fact_conformance_replay.py", + "mapping/_entry_fact_conformance_wire.py", + "mapping/entry_fact_conformance.py", + "mapping/entry_fact_conformance_manifest.py", + } + production = tuple(path for path in SRC.rglob("*.py") if path.relative_to(SRC).as_posix() not in conformance_only) + imports: dict[str, list[str]] = {"load_entry_pack_patterns": [], "match_entry_fact": []} + for path in production: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + for alias in node.names: + if alias.name in imports: + imports[alias.name].append(path.relative_to(SRC).as_posix()) + if not isinstance(node, ast.Call): + continue + name = node.func.id if isinstance(node.func, ast.Name) else node.func.attr if isinstance(node.func, ast.Attribute) else "" + if name in calls: + calls[name].append(path.relative_to(SRC).as_posix()) + + assert imports["load_entry_pack_patterns"] == ["mapping/legacy.py"] + assert calls["load_entry_pack_patterns"] == ["mapping/legacy.py"] + assert imports["match_entry_fact"] == ["mapping/entry_fact_integration.py"] + assert calls["match_entry_fact"] == ["mapping/entry_fact_integration.py"] + assert calls["persist_observation"] == ["mapping/entry_fact_matcher.py"] + assert calls["persist_match"] == ["mapping/entry_fact_matcher.py"] + assert calls["StructuralObservation"] == ["core/factvalidate.py", "mapping/entry_fact_matcher.py"] + assert calls["FactMatchEvidence"] == ["core/factvalidate.py", "mapping/entry_fact_matcher.py"] + assert calls["FactMatchArtifactWriter"] == [] + assert "load_entry_pack_patterns" not in (SRC / "mapping" / "entry_fact_integration.py").read_text(encoding="utf-8") + legacy_source = (SRC / "mapping" / "legacy.py").read_text(encoding="utf-8") + assert "FactMatchEvidence" not in legacy_source and "StructuralObservation" not in legacy_source