From d9a819c8532d5615b7806ae643fe7e1559907cef Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 2 Nov 2026 13:16:00 +0100 Subject: [PATCH] feat(detection-core): add syntax-only migration boundary --- .kiro/specs/detection-core/tasks.md | 2 +- tests/core_migration_boundary.py | 453 ++++++++++++++++++++++++++ tests/core_migration_scope.py | 206 ++++++++++++ tests/test_core_migration_boundary.py | 390 ++++++++++++++++++++++ 4 files changed, 1050 insertions(+), 1 deletion(-) create mode 100644 tests/core_migration_boundary.py create mode 100644 tests/core_migration_scope.py create mode 100644 tests/test_core_migration_boundary.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 3160d72..b9847fe 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -367,7 +367,7 @@ - Keep the canonical v2 paths private until the final atomic cutover. - Done when the old named API remains compatible before cutover, direct imports of moved legacy symbols from mixed modules fail, and no shim owns an independent implementation. - _Requirements: 14.3, 15.3_ -- [ ] 10.3 Build the syntax-only migration-boundary instrument +- [x] 10.3 Build the syntax-only migration-boundary instrument - Prove every scoped source file is readable and nonzero before interpreting a zero result. - Report explicit compatibility-namespace imports, designated shim imports, named top-level v1 symbols, writer-facade references, and module-valued or dynamic core access without evaluating alias or data flow. - Validate immutable inventory identities and their monotone open, `v2_migrated`, or `retired` status transitions. diff --git a/tests/core_migration_boundary.py b/tests/core_migration_boundary.py new file mode 100644 index 0000000..6f38c10 --- /dev/null +++ b/tests/core_migration_boundary.py @@ -0,0 +1,453 @@ +"""Syntax-only cutover boundary for the detection-core v1 migration.""" + +from __future__ import annotations + +import ast +from collections.abc import Sequence +from dataclasses import dataclass +from enum import Enum +from pathlib import Path + +from core_migration_inventory import ( + LegacyInventoryEntry, + LegacyMigrationInventory, + MigrationDisposition, + validate_inventory_transition, +) +from core_migration_scope import ( + COMPATIBILITY_PATHS, + COMPATIBILITY_SHIMS, + FROZEN_TOP_LEVEL_V1, + REPO, + MigrationScopeError, + authoritative_cutover_scope, + boundary_python, + cutover_residue_error, + cutover_scope_error, + expected_boundary_scope, + partition_boundary_sites, + relative_path, + scope_identity, +) +from core_migration_scope import ( + static_top_level_v1 as _static_top_level_v1, +) + + +class LegacyBoundaryError(ValueError): + """The structural migration boundary is unreadable or not cutover-ready.""" + + +class BoundaryPhase(Enum): + MIGRATION = "migration" + POST_DELETION = "post_deletion" + + +@dataclass(frozen=True) +class LegacyBoundaryReport: + phase: BoundaryPhase + files_read: int + bytes_read: int + scanned_scope_paths: tuple[str, ...] + scanned_scope_digest: str + expected_scope_paths: tuple[str, ...] + expected_scope_digest: str + open_inventory_entries: tuple[LegacyInventoryEntry, ...] + legacy_import_sites: tuple[str, ...] + legacy_writer_sites: tuple[str, ...] + forbidden_dynamic_sites: tuple[str, ...] + consumer_legacy_import_sites: tuple[str, ...] + consumer_legacy_writer_sites: tuple[str, ...] + compatibility_implementation_sites: tuple[str, ...] + + +@dataclass(frozen=True) +class _Bindings: + core_modules: frozenset[str] + compatibility_modules: frozenset[str] + writer_modules: frozenset[str] + writer_members: frozenset[str] + import_modules: frozenset[str] + import_functions: frozenset[str] + + +def static_top_level_v1(root: Path = REPO) -> tuple[str, ...]: + try: + return _static_top_level_v1(root) + except MigrationScopeError as exc: + raise LegacyBoundaryError(str(exc)) from exc + + +def _dotted(node: ast.AST) -> str: + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + prefix = _dotted(node.value) + return f"{prefix}.{node.attr}" if prefix else "" + return "" + + +def _parents(tree: ast.AST) -> dict[ast.AST, ast.AST]: + return {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)} + + +def _symbol(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> str: + names = [] + current = node + while current in parents: + current = parents[current] + if isinstance(current, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + names.append(current.name) + return ".".join(reversed(names)) or "" + + +def _site(relative: str, node: ast.AST, parents: dict[ast.AST, ast.AST], kind: str, detail: str = "") -> str: + suffix = f":{detail}" if detail else "" + return f"{relative}:{getattr(node, 'lineno', 0)}:{_symbol(node, parents)}:{kind}{suffix}" + + +def _bindings(tree: ast.Module) -> _Bindings: + core_modules = {"openultrasast.core"} + compatibility_modules = {"openultrasast.core.legacy_v1"} + writer_modules = {"openultrasast.core.legacy_v1.writer"} + writer_members: set[str] = set() + import_modules = {"importlib"} + import_functions = {"__import__"} + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "openultrasast.core": + core_modules.add(alias.asname or "openultrasast.core") + elif alias.name == "openultrasast.core.legacy_v1": + compatibility_modules.add(alias.asname or alias.name) + elif alias.name == "openultrasast.core.legacy_v1.writer": + writer_modules.add(alias.asname or alias.name) + elif alias.name == "importlib": + import_modules.add(alias.asname or "importlib") + elif isinstance(node, ast.ImportFrom): + _update_from_bindings(node, core_modules, compatibility_modules, writer_modules, writer_members, import_functions) + return _Bindings( + frozenset(core_modules), + frozenset(compatibility_modules), + frozenset(writer_modules), + frozenset(writer_members), + frozenset(import_modules), + frozenset(import_functions), + ) + + +def _update_from_bindings( + node: ast.ImportFrom, + core_modules: set[str], + compatibility_modules: set[str], + writer_modules: set[str], + writer_members: set[str], + import_functions: set[str], +) -> None: + if node.module == "openultrasast": + core_modules.update(alias.asname or alias.name for alias in node.names if alias.name == "core") + elif node.module == "openultrasast.core": + compatibility_modules.update(alias.asname or alias.name for alias in node.names if alias.name == "legacy_v1") + elif node.module == "openultrasast.core.legacy_v1": + writer_modules.update(alias.asname or alias.name for alias in node.names if alias.name == "writer") + elif node.module == "openultrasast.core.legacy_v1.writer": + writer_members.update(alias.asname or alias.name for alias in node.names) + elif node.module == "importlib": + import_functions.update(alias.asname or alias.name for alias in node.names if alias.name == "import_module") + + +def _import_sites(relative: str, tree: ast.Module, parents: dict[ast.AST, ast.AST]) -> tuple[set[str], set[str], set[str]]: + legacy: set[str] = set() + writers: set[str] = set() + dynamic: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + _classify_import(relative, node, alias.name, parents, legacy, writers, dynamic) + elif isinstance(node, ast.ImportFrom): + _classify_import_from(relative, node, parents, legacy, writers, dynamic) + return legacy, writers, dynamic + + +def _classify_import( + relative: str, + node: ast.Import, + module: str, + parents: dict[ast.AST, ast.AST], + legacy: set[str], + writers: set[str], + dynamic: set[str], +) -> None: + if module == "openultrasast.core": + dynamic.add(_site(relative, node, parents, "module_valued_core_import", module)) + if module.startswith("openultrasast.core.legacy_v1"): + legacy.add(_site(relative, node, parents, "legacy_namespace_import", module)) + if module == "openultrasast.core.legacy_v1.writer": + writers.add(_site(relative, node, parents, "writer_facade_import", module)) + if module in COMPATIBILITY_SHIMS: + legacy.add(_site(relative, node, parents, "compatibility_shim_import", module)) + + +def _classify_import_from( + relative: str, + node: ast.ImportFrom, + parents: dict[ast.AST, ast.AST], + legacy: set[str], + writers: set[str], + dynamic: set[str], +) -> None: + module = node.module or "" + names = {alias.name for alias in node.names} + if module == "openultrasast" and "core" in names: + dynamic.add(_site(relative, node, parents, "module_valued_core_import", "openultrasast.core")) + if module.startswith("openultrasast.core.legacy_v1") or (module == "openultrasast.core" and "legacy_v1" in names): + legacy.add(_site(relative, node, parents, "legacy_namespace_import", module or "openultrasast.core.legacy_v1")) + if module == "openultrasast.core.legacy_v1.writer" or (module == "openultrasast.core.legacy_v1" and "writer" in names): + writers.add(_site(relative, node, parents, "writer_facade_import", module)) + if module == "openultrasast.core": + if "*" in names: + dynamic.add(_site(relative, node, parents, "top_level_star_import", module)) + for name in sorted(names.intersection(FROZEN_TOP_LEVEL_V1)): + legacy.add(_site(relative, node, parents, "top_level_v1_import", name)) + for name in sorted(names): + shim = f"openultrasast.core.{name}" + if shim in COMPATIBILITY_SHIMS: + legacy.add(_site(relative, node, parents, "compatibility_shim_import", shim)) + if module in COMPATIBILITY_SHIMS: + legacy.add(_site(relative, node, parents, "compatibility_shim_import", module)) + + +def _reference_sites( + relative: str, + tree: ast.Module, + parents: dict[ast.AST, ast.AST], + bindings: _Bindings, +) -> tuple[set[str], set[str], set[str]]: + legacy: set[str] = set() + writers: set[str] = set() + dynamic: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Attribute): + _classify_attribute(relative, node, parents, bindings, legacy, writers) + elif isinstance(node, ast.Call): + _classify_call(relative, node, parents, bindings, writers, dynamic) + return legacy, writers, dynamic + + +def _classify_attribute( + relative: str, + node: ast.Attribute, + parents: dict[ast.AST, ast.AST], + bindings: _Bindings, + legacy: set[str], + writers: set[str], +) -> None: + base = _dotted(node.value) + if base in bindings.core_modules and node.attr in FROZEN_TOP_LEVEL_V1: + legacy.add(_site(relative, node, parents, "top_level_v1_attribute", node.attr)) + compatibility_writer_bases = {f"{module}.writer" for module in bindings.compatibility_modules} + if base in bindings.writer_modules or base in compatibility_writer_bases: + writers.add(_site(relative, node, parents, "writer_facade_reference", node.attr)) + + +def _classify_call( + relative: str, + node: ast.Call, + parents: dict[ast.AST, ast.AST], + bindings: _Bindings, + writers: set[str], + dynamic: set[str], +) -> None: + function = _dotted(node.func) + if isinstance(node.func, ast.Name) and node.func.id in bindings.writer_members: + writers.add(_site(relative, node, parents, "writer_facade_call", node.func.id)) + if isinstance(node.func, ast.Attribute): + base = _dotted(node.func.value) + compatibility_bases = {f"{module}.writer" for module in bindings.compatibility_modules} + if base in bindings.writer_modules or base in compatibility_bases: + writers.add(_site(relative, node, parents, "writer_facade_call", node.func.attr)) + if ( + _is_import_call(function, bindings) + and (target := _import_target(node, function)) is not None + and (target == "openultrasast.core" or target.startswith("openultrasast.core.")) + ): + dynamic.add(_site(relative, node, parents, "dynamic_core_import", target)) + if ( + function == "getattr" + and len(node.args) >= 2 + and (name := _constant_argument(node, 1, "name")) in FROZEN_TOP_LEVEL_V1 + and _is_direct_core_object(node.args[0], bindings) + ): + dynamic.add(_site(relative, node, parents, "dynamic_compatibility_getattr", name or "")) + + +def _is_import_call(function: str, bindings: _Bindings) -> bool: + if function in bindings.import_functions: + return True + return any(function == f"{module}.import_module" for module in bindings.import_modules) + + +def _constant_argument(node: ast.Call, index: int, keyword: str) -> str | None: + value: ast.AST | None = node.args[index] if len(node.args) > index else None + if value is None: + value = next((item.value for item in node.keywords if item.arg == keyword), None) + return value.value if isinstance(value, ast.Constant) and isinstance(value.value, str) else None + + +def _import_target(node: ast.Call, function: str) -> str | None: + name = _constant_argument(node, 0, "name") + if name is None or not name.startswith("."): + return name + if function == "__import__" or (package := _constant_argument(node, 1, "package")) is None: + return None + level = len(name) - len(name.lstrip(".")) + package_parts = package.split(".") + if level > len(package_parts): + return None + resolved = [*package_parts[: len(package_parts) - level + 1]] + if suffix := name[level:]: + resolved.append(suffix) + return ".".join(resolved) + + +def _is_direct_core_object(node: ast.AST, bindings: _Bindings) -> bool: + dotted = _dotted(node) + if dotted in bindings.core_modules or dotted in bindings.compatibility_modules: + return True + if not isinstance(node, ast.Call) or not _is_import_call(_dotted(node.func), bindings): + return False + target = _import_target(node, _dotted(node.func)) + return target is not None and (target == "openultrasast.core" or target.startswith("openultrasast.core.")) + + +def _inspect_file(path: Path, *, root: Path) -> tuple[int, set[str], set[str], set[str]]: + relative = relative_path(path, root) + try: + content = path.read_bytes() + except OSError as exc: + raise LegacyBoundaryError(f"boundary input is not readable: {relative}") from exc + if not content: + raise LegacyBoundaryError(f"boundary input must contain nonzero bytes: {relative}") + try: + tree = ast.parse(content, filename=relative) + except (SyntaxError, UnicodeDecodeError) as exc: + raise LegacyBoundaryError(f"boundary input is not parseable: {relative}") from exc + parents = _parents(tree) + imports = _import_sites(relative, tree, parents) + references = _reference_sites(relative, tree, parents, _bindings(tree)) + return ( + len(content), + imports[0] | references[0], + imports[1] | references[1], + imports[2] | references[2], + ) + + +def inspect_legacy_boundary( + paths: Sequence[Path], + *, + frozen_inventory: LegacyMigrationInventory, + candidate_inventory: LegacyMigrationInventory | None = None, + expected_scope: Sequence[Path] | None = None, + root: Path = REPO, + phase: BoundaryPhase = BoundaryPhase.MIGRATION, +) -> LegacyBoundaryReport: + """Inspect explicit migration syntax after proving every declared input readable.""" + candidate = candidate_inventory or frozen_inventory + validate_inventory_transition(frozen_inventory, candidate) + legacy, writers, dynamic = set(), set(), set() + bytes_read = 0 + unique_paths = tuple(sorted(set(paths), key=lambda value: value.as_posix())) + if not unique_paths: + raise LegacyBoundaryError("boundary scope must be nonempty") + if not isinstance(phase, BoundaryPhase): + raise LegacyBoundaryError("boundary phase must use BoundaryPhase") + complete_paths = expected_boundary_scope(unique_paths, expected_scope, root, post_deletion=phase is BoundaryPhase.POST_DELETION) + if complete_paths is None: + raise LegacyBoundaryError("isolated post-deletion scope requires an explicit expected scope") + scanned_scope, scanned_digest = scope_identity(unique_paths, root=root) + expected_paths, expected_digest = scope_identity(complete_paths, root=root) + for path in unique_paths: + size, file_legacy, file_writers, file_dynamic = _inspect_file(path, root=root) + bytes_read += size + legacy.update(file_legacy) + writers.update(file_writers) + dynamic.update(file_dynamic) + compatibility, consumer_imports, consumer_writers = partition_boundary_sites( + legacy, writers, COMPATIBILITY_PATHS, post_deletion=phase is BoundaryPhase.POST_DELETION + ) + open_entries = tuple(entry for entry in candidate.entries if entry.disposition is MigrationDisposition.OPEN) + return LegacyBoundaryReport( + phase, + len(unique_paths), + bytes_read, + scanned_scope, + scanned_digest, + expected_paths, + expected_digest, + open_entries, + tuple(sorted(legacy)), + tuple(sorted(writers)), + tuple(sorted(dynamic)), + consumer_imports, + consumer_writers, + compatibility, + ) + + +def require_cutover_ready( + report: LegacyBoundaryReport, + *, + frozen_inventory: LegacyMigrationInventory, + candidate_inventory: LegacyMigrationInventory, + root: Path = REPO, + isolated_scope: Sequence[Path] | None = None, +) -> None: + """Remeasure an independently authoritative scope before allowing cutover.""" + if report.phase is not BoundaryPhase.POST_DELETION: + raise LegacyBoundaryError("cutover requires a post-deletion boundary report") + try: + authoritative_scope = authoritative_cutover_scope(root, isolated_scope) + except MigrationScopeError as exc: + raise LegacyBoundaryError(str(exc)) from exc + scope_error = cutover_scope_error( + report.files_read, + report.bytes_read, + report.scanned_scope_paths, + report.scanned_scope_digest, + report.expected_scope_paths, + report.expected_scope_digest, + ) + if scope_error: + raise LegacyBoundaryError(f"cutover {scope_error}") + residue_error = cutover_residue_error( + len(report.open_inventory_entries), + report.compatibility_implementation_sites, + report.legacy_import_sites or report.consumer_legacy_import_sites, + report.legacy_writer_sites or report.consumer_legacy_writer_sites, + report.forbidden_dynamic_sites, + ) + if residue_error: + raise LegacyBoundaryError(f"cutover {residue_error}") + authoritative = inspect_legacy_boundary( + authoritative_scope, + frozen_inventory=frozen_inventory, + candidate_inventory=candidate_inventory, + expected_scope=authoritative_scope, + root=root, + phase=BoundaryPhase.POST_DELETION, + ) + if report != authoritative: + raise LegacyBoundaryError("cutover report does not match the authoritative boundary scan") + + +__all__ = [ + "BoundaryPhase", + "FROZEN_TOP_LEVEL_V1", + "LegacyBoundaryError", + "LegacyBoundaryReport", + "boundary_python", + "inspect_legacy_boundary", + "require_cutover_ready", + "static_top_level_v1", +] diff --git a/tests/core_migration_scope.py b/tests/core_migration_scope.py new file mode 100644 index 0000000..35265d4 --- /dev/null +++ b/tests/core_migration_scope.py @@ -0,0 +1,206 @@ +"""Deterministic source and compatibility-surface manifests for migration tests.""" + +from __future__ import annotations + +import ast +import hashlib +import subprocess +from collections.abc import Sequence +from pathlib import Path, PurePosixPath + +REPO = Path(__file__).resolve().parents[1] +_TASK_FILES = ( + "tests/core_migration_boundary.py", + "tests/core_migration_scope.py", + "tests/test_core_migration_boundary.py", +) +_FIXTURE_PREFIXES = ("benchmarks/fixtures/", "tests/fixtures/") +COMPATIBILITY_PATHS = frozenset( + """src/openultrasast/core/__init__.py src/openultrasast/core/_legacy_factpack.py + src/openultrasast/core/admission.py src/openultrasast/core/consumer.py + src/openultrasast/core/trace.py src/openultrasast/core/verdict.py""".split() # noqa: SIM905 +) +COMPATIBILITY_SHIMS = frozenset( + """openultrasast.core._legacy_factpack openultrasast.core.admission + openultrasast.core.consumer openultrasast.core.trace openultrasast.core.verdict""".split() # noqa: SIM905 +) +FROZEN_TOP_LEVEL_V1 = tuple( + """REPORT_SCHEMA_KEYS AdmissionManifest AdmissionManifestError BoundedSilence Budget BudgetGateFailure + Complexity Coverage FactPack FactPackError FittedComponentBlocked Guard Hop InstrumentFailure Judgment ProbeResult + ProvenClaim ProvenanceMismatch PushRequest PushVerdict ResolutionOutcome ShortcutDetected TraceRecord Witness + admit_fact_pack as_clean assert_no_shortcut check_budget check_reads consume_verdict derive_budget + require_admission_evidence reversed_pair_probe stamp validate_admission_manifest validate_report_schema + verify_fact_pack verify_stamp""".split() # noqa: SIM905 +) + + +class MigrationScopeError(ValueError): + """A scope or compatibility manifest could not be read exactly.""" + + +def _owned_python(relative: str) -> bool: + if not relative.endswith(".py") or relative.startswith(_FIXTURE_PREFIXES): + return False + path = PurePosixPath(relative) + if relative.startswith("benchmarks/pairs/"): + return path.name not in {"fixed.py", "vuln.py"} and path.parts[2] != "sast" + return True + + +def boundary_python(root: Path = REPO) -> tuple[Path, ...]: + """Return owned tracked Python, excluding only harvested fixture material.""" + completed = subprocess.run(["git", "ls-files", "*.py"], cwd=root, check=True, capture_output=True, text=True) + relatives = {value for value in completed.stdout.splitlines() if _owned_python(value)} + relatives.update(path for path in _TASK_FILES if (root / path).is_file()) + return tuple(root / relative for relative in sorted(relatives)) + + +def expected_boundary_scope( + scanned: Sequence[Path], supplied: Sequence[Path] | None, root: Path, *, post_deletion: bool +) -> Sequence[Path] | None: + if not post_deletion: + return supplied or scanned + return boundary_python(root) if root.resolve() == REPO else supplied + + +def authoritative_cutover_scope(root: Path, isolated_scope: Sequence[Path] | None) -> tuple[Path, ...]: + """Derive cutover scope independently of the supplied measurement report.""" + if root.resolve() == REPO.resolve(): + if isolated_scope is not None: + raise MigrationScopeError("repository scope cannot be overridden") + return boundary_python(root) + if isolated_scope is None: + raise MigrationScopeError("isolated cutover requires an independent scope") + scope = tuple(isolated_scope) + if not scope: + raise MigrationScopeError("isolated cutover scope must be nonempty") + return scope + + +def scope_identity(paths: Sequence[Path], *, root: Path) -> tuple[tuple[str, ...], str]: + """Return sorted relative paths and an unambiguous SHA-256 path manifest.""" + relatives = tuple(sorted({relative_path(path, root) for path in paths})) + return relatives, manifest_digest(relatives) + + +def manifest_digest(relatives: Sequence[str]) -> str: + manifest = "".join(f"{len(relative)}:{relative}\n" for relative in relatives).encode() + return hashlib.sha256(manifest).hexdigest() + + +def scope_identity_error( + scanned_paths: Sequence[str], scanned_digest: str, expected_paths: Sequence[str], expected_digest: str +) -> str | None: + if scanned_digest != manifest_digest(scanned_paths): + return "scope manifest is invalid" + if expected_digest != manifest_digest(expected_paths): + return "expected scope identity is invalid" + if tuple(scanned_paths) != tuple(expected_paths): + return "requires the complete repository scope" + return None + + +def cutover_scope_error( + files_read: int, + bytes_read: int, + scanned_paths: Sequence[str], + scanned_digest: str, + expected_paths: Sequence[str], + expected_digest: str, +) -> str | None: + if files_read <= 0 or bytes_read <= 0: + return "requires positive input read proof" + if files_read != len(scanned_paths): + return "file count does not match scanned scope" + if tuple(scanned_paths) != tuple(sorted(set(scanned_paths))): + return "scanned scope must be normalized, sorted, and unique" + return scope_identity_error(scanned_paths, scanned_digest, expected_paths, expected_digest) + + +def cutover_residue_error( + open_count: int, + compatibility: Sequence[str], + imports: Sequence[str], + writers: Sequence[str], + dynamic: Sequence[str], +) -> str | None: + checks = ( + (open_count > 0, f"has {open_count} open inventory entries"), + (bool(compatibility), f"retains compatibility implementation: {compatibility[0] if compatibility else ''}"), + (bool(imports), f"has legacy import sites: {imports[0] if imports else ''}"), + (bool(writers), f"has legacy writer sites: {writers[0] if writers else ''}"), + (bool(dynamic), f"has forbidden dynamic sites: {dynamic[0] if dynamic else ''}"), + ) + return next((message for failed, message in checks if failed), None) + + +def partition_boundary_sites( + legacy: set[str], writers: set[str], compatibility_paths: frozenset[str], *, post_deletion: bool +) -> tuple[tuple[str, ...], tuple[str, ...], tuple[str, ...]]: + if post_deletion: + return (), tuple(sorted(legacy)), tuple(sorted(writers)) + compatibility = { + site + for site in legacy | writers + if (path := site.split(":", 1)[0]) in compatibility_paths or path.startswith("src/openultrasast/core/legacy_v1/") + } + return tuple(sorted(compatibility)), tuple(sorted(legacy - compatibility)), tuple(sorted(writers - compatibility)) + + +def relative_path(path: Path, root: Path) -> str: + try: + return path.relative_to(root).as_posix() + except ValueError: + return path.as_posix() + + +def static_top_level_v1(root: Path = REPO) -> tuple[str, ...]: + """Read the migration shim's literal ``__all__`` without importing it.""" + path = root / "src/openultrasast/core/__init__.py" + try: + content = path.read_bytes() + except OSError as exc: + raise MigrationScopeError("top-level compatibility surface is not readable") from exc + if not content: + raise MigrationScopeError("top-level compatibility surface must contain nonzero bytes") + try: + tree = ast.parse(content, filename=path.as_posix()) + except (SyntaxError, UnicodeDecodeError) as exc: + raise MigrationScopeError("top-level compatibility surface is not parseable") from exc + declared = [_literal_all(node) for node in tree.body if isinstance(node, (ast.Assign, ast.AnnAssign))] + values = [value for value in declared if value is not None] + if len(values) != 1: + raise MigrationScopeError("top-level compatibility surface must declare one literal __all__") + return values[0] + + +def _literal_all(node: ast.Assign | ast.AnnAssign) -> tuple[str, ...] | None: + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + if not any(isinstance(target, ast.Name) and target.id == "__all__" for target in targets): + return None + value = node.value + if not isinstance(value, (ast.List, ast.Tuple)): + raise MigrationScopeError("top-level compatibility __all__ must be a literal sequence") + if not all(isinstance(item, ast.Constant) and isinstance(item.value, str) for item in value.elts): + raise MigrationScopeError("top-level compatibility __all__ must contain literal names") + return tuple(item.value for item in value.elts if isinstance(item, ast.Constant) and isinstance(item.value, str)) + + +__all__ = [ + "FROZEN_TOP_LEVEL_V1", + "COMPATIBILITY_PATHS", + "COMPATIBILITY_SHIMS", + "MigrationScopeError", + "REPO", + "authoritative_cutover_scope", + "boundary_python", + "cutover_scope_error", + "cutover_residue_error", + "expected_boundary_scope", + "manifest_digest", + "partition_boundary_sites", + "relative_path", + "scope_identity_error", + "scope_identity", + "static_top_level_v1", +] diff --git a/tests/test_core_migration_boundary.py b/tests/test_core_migration_boundary.py new file mode 100644 index 0000000..7f14ddb --- /dev/null +++ b/tests/test_core_migration_boundary.py @@ -0,0 +1,390 @@ +"""Task 10.3: syntax-only legacy migration boundary instrument.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +import pytest +from core_migration_boundary import ( + FROZEN_TOP_LEVEL_V1, + BoundaryPhase, + LegacyBoundaryError, + LegacyBoundaryReport, + boundary_python, + inspect_legacy_boundary, + require_cutover_ready, + static_top_level_v1, +) +from core_migration_inventory import ( + LegacyInventoryError, + LegacyMigrationInventory, + MigrationDisposition, + load_migration_inventory, +) + +REPO = Path(__file__).resolve().parents[1] +INVENTORY = REPO / "tests/fixtures/detection_core/v1/migration_inventory.json" + + +def _write(root: Path, relative: str, source: str) -> Path: + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source, encoding="utf-8") + return path + + +def _closed_inventory() -> LegacyMigrationInventory: + frozen = load_migration_inventory(INVENTORY) + entries = tuple(replace(entry, disposition=MigrationDisposition.V2_MIGRATED) for entry in frozen.entries) + return replace(frozen, entries=entries) + + +def _require_cutover( + report: LegacyBoundaryReport, + *, + frozen: LegacyMigrationInventory, + candidate: LegacyMigrationInventory, + root: Path, + isolated_scope: tuple[Path, ...] | None = None, +) -> None: + require_cutover_ready( + report, + frozen_inventory=frozen, + candidate_inventory=candidate, + root=root, + isolated_scope=isolated_scope, + ) + + +def test_boundary_reports_explicit_syntax_without_interpreting_flow(tmp_path: Path) -> None: + source = """try: + from openultrasast.core.legacy_v1 import Budget +finally: + pass +from openultrasast.core import Budget as PublicBudget +from openultrasast.core import trace as trace_shim +import openultrasast.core as core_api +from openultrasast.core.legacy_v1 import writer +items = [__import__("openultrasast.core.legacy_v1") for _ in ()] +consume(getattr(core_api, "Budget")) +writer.canonical_json({}) +alias = core_api +consume(alias) +""" + path = _write(tmp_path, "src/example.py", source) + frozen = load_migration_inventory(INVENTORY) + + report = inspect_legacy_boundary((path,), frozen_inventory=frozen, root=tmp_path) + + assert report.files_read == 1 + assert report.bytes_read == len(source.encode()) + assert any("legacy_namespace_import" in site and ":2:" in site for site in report.legacy_import_sites) + assert any("top_level_v1_import:Budget" in site for site in report.legacy_import_sites) + assert any("compatibility_shim_import:openultrasast.core.trace" in site for site in report.legacy_import_sites) + assert any("writer_facade_import" in site for site in report.legacy_writer_sites) + assert any("writer_facade_reference:canonical_json" in site for site in report.legacy_writer_sites) + assert any("writer_facade_call:canonical_json" in site for site in report.legacy_writer_sites) + assert any("module_valued_core_import" in site for site in report.forbidden_dynamic_sites) + assert any("dynamic_core_import" in site for site in report.forbidden_dynamic_sites) + assert any("dynamic_compatibility_getattr:Budget" in site for site in report.forbidden_dynamic_sites) + assert not any("alias" in site for site in (*report.legacy_import_sites, *report.legacy_writer_sites)) + + +def test_boundary_refuses_missing_empty_and_unparseable_inputs(tmp_path: Path) -> None: + frozen = load_migration_inventory(INVENTORY) + missing = tmp_path / "missing.py" + empty = _write(tmp_path, "empty.py", "") + broken = _write(tmp_path, "broken.py", "def nope(:\n") + + with pytest.raises(LegacyBoundaryError, match="readable"): + inspect_legacy_boundary((missing,), frozen_inventory=frozen, root=tmp_path) + with pytest.raises(LegacyBoundaryError, match="nonzero"): + inspect_legacy_boundary((empty,), frozen_inventory=frozen, root=tmp_path) + with pytest.raises(LegacyBoundaryError, match="parseable"): + inspect_legacy_boundary((broken,), frozen_inventory=frozen, root=tmp_path) + + +def test_boundary_reports_direct_dynamic_compatibility_attribute_access(tmp_path: Path) -> None: + source = """import openultrasast.core.legacy_v1 as legacy +import importlib +first = getattr(legacy, "Budget") +second = getattr(importlib.import_module("openultrasast.core.legacy_v1"), "Budget") +encoded = legacy.writer.canonical_json({}) +""" + path = _write(tmp_path, "dynamic.py", source) + frozen = load_migration_inventory(INVENTORY) + + report = inspect_legacy_boundary((path,), frozen_inventory=frozen, root=tmp_path) + + assert sum("dynamic_compatibility_getattr:Budget" in site for site in report.forbidden_dynamic_sites) == 2 + assert any("writer_facade_reference:canonical_json" in site for site in report.legacy_writer_sites) + + +def test_migration_phase_separates_compatibility_implementation_from_consumers(tmp_path: Path) -> None: + compatibility = _write( + tmp_path, + "src/openultrasast/core/trace.py", + "from openultrasast.core.legacy_v1.contracts import TraceRecord\n", + ) + consumer = _write(tmp_path, "src/product.py", "from openultrasast.core.legacy_v1 import TraceRecord\n") + frozen = load_migration_inventory(INVENTORY) + + migration = inspect_legacy_boundary((compatibility, consumer), frozen_inventory=frozen, root=tmp_path) + post_deletion = inspect_legacy_boundary( + (compatibility, consumer), + frozen_inventory=frozen, + expected_scope=(compatibility, consumer), + root=tmp_path, + phase=BoundaryPhase.POST_DELETION, + ) + + assert len(migration.compatibility_implementation_sites) == 1 + assert len(migration.consumer_legacy_import_sites) == 1 + assert post_deletion.compatibility_implementation_sites == () + assert len(post_deletion.consumer_legacy_import_sites) == 2 + + +def test_cutover_requires_terminal_inventory_and_structural_zero(tmp_path: Path) -> None: + clean = _write(tmp_path, "clean.py", "VALUE = 1\n") + frozen = load_migration_inventory(INVENTORY) + open_report = inspect_legacy_boundary( + (clean,), frozen_inventory=frozen, expected_scope=(clean,), root=tmp_path, phase=BoundaryPhase.POST_DELETION + ) + closed = _closed_inventory() + migration_report = inspect_legacy_boundary((clean,), frozen_inventory=frozen, candidate_inventory=closed, root=tmp_path) + clean_report = inspect_legacy_boundary( + (clean,), + frozen_inventory=frozen, + candidate_inventory=closed, + expected_scope=(clean,), + root=tmp_path, + phase=BoundaryPhase.POST_DELETION, + ) + forbidden = _write(tmp_path, "forbidden.py", "import openultrasast.core as core_api\n") + forbidden_report = inspect_legacy_boundary( + (forbidden,), + frozen_inventory=frozen, + candidate_inventory=closed, + expected_scope=(forbidden,), + root=tmp_path, + phase=BoundaryPhase.POST_DELETION, + ) + + with pytest.raises(LegacyBoundaryError, match="open inventory"): + _require_cutover(open_report, frozen=frozen, candidate=frozen, root=tmp_path, isolated_scope=(clean,)) + with pytest.raises(LegacyBoundaryError, match="post-deletion"): + _require_cutover(migration_report, frozen=frozen, candidate=closed, root=tmp_path, isolated_scope=(clean,)) + _require_cutover(clean_report, frozen=frozen, candidate=closed, root=tmp_path, isolated_scope=(clean,)) + with pytest.raises(LegacyBoundaryError, match="positive input"): + _require_cutover(replace(clean_report, files_read=0), frozen=frozen, candidate=closed, root=tmp_path, isolated_scope=(clean,)) + with pytest.raises(LegacyBoundaryError, match="compatibility implementation"): + _require_cutover( + replace(clean_report, compatibility_implementation_sites=("trace.py:1::legacy",)), + frozen=frozen, + candidate=closed, + root=tmp_path, + isolated_scope=(clean,), + ) + with pytest.raises(LegacyBoundaryError, match="forbidden dynamic"): + _require_cutover(forbidden_report, frozen=frozen, candidate=closed, root=tmp_path, isolated_scope=(forbidden,)) + + +def test_cutover_refuses_empty_scope_and_migration_compatibility(tmp_path: Path) -> None: + frozen = load_migration_inventory(INVENTORY) + closed = _closed_inventory() + compatibility = _write( + tmp_path, + "src/openultrasast/core/trace.py", + "from openultrasast.core.legacy_v1.contracts import TraceRecord\n", + ) + + with pytest.raises(LegacyBoundaryError, match="nonempty"): + inspect_legacy_boundary((), frozen_inventory=frozen, candidate_inventory=closed, root=tmp_path) + migration = inspect_legacy_boundary((compatibility,), frozen_inventory=frozen, candidate_inventory=closed, root=tmp_path) + assert migration.phase is BoundaryPhase.MIGRATION + assert migration.compatibility_implementation_sites + with pytest.raises(LegacyBoundaryError, match="post-deletion"): + _require_cutover(migration, frozen=frozen, candidate=closed, root=tmp_path, isolated_scope=(compatibility,)) + + +def test_complete_frozen_top_level_surface_is_reported_and_star_is_forbidden(tmp_path: Path) -> None: + assert static_top_level_v1(REPO) == FROZEN_TOP_LEVEL_V1 + assert len(FROZEN_TOP_LEVEL_V1) == 38 + imports = ", ".join(FROZEN_TOP_LEVEL_V1) + attributes = "\n".join(f"value_{index} = core_api.{name}" for index, name in enumerate(FROZEN_TOP_LEVEL_V1)) + dynamic = "\n".join(f'dynamic_{index} = getattr(core_api, "{name}")' for index, name in enumerate(FROZEN_TOP_LEVEL_V1)) + source = f"from openultrasast.core import {imports}\nimport openultrasast.core as core_api\n{attributes}\n{dynamic}\n" + path = _write(tmp_path, "surface.py", source) + star = _write(tmp_path, "star.py", "from openultrasast.core import *\n") + frozen = load_migration_inventory(INVENTORY) + + report = inspect_legacy_boundary((path, star), frozen_inventory=frozen, root=tmp_path) + + assert sum("top_level_v1_import" in site for site in report.legacy_import_sites) == 38 + assert sum("top_level_v1_attribute" in site for site in report.legacy_import_sites) == 38 + assert sum("dynamic_compatibility_getattr" in site for site in report.forbidden_dynamic_sites) == 38 + assert any("top_level_star_import" in site for site in report.forbidden_dynamic_sites) + + +def test_dynamic_import_supports_constant_name_keyword_only(tmp_path: Path) -> None: + source = """import importlib +from importlib import import_module as load_module +first = importlib.import_module(name="openultrasast.core.legacy_v1") +second = load_module(name="openultrasast.core") +third = __import__(name="openultrasast.core.trace") +unrelated(name="openultrasast.core.legacy_v1") +text = "openultrasast.core" +variable = "openultrasast.core" +ignored = importlib.import_module(name=variable) +""" + path = _write(tmp_path, "keyword.py", source) + frozen = load_migration_inventory(INVENTORY) + + report = inspect_legacy_boundary((path,), frozen_inventory=frozen, root=tmp_path) + + assert sum("dynamic_core_import" in site for site in report.forbidden_dynamic_sites) == 3 + + +def test_dynamic_import_resolves_literal_relative_names(tmp_path: Path) -> None: + source = """import importlib +first = importlib.import_module(".core", "openultrasast") +second = importlib.import_module(name=".legacy_v1", package="openultrasast.core") +third = importlib.import_module(".trace", package="openultrasast.core") +unrelated(".core", "openultrasast") +ignored = importlib.import_module(".core", package=variable) +text = ".legacy_v1 openultrasast.core" +""" + path = _write(tmp_path, "relative.py", source) + frozen = load_migration_inventory(INVENTORY) + + report = inspect_legacy_boundary((path,), frozen_inventory=frozen, root=tmp_path) + + assert sum("dynamic_core_import" in site for site in report.forbidden_dynamic_sites) == 3 + assert any("openultrasast.core.legacy_v1" in site for site in report.forbidden_dynamic_sites) + + +def test_boundary_validates_inventory_identity_and_status_transitions(tmp_path: Path) -> None: + clean = _write(tmp_path, "clean.py", "VALUE = 1\n") + frozen = load_migration_inventory(INVENTORY) + changed = replace(frozen.entries[0], symbol="changed") + candidate = replace(frozen, entries=(changed, *frozen.entries[1:])) + + with pytest.raises(LegacyInventoryError, match="identities"): + inspect_legacy_boundary((clean,), frozen_inventory=frozen, candidate_inventory=candidate, root=tmp_path) + + +def test_repository_scope_proves_nonzero_inputs_before_reporting_sites() -> None: + paths = boundary_python() + frozen = load_migration_inventory(INVENTORY) + + report = inspect_legacy_boundary(paths, frozen_inventory=frozen, root=REPO) + + assert report.files_read == len(paths) > 0 + assert report.bytes_read > 0 + assert report.legacy_import_sites + assert report.open_inventory_entries + + +def test_cutover_rejects_positive_but_incomplete_repository_scope() -> None: + path = REPO / "tests/conftest.py" + frozen = load_migration_inventory(INVENTORY) + report = inspect_legacy_boundary( + (path,), + frozen_inventory=frozen, + candidate_inventory=_closed_inventory(), + expected_scope=(path,), + root=REPO, + phase=BoundaryPhase.POST_DELETION, + ) + + assert report.files_read == 1 + assert report.scanned_scope_paths == ("tests/conftest.py",) + assert report.scanned_scope_digest != report.expected_scope_digest + self_attested = replace( + report, + expected_scope_paths=report.scanned_scope_paths, + expected_scope_digest=report.scanned_scope_digest, + ) + for candidate in (report, self_attested): + with pytest.raises(LegacyBoundaryError, match="complete repository scope|authoritative"): + _require_cutover(candidate, frozen=frozen, candidate=_closed_inventory(), root=REPO) + + +def test_cutover_gate_remeasures_scope_and_rejects_forged_reports(tmp_path: Path) -> None: + frozen = load_migration_inventory(INVENTORY) + closed = _closed_inventory() + clean = _write(tmp_path, "clean.py", "VALUE = 1\n") + other = _write(tmp_path, "other.py", "OTHER = 2\n") + forbidden = _write(tmp_path, "forbidden.py", "import openultrasast.core as core_api\n") + report = inspect_legacy_boundary( + (clean,), + frozen_inventory=frozen, + candidate_inventory=closed, + expected_scope=(clean, other), + root=tmp_path, + phase=BoundaryPhase.POST_DELETION, + ) + self_attested = replace( + report, + expected_scope_paths=report.scanned_scope_paths, + expected_scope_digest=report.scanned_scope_digest, + ) + same_count_other = inspect_legacy_boundary( + (other,), + frozen_inventory=frozen, + candidate_inventory=closed, + expected_scope=(other,), + root=tmp_path, + phase=BoundaryPhase.POST_DELETION, + ) + forbidden_report = inspect_legacy_boundary( + (clean, forbidden), + frozen_inventory=frozen, + candidate_inventory=closed, + expected_scope=(clean, forbidden), + root=tmp_path, + phase=BoundaryPhase.POST_DELETION, + ) + findings_removed = replace(forbidden_report, forbidden_dynamic_sites=()) + + for forged, scope in ( + (self_attested, (clean, other)), + (same_count_other, (clean,)), + (findings_removed, (clean, forbidden)), + ): + with pytest.raises(LegacyBoundaryError, match="authoritative|complete repository scope"): + _require_cutover( + forged, + frozen=frozen, + candidate=closed, + root=tmp_path, + isolated_scope=scope, + ) + + +def test_repository_cutover_rejects_caller_scope_override() -> None: + frozen = load_migration_inventory(INVENTORY) + closed = _closed_inventory() + path = REPO / "tests/conftest.py" + report = inspect_legacy_boundary( + (path,), + frozen_inventory=frozen, + candidate_inventory=closed, + expected_scope=(path,), + root=REPO, + phase=BoundaryPhase.POST_DELETION, + ) + + with pytest.raises(LegacyBoundaryError, match="repository scope cannot be overridden"): + _require_cutover(report, frozen=frozen, candidate=closed, root=REPO, isolated_scope=(path,)) + + +def test_repository_scope_includes_owned_pair_tools_but_not_harvested_material() -> None: + relatives = {path.relative_to(REPO).as_posix() for path in boundary_python()} + + assert "benchmarks/pairs/catalog_gen.py" in relatives + assert "benchmarks/pairs/harvest.py" in relatives + assert "benchmarks/pairs/github/sven-cwe-078-os-system/fixed.py" not in relatives + assert "benchmarks/pairs/sast/owasp-python/BenchmarkTest00011.py" not in relatives + assert "benchmarks/fixtures/python-vulnerable/app.py" not in relatives