diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 0cd48cd..5d9eb61 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -51,6 +51,12 @@ repos: - mypy==2.2.0 pass_filenames: false always_run: true + - id: production-dependency-boundary + name: production dependency boundary contract + entry: python -B scripts/verify_production_dependencies.py + language: python + pass_filenames: false + always_run: true # Secret detection - repo: https://github.com/Yelp/detect-secrets diff --git a/scripts/verify_production_dependencies.py b/scripts/verify_production_dependencies.py new file mode 100644 index 0000000..648ef36 --- /dev/null +++ b/scripts/verify_production_dependencies.py @@ -0,0 +1,877 @@ +"""Verify the repository's source-level production dependency contract. + +The verifier deliberately uses only Git metadata and Python's standard-library +parser. It never imports analyzed modules. +""" + +from __future__ import annotations + +import argparse +import ast +import json +import re +import subprocess +import sys +import tokenize +from collections import defaultdict +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from datetime import date +from pathlib import Path, PurePosixPath +from typing import Any + +POLICY_PATH = "tests/architecture_dependency_policy.json" +MAX_FINDINGS = 50 +_TOP_LEVEL_KEYS = { + "schema_version", + "review", + "tracked_roots", + "domains", + "allowed_dependencies", + "compatibility_exceptions", + "accepted_cycles", + "dynamic_imports", +} +_REVIEW_KEYS = { + "owner", + "reviewed_at", + "next_review_by", + "static_analysis_policy_schema", +} +_EXCEPTION_KEYS = { + "importer", + "imported", + "owner", + "rationale", + "review_condition", +} +_CYCLE_KEYS = {"modules", "owner", "rationale", "review_condition"} +_DYNAMIC_KEYS = { + "path", + "scope", + "callee", + "target_kind", + "target", + "owner", + "rationale", + "review_condition", +} +_METADATA_KEYS = ("owner", "rationale", "review_condition") +_DOMAIN_RE = re.compile(r"^[a-z][a-z0-9_-]*$") + + +@dataclass(frozen=True, order=True) +class Finding: + """A deterministic, content-free policy finding.""" + + rule_id: str + path: str + line: int = 0 + subject: str = "" + + @property + def code(self) -> str: + """Compatibility alias for callers using diagnostic terminology.""" + + return self.rule_id + + @property + def identity(self) -> str: + """Return the bounded identity without exposing source content.""" + + return self.subject + + def render(self) -> str: + return render_findings((self,)) + + +@dataclass(frozen=True, order=True) +class DynamicImport: + path: str + scope: str + callee: str + target_kind: str + target: str + line: int = 0 + + @property + def identity(self) -> tuple[str, str, str, str, str]: + return ( + self.path, + self.scope, + self.callee, + self.target_kind, + self.target, + ) + + +@dataclass(frozen=True) +class Analysis: + owned_paths: tuple[str, ...] + static_edges: tuple[tuple[str, str], ...] + dynamic_imports: tuple[DynamicImport, ...] + cycles: tuple[tuple[str, ...], ...] + findings: tuple[Finding, ...] + + +@dataclass +class _PolicyContext: + tracked_files: set[str] + owned_paths: set[str] + path_domains: dict[str, str] + path_modules: dict[str, str] + module_paths: dict[str, str] + allowed_dependencies: dict[str, set[str]] + compatibility_exceptions: set[tuple[str, str]] + accepted_cycles: set[frozenset[str]] + dynamic_imports: dict[tuple[str, str, str, str, str], Mapping[str, Any]] + + +def _finding( + rule_id: str, path: str = ".", *, line: int = 0, subject: str = "" +) -> Finding: + return Finding(rule_id=rule_id, path=path, line=line, subject=subject) + + +def _safe_relative_path(value: Any) -> bool: + if not isinstance(value, str) or not value or "\\" in value: + return False + if re.match(r"^[A-Za-z]:", value): + return False + path = PurePosixPath(value) + return ( + not path.is_absolute() + and value == path.as_posix() + and "." not in path.parts + and ".." not in path.parts + ) + + +def _within_root(path: str, root: str) -> bool: + return path == root or path.startswith(f"{root.rstrip('/')}/") + + +def _module_name(path: str) -> str: + parts = list(PurePosixPath(path).with_suffix("").parts) + if parts[-1] == "__init__": + parts.pop() + return ".".join(parts) or "__init__" + + +def _tracked_python_files(repo_root: Path) -> tuple[set[str], list[Finding]]: + result = subprocess.run( + ["git", "ls-files", "--cached", "--", "*.py"], + cwd=repo_root, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=False, + shell=False, + ) + if result.returncode != 0: + return set(), [_finding("TRACKED_DISCOVERY_FAILED")] + return ( + { + line.strip().replace("\\", "/") + for line in result.stdout.splitlines() + if line.strip() + }, + [], + ) + + +def _validate_review_metadata( + entry: Mapping[str, Any], + *, + path: str, + findings: list[Finding], +) -> None: + if any( + not isinstance(entry.get(key), str) or not str(entry.get(key)).strip() + for key in _METADATA_KEYS + ): + findings.append(_finding("POLICY_REVIEW_METADATA", subject=path)) + + +def _validate_policy( + repo_root: Path, + policy: Mapping[str, Any], + tracked_files: Iterable[str] | None, +) -> tuple[_PolicyContext, list[Finding]]: + findings: list[Finding] = [] + unknown_keys = set(policy) - _TOP_LEVEL_KEYS + for key in sorted(unknown_keys): + findings.append(_finding("POLICY_UNKNOWN_KEY", subject=key)) + if policy.get("schema_version") != 1: + findings.append(_finding("POLICY_SCHEMA_VERSION")) + + review = policy.get("review") + if not isinstance(review, Mapping): + findings.append(_finding("POLICY_REVIEW_METADATA", subject="review")) + else: + for key in sorted(set(review) - _REVIEW_KEYS): + findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"review.{key}")) + if not isinstance(review.get("owner"), str) or not review["owner"].strip(): + findings.append(_finding("POLICY_REVIEW_METADATA", subject="review.owner")) + parsed_dates: dict[str, date] = {} + for key in ("reviewed_at", "next_review_by"): + try: + parsed_dates[key] = date.fromisoformat(str(review.get(key, ""))) + except ValueError: + findings.append( + _finding("POLICY_REVIEW_METADATA", subject=f"review.{key}") + ) + if ( + len(parsed_dates) == 2 + and parsed_dates["next_review_by"] < parsed_dates["reviewed_at"] + ): + findings.append( + _finding("POLICY_REVIEW_METADATA", subject="review.date_order") + ) + + if tracked_files is None: + discovered, discovery_findings = _tracked_python_files(repo_root) + findings.extend(discovery_findings) + else: + discovered = { + str(path).replace("\\", "/") + for path in tracked_files + if str(path).endswith(".py") + } + + roots_value = policy.get("tracked_roots") + roots = roots_value if isinstance(roots_value, list) else [] + if not isinstance(roots_value, list) or not roots: + findings.append(_finding("ROOTS_INVALID")) + valid_roots: list[str] = [] + seen_roots: set[str] = set() + for index, value in enumerate(roots): + subject = f"tracked_roots[{index}]" + if not _safe_relative_path(value): + findings.append(_finding("PATH_UNSAFE", subject=subject)) + continue + root = str(value) + if root in seen_roots: + findings.append(_finding("ROOT_DUPLICATE", path=root)) + continue + seen_roots.add(root) + valid_roots.append(root) + candidate = repo_root / root + try: + candidate.resolve().relative_to(repo_root.resolve()) + except ValueError: + findings.append(_finding("PATH_UNSAFE", path=root)) + continue + if not candidate.exists(): + findings.append(_finding("ROOT_MISSING", path=root)) + + domains_value = policy.get("domains") + domains = domains_value if isinstance(domains_value, Mapping) else {} + if not domains: + findings.append(_finding("DOMAINS_INVALID")) + valid_domain_names = { + str(name) + for name in domains + if isinstance(name, str) and _DOMAIN_RE.fullmatch(name) + } + for name in domains: + if name not in valid_domain_names: + findings.append(_finding("DOMAIN_UNKNOWN", subject=str(name))) + + owned_paths: set[str] = set() + path_domains: dict[str, str] = {} + path_modules: dict[str, str] = {} + module_paths: dict[str, str] = {} + for domain_name, entries in domains.items(): + if domain_name not in valid_domain_names: + continue + if not isinstance(entries, list): + findings.append(_finding("OWNERSHIP_INVALID", subject=str(domain_name))) + continue + for index, value in enumerate(entries): + subject = f"domains.{domain_name}[{index}]" + if not _safe_relative_path(value): + findings.append(_finding("PATH_UNSAFE", subject=subject)) + continue + path = str(value) + if not path.endswith(".py"): + findings.append(_finding("OWNERSHIP_INVALID", path=path)) + continue + if path in path_domains: + findings.append(_finding("OWN_DUPLICATE", path=path)) + continue + path_domains[path] = str(domain_name) + owned_paths.add(path) + if not any(_within_root(path, root) for root in valid_roots): + findings.append(_finding("OWN_OUTSIDE_ROOT", path=path)) + if path not in discovered: + findings.append(_finding("OWN_NOT_TRACKED", path=path)) + candidate = repo_root / path + try: + candidate.resolve().relative_to(repo_root.resolve()) + except ValueError: + findings.append(_finding("PATH_UNSAFE", path=path)) + continue + if not candidate.is_file(): + findings.append(_finding("OWN_MISSING", path=path)) + module = _module_name(path) + if module in module_paths: + findings.append( + _finding("OWN_MODULE_COLLISION", path=path, subject=module) + ) + else: + path_modules[path] = module + module_paths[module] = path + + tracked_in_roots = { + path + for path in discovered + if any(_within_root(path, root) for root in valid_roots) + } + for path in sorted(tracked_in_roots - owned_paths): + findings.append(_finding("OWN_UNOWNED_MODULE", path=path)) + + allowed_value = policy.get("allowed_dependencies") + allowed_raw = allowed_value if isinstance(allowed_value, Mapping) else {} + if not isinstance(allowed_value, Mapping): + findings.append(_finding("DEPENDENCIES_INVALID")) + for domain in sorted(valid_domain_names - set(allowed_raw)): + findings.append(_finding("DOMAIN_DIRECTION_MISSING", subject=domain)) + for domain in sorted(set(allowed_raw) - valid_domain_names): + findings.append(_finding("DOMAIN_UNKNOWN", subject=str(domain))) + allowed_dependencies: dict[str, set[str]] = {} + for domain in sorted(valid_domain_names): + values = allowed_raw.get(domain, []) + if not isinstance(values, list): + findings.append(_finding("DEPENDENCIES_INVALID", subject=domain)) + values = [] + accepted: set[str] = set() + for target in values: + if target not in valid_domain_names: + findings.append( + _finding( + "DOMAIN_UNKNOWN", + subject=f"{domain}->{target}", + ) + ) + else: + accepted.add(str(target)) + allowed_dependencies[domain] = accepted + + compatibility_exceptions: set[tuple[str, str]] = set() + exception_entries = policy.get("compatibility_exceptions") + if not isinstance(exception_entries, list): + findings.append(_finding("EXCEPTIONS_INVALID")) + exception_entries = [] + for index, entry in enumerate(exception_entries): + subject = f"compatibility_exceptions[{index}]" + if not isinstance(entry, Mapping): + findings.append(_finding("EXCEPTIONS_INVALID", subject=subject)) + continue + for key in sorted(set(entry) - _EXCEPTION_KEYS): + findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}")) + _validate_review_metadata(entry, path=subject, findings=findings) + edge = (str(entry.get("importer", "")), str(entry.get("imported", ""))) + if edge in compatibility_exceptions: + findings.append(_finding("DEP_DUPLICATE_EXCEPTION", subject=subject)) + compatibility_exceptions.add(edge) + if edge[0] not in module_paths or edge[1] not in module_paths: + findings.append(_finding("DEP_EXCEPTION_MODULE_UNKNOWN", subject=subject)) + + accepted_cycles: set[frozenset[str]] = set() + cycle_entries = policy.get("accepted_cycles") + if not isinstance(cycle_entries, list): + findings.append(_finding("CYCLES_INVALID")) + cycle_entries = [] + for index, entry in enumerate(cycle_entries): + subject = f"accepted_cycles[{index}]" + if not isinstance(entry, Mapping): + findings.append(_finding("CYCLES_INVALID", subject=subject)) + continue + for key in sorted(set(entry) - _CYCLE_KEYS): + findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}")) + _validate_review_metadata(entry, path=subject, findings=findings) + modules = entry.get("modules") + if ( + not isinstance(modules, list) + or len(modules) < 2 + or any(not isinstance(module, str) for module in modules) + ): + findings.append(_finding("CYCLES_INVALID", subject=subject)) + continue + cycle = frozenset(modules) + if len(cycle) != len(modules): + findings.append(_finding("CYCLE_DUPLICATE_MODULE", subject=subject)) + if cycle in accepted_cycles: + findings.append(_finding("CYCLE_DUPLICATE_BASELINE", subject=subject)) + accepted_cycles.add(cycle) + if any(module not in module_paths for module in cycle): + findings.append(_finding("CYCLE_MODULE_UNKNOWN", subject=subject)) + + dynamic_imports: dict[tuple[str, str, str, str, str], Mapping[str, Any]] = {} + dynamic_entries = policy.get("dynamic_imports") + if not isinstance(dynamic_entries, list): + findings.append(_finding("DYNAMIC_INVALID")) + dynamic_entries = [] + for index, entry in enumerate(dynamic_entries): + subject = f"dynamic_imports[{index}]" + if not isinstance(entry, Mapping): + findings.append(_finding("DYNAMIC_INVALID", subject=subject)) + continue + for key in sorted(set(entry) - _DYNAMIC_KEYS): + findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}")) + _validate_review_metadata(entry, path=subject, findings=findings) + path_value = entry.get("path") + if not _safe_relative_path(path_value): + findings.append(_finding("PATH_UNSAFE", subject=subject)) + continue + dynamic_path = str(path_value) + if dynamic_path not in owned_paths: + findings.append(_finding("DYNAMIC_PATH_UNOWNED", path=dynamic_path)) + target_kind = entry.get("target_kind") + identity = ( + dynamic_path, + str(entry.get("scope", "")), + str(entry.get("callee", "")), + str(target_kind), + str(entry.get("target", "")), + ) + if ( + not identity[1] + or identity[2] + not in {"__import__", "importlib.import_module", "import_module"} + or target_kind not in {"literal", "expression"} + or not identity[4] + ): + findings.append(_finding("DYNAMIC_INVALID", path=dynamic_path)) + if identity in dynamic_imports: + findings.append(_finding("DYNAMIC_DUPLICATE", path=dynamic_path)) + dynamic_imports[identity] = entry + + context = _PolicyContext( + tracked_files=discovered, + owned_paths=owned_paths, + path_domains=path_domains, + path_modules=path_modules, + module_paths=module_paths, + allowed_dependencies=allowed_dependencies, + compatibility_exceptions=compatibility_exceptions, + accepted_cycles=accepted_cycles, + dynamic_imports=dynamic_imports, + ) + return context, findings + + +def _resolve_relative_import( + current_module: str, + current_path: str, + node: ast.ImportFrom, +) -> str: + if not node.level: + return node.module or "" + is_package = current_path.endswith("/__init__.py") or current_path == "__init__.py" + if current_path == "__init__.py": + package_parts: list[str] = [] + else: + package_parts = ( + current_module.split(".") if is_package else current_module.split(".")[:-1] + ) + ascend = node.level - 1 + if ascend > len(package_parts): + prefix: list[str] = [] + elif ascend: + prefix = package_parts[:-ascend] + else: + prefix = package_parts + if node.module: + prefix.extend(node.module.split(".")) + return ".".join(prefix) + + +class _SourceVisitor(ast.NodeVisitor): + def __init__( + self, + *, + path: str, + module: str, + module_paths: Mapping[str, str], + ) -> None: + self.path = path + self.module = module + self.module_paths = module_paths + self.edges: set[tuple[str, str]] = set() + self.dynamic_imports: list[DynamicImport] = [] + self.scope: list[str] = [] + self.builtins_aliases: set[str] = {"builtins"} + self.builtin_import_aliases: set[str] = {"__import__"} + self.importlib_aliases: set[str] = {"importlib"} + self.import_module_aliases: set[str] = set() + + def _add_edge(self, imported: str) -> None: + # IMPORTANT: require an exact owned module. Falling back to the nearest + # package turns missing optional submodules into false dependency edges. + target = imported if imported in self.module_paths else None + if target and target != self.module: + self.edges.add((self.module, target)) + + def visit_Import(self, node: ast.Import) -> None: + for alias in node.names: + if alias.name == "builtins": + self.builtins_aliases.add(alias.asname or alias.name) + if alias.name == "importlib": + self.importlib_aliases.add(alias.asname or alias.name) + self._add_edge(alias.name) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + base = _resolve_relative_import(self.module, self.path, node) + if node.level == 0 and node.module == "importlib": + for alias in node.names: + if alias.name == "import_module": + self.import_module_aliases.add(alias.asname or alias.name) + if node.level == 0 and node.module == "builtins": + for alias in node.names: + if alias.name == "__import__": + self.builtin_import_aliases.add(alias.asname or alias.name) + exact_children: list[str] = [] + for alias in node.names: + candidate = f"{base}.{alias.name}" if base else alias.name + if candidate in self.module_paths: + exact_children.append(candidate) + if exact_children: + for candidate in exact_children: + self._add_edge(candidate) + elif base: + self._add_edge(base) + + def _visit_scoped( + self, + node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef, + ) -> None: + self.scope.append(node.name) + self.generic_visit(node) + self.scope.pop() + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + self._visit_scoped(node) + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + self._visit_scoped(node) + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self._visit_scoped(node) + + def visit_Call(self, node: ast.Call) -> None: + callee = "" + if isinstance(node.func, ast.Name): + if node.func.id in self.builtin_import_aliases: + callee = "__import__" + elif node.func.id in self.import_module_aliases: + callee = "import_module" + elif ( + isinstance(node.func, ast.Attribute) + and node.func.attr == "__import__" + and isinstance(node.func.value, ast.Name) + and node.func.value.id in self.builtins_aliases + ): + callee = "__import__" + elif ( + isinstance(node.func, ast.Attribute) + and node.func.attr == "import_module" + and isinstance(node.func.value, ast.Name) + and node.func.value.id in self.importlib_aliases + ): + callee = "importlib.import_module" + if callee: + if ( + node.args + and isinstance(node.args[0], ast.Constant) + and isinstance(node.args[0].value, str) + ): + target_kind = "literal" + target = node.args[0].value + elif node.args: + target_kind = "expression" + argument = node.args[0] + target = ( + argument.id + if isinstance(argument, ast.Name) + else f"<{type(argument).__name__}>" + ) + else: + target_kind = "expression" + target = "" + self.dynamic_imports.append( + DynamicImport( + path=self.path, + scope=".".join(self.scope) or "", + callee=callee, + target_kind=target_kind, + target=target, + line=node.lineno, + ) + ) + self.generic_visit(node) + + +def _strongly_connected_components( + modules: Iterable[str], + edges: Iterable[tuple[str, str]], +) -> tuple[tuple[str, ...], ...]: + adjacency: dict[str, set[str]] = defaultdict(set) + for importer, imported in edges: + adjacency[importer].add(imported) + next_index = 0 + indices: dict[str, int] = {} + low_links: dict[str, int] = {} + stack: list[str] = [] + on_stack: set[str] = set() + components: list[tuple[str, ...]] = [] + + def visit(module: str) -> None: + nonlocal next_index + indices[module] = next_index + low_links[module] = next_index + next_index += 1 + stack.append(module) + on_stack.add(module) + for imported in sorted(adjacency[module]): + if imported not in indices: + visit(imported) + low_links[module] = min(low_links[module], low_links[imported]) + elif imported in on_stack: + low_links[module] = min(low_links[module], indices[imported]) + if low_links[module] != indices[module]: + return + component: list[str] = [] + while True: + member = stack.pop() + on_stack.remove(member) + component.append(member) + if member == module: + break + if len(component) > 1: + components.append(tuple(sorted(component))) + + for module in sorted(modules): + if module not in indices: + visit(module) + return tuple(sorted(components)) + + +def analyze_repository( + repo_root: Path, + policy: Mapping[str, Any], + *, + tracked_files: Iterable[str] | None = None, +) -> Analysis: + """Analyze a repository without importing or executing its source modules.""" + + repo_root = repo_root.resolve() + context, findings = _validate_policy(repo_root, policy, tracked_files) + edges: set[tuple[str, str]] = set() + dynamic_imports: list[DynamicImport] = [] + for path in sorted(context.owned_paths): + source_path = repo_root / path + if not source_path.is_file() or path not in context.path_modules: + continue + try: + # IMPORTANT: tokenize.open handles encoding cookies and existing UTF-8 BOMs + # without rewriting source or importing production modules. + with tokenize.open(source_path) as source_file: + tree = ast.parse(source_file.read(), filename=path) + except (OSError, SyntaxError, UnicodeError) as exc: + findings.append( + _finding("SOURCE_PARSE", path=path, subject=type(exc).__name__) + ) + continue + visitor = _SourceVisitor( + path=path, + module=context.path_modules[path], + module_paths=context.module_paths, + ) + visitor.visit(tree) + edges.update(visitor.edges) + dynamic_imports.extend(visitor.dynamic_imports) + + for importer, imported in sorted(edges): + importer_path = context.module_paths.get(importer, ".") + importer_domain = context.path_domains.get(importer_path) + imported_path = context.module_paths.get(imported, ".") + imported_domain = context.path_domains.get(imported_path) + if not importer_domain or not imported_domain: + continue + allowed = imported_domain in context.allowed_dependencies.get( + importer_domain, set() + ) + exception = (importer, imported) in context.compatibility_exceptions + if not allowed and not exception: + findings.append( + _finding( + "DEP_FORBIDDEN_DIRECTION", + path=importer_path, + subject=f"{importer}->{imported}", + ) + ) + for importer, imported in sorted(context.compatibility_exceptions): + if (importer, imported) not in edges: + path = context.module_paths.get(importer, ".") + findings.append( + _finding( + "DEP_STALE_EXCEPTION", + path=path, + subject=f"{importer}->{imported}", + ) + ) + + cycles = _strongly_connected_components(context.module_paths, edges) + current_cycle_sets = {frozenset(cycle) for cycle in cycles} + for cycle in cycles: + if frozenset(cycle) not in context.accepted_cycles: + path = context.module_paths.get(cycle[0], ".") + findings.append(_finding("CYCLE_NEW", path=path, subject="|".join(cycle))) + for accepted_cycle in sorted( + context.accepted_cycles, key=lambda item: sorted(item) + ): + if accepted_cycle not in current_cycle_sets: + first = sorted(accepted_cycle)[0] if accepted_cycle else "" + path = context.module_paths.get(first, ".") + findings.append( + _finding( + "CYCLE_STALE", + path=path, + subject="|".join(sorted(accepted_cycle)), + ) + ) + + current_dynamic = {site.identity: site for site in dynamic_imports} + for identity, site in sorted(current_dynamic.items()): + if identity not in context.dynamic_imports: + rule_id = ( + "DYNAMIC_UNREGISTERED_LITERAL" + if site.target_kind == "literal" + else "DYNAMIC_UNREGISTERED_EXPRESSION" + ) + findings.append( + _finding( + rule_id, + path=site.path, + line=site.line, + subject=f"{site.scope}:{site.callee}", + ) + ) + for identity in sorted(context.dynamic_imports): + if identity not in current_dynamic: + path, scope, callee, _, _ = identity + findings.append( + _finding( + "DYNAMIC_STALE", + path=path, + subject=f"{scope}:{callee}", + ) + ) + + return Analysis( + owned_paths=tuple(sorted(context.owned_paths)), + static_edges=tuple(sorted(edges)), + dynamic_imports=tuple(sorted(dynamic_imports)), + cycles=cycles, + findings=tuple(sorted(set(findings))), + ) + + +def verify_repository( + repo_root: Path, + policy: Mapping[str, Any], + *, + tracked_files: Iterable[str] | None = None, +) -> tuple[Finding, ...]: + return analyze_repository( + repo_root, + policy, + tracked_files=tracked_files, + ).findings + + +def evaluate_repository( + repo_root: Path, + policy: Mapping[str, Any], + *, + tracked_files: Iterable[str] | None = None, +) -> list[Finding]: + """Compatibility facade returning the deterministic findings as a list.""" + + return list(verify_repository(repo_root, policy, tracked_files=tracked_files)) + + +def render_findings(findings: Sequence[Finding]) -> str: + lines: list[str] = [] + for finding in sorted(findings): + location = finding.path + if finding.line: + location = f"{location}:{finding.line}" + # Security boundary: CLI output is limited to rule IDs and repository-relative + # locations. Internal graph identities remain available to in-process tests. + lines.append(f"{finding.rule_id} {location}") + return "\n".join(lines) + + +def _load_policy(path: Path) -> Mapping[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, Mapping): + raise ValueError("policy root must be an object") + return value + + +def run_cli( + repo_root: Path, + policy_path: Path, + *, + max_findings: int = MAX_FINDINGS, +) -> tuple[int, list[str]]: + """Run the bounded CLI contract without printing or leaking host paths.""" + + repo_root = repo_root.resolve() + policy_path = policy_path if policy_path.is_absolute() else repo_root / policy_path + try: + policy_path.resolve().relative_to(repo_root) + except ValueError: + return 2, ["POLICY_PATH_OUTSIDE ."] + try: + policy = _load_policy(policy_path) + except (OSError, UnicodeError, json.JSONDecodeError, ValueError): + return 2, [f"POLICY_JSON_INVALID {POLICY_PATH}"] + + findings = verify_repository(repo_root, policy) + if not findings: + return 0, ["DEPENDENCY_POLICY_PASS ."] + + limit = max(1, min(int(max_findings), MAX_FINDINGS)) + visible = findings[:limit] + lines = [finding.render() for finding in visible] + omitted = len(findings) - len(visible) + if omitted: + lines.append(f"FINDINGS_TRUNCATED - {omitted} omitted") + return 1, lines + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo-root", type=Path, default=Path(__file__).parents[1]) + parser.add_argument("--policy", type=Path) + parser.add_argument("--max-findings", type=int, default=MAX_FINDINGS) + args = parser.parse_args(argv) + repo_root = args.repo_root.resolve() + policy_path = args.policy or (repo_root / POLICY_PATH) + exit_code, lines = run_cli( + repo_root, + policy_path, + max_findings=args.max_findings, + ) + for line in lines: + print(line) + return exit_code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/architecture_dependency_policy.json b/tests/architecture_dependency_policy.json new file mode 100644 index 0000000..be17ef4 --- /dev/null +++ b/tests/architecture_dependency_policy.json @@ -0,0 +1,543 @@ +{ + "schema_version": 1, + "review": { + "owner": "architecture-maintainers", + "reviewed_at": "2026-07-31", + "next_review_by": "2026-10-31", + "static_analysis_policy_schema": 1 + }, + "tracked_roots": [ + "__init__.py", + "config.py", + "api", + "connector", + "models", + "nodes", + "services", + "scripts" + ], + "domains": { + "api": [ + "api/__init__.py", + "api/approvals.py", + "api/assist.py", + "api/bridge.py", + "api/capabilities.py", + "api/checkpoints_handler.py", + "api/config.py", + "api/config_llm_handlers.py", + "api/config_model_handlers.py", + "api/config_projection_handlers.py", + "api/connector_contracts.py", + "api/errors.py", + "api/events.py", + "api/model_manager.py", + "api/packs.py", + "api/pnginfo.py", + "api/preflight_handler.py", + "api/presets.py", + "api/remote_admin.py", + "api/rewrite_recipes.py", + "api/route_handlers.py", + "api/route_orchestration.py", + "api/route_registrars.py", + "api/routes.py", + "api/schedules.py", + "api/secrets.py", + "api/security_doctor.py", + "api/templates.py", + "api/tools.py", + "api/triggers.py", + "api/webhook.py", + "api/webhook_submit.py", + "api/webhook_validate.py" + ], + "connector": [ + "connector/__init__.py", + "connector/__main__.py", + "connector/channels/__init__.py", + "connector/channels/kakaotalk.py", + "connector/command_firewall.py", + "connector/config.py", + "connector/contract.py", + "connector/jobs_summary.py", + "connector/llm_client.py", + "connector/media_response.py", + "connector/media_store.py", + "connector/openclaw_client.py", + "connector/platforms/__init__.py", + "connector/platforms/discord_gateway.py", + "connector/platforms/feishu_delivery_handlers.py", + "connector/platforms/feishu_ingress_handlers.py", + "connector/platforms/feishu_installation_handlers.py", + "connector/platforms/feishu_installation_manager.py", + "connector/platforms/feishu_long_connection.py", + "connector/platforms/feishu_webhook.py", + "connector/platforms/kakao_webhook.py", + "connector/platforms/line_webhook.py", + "connector/platforms/slack_delivery_handlers.py", + "connector/platforms/slack_ingress_handlers.py", + "connector/platforms/slack_installation_handlers.py", + "connector/platforms/slack_installation_manager.py", + "connector/platforms/slack_socket_mode.py", + "connector/platforms/slack_webhook.py", + "connector/platforms/telegram_polling.py", + "connector/platforms/wechat_webhook.py", + "connector/platforms/whatsapp_webhook.py", + "connector/prompts.py", + "connector/rate_limiter.py", + "connector/reply_visibility.py", + "connector/results_poller.py", + "connector/router.py", + "connector/router_admin_handlers.py", + "connector/router_chat_handlers.py", + "connector/router_dispatch.py", + "connector/router_execution_handlers.py", + "connector/security_profile.py", + "connector/semantic_guard.py", + "connector/state.py", + "connector/transport_contract.py" + ], + "models": [ + "models/__init__.py", + "models/schemas.py" + ], + "nodes": [ + "nodes/__init__.py", + "nodes/batch_variants.py", + "nodes/image_to_prompt.py", + "nodes/portability_contract.py", + "nodes/prompt_planner.py", + "nodes/prompt_refiner.py" + ], + "root": [ + "__init__.py", + "config.py" + ], + "scripts": [ + "scripts/check_deployment_profile.py", + "scripts/check_openapi_sync.py", + "scripts/check_supply_chain_hardening.py", + "scripts/compatibility_matrix_refresh.py", + "scripts/contract_digest.py", + "scripts/devtools/debug_s35_import.py", + "scripts/devtools/verify_s30_doctor.py", + "scripts/generate_openapi_spec.py", + "scripts/generate_provenance.py", + "scripts/lint_implementation_record.py", + "scripts/openclaw_smoke_import.py", + "scripts/operator_doctor.py", + "scripts/precommit_black_single.py", + "scripts/precommit_block_sensitive_files.py", + "scripts/preflight_check.py", + "scripts/quality_governance_common.py", + "scripts/regenerate_openapi_if_needed.py", + "scripts/registry_publish_guard.py", + "scripts/report_coverage_governance.py", + "scripts/run_adversarial_gate.py", + "scripts/run_backend_coverage.py", + "scripts/run_crypto_lifecycle_drills.py", + "scripts/run_mutation_test.py", + "scripts/run_unittests.py", + "scripts/start_sidecar.py", + "scripts/verify_api_config_contract.py", + "scripts/verify_api_route_contract.py", + "scripts/verify_audit_chain.py", + "scripts/verify_connector_router_contract.py", + "scripts/verify_exception_boundary_policy.py", + "scripts/verify_platform_adapter_contract.py", + "scripts/verify_production_dependencies.py", + "scripts/verify_provenance.py", + "scripts/verify_quality_governance.py", + "scripts/verify_static_analysis_policy.py", + "scripts/verify_test_debt_governance.py" + ], + "services": [ + "services/__init__.py", + "services/access_control.py", + "services/aiohttp_compat.py", + "services/approvals/__init__.py", + "services/approvals/models.py", + "services/approvals/service.py", + "services/approvals/storage.py", + "services/async_utils.py", + "services/audit.py", + "services/audit_events.py", + "services/audit_pipeline.py", + "services/automation_composer.py", + "services/bridge_handshake.py", + "services/bridge_token_lifecycle.py", + "services/cache/__init__.py", + "services/cache/ttl_cache.py", + "services/callback_delivery.py", + "services/capabilities.py", + "services/chatops/__init__.py", + "services/chatops/network_errors.py", + "services/chatops/retry.py", + "services/chatops/session_scope.py", + "services/chatops/targets.py", + "services/chatops/transport_contract.py", + "services/chatops/webhook_adapter.py", + "services/checkpoints.py", + "services/comfyui_history.py", + "services/compatibility_matrix_governance.py", + "services/config_layers.py", + "services/connector_allowlist_posture.py", + "services/connector_callback_contract.py", + "services/connector_extraction_contract.py", + "services/connector_installation_registry.py", + "services/connector_replay_lifecycle.py", + "services/constrained_transforms.py", + "services/control_plane.py", + "services/control_plane_adapter.py", + "services/crypto_lifecycle_drills.py", + "services/csrf_protection.py", + "services/delivery/__init__.py", + "services/delivery/http_callback.py", + "services/delivery/router.py", + "services/deployment_profile.py", + "services/diagnostics_flags.py", + "services/effective_config.py", + "services/endpoint_manifest.py", + "services/execution_budgets.py", + "services/failover.py", + "services/idempotency_store.py", + "services/image_utils.py", + "services/import_fallback.py", + "services/integrity.py", + "services/internal_content.py", + "services/job_events.py", + "services/jobs_read_model.py", + "services/jobs_security.py", + "services/legacy_compat.py", + "services/llm_client.py", + "services/llm_model_list.py", + "services/llm_output.py", + "services/log_tail.py", + "services/management_query.py", + "services/metrics.py", + "services/model_manager.py", + "services/model_manager_catalog.py", + "services/model_manager_tasks.py", + "services/model_manager_transfer.py", + "services/modules.py", + "services/observability/backpressure.py", + "services/openapi_generation.py", + "services/operator_doctor.py", + "services/operator_guidance.py", + "services/package_hygiene.py", + "services/packs/pack_archive.py", + "services/packs/pack_manifest.py", + "services/packs/pack_registry.py", + "services/packs/pack_types.py", + "services/parameter_lab.py", + "services/paths.py", + "services/permission_posture.py", + "services/planner.py", + "services/planner_registry.py", + "services/plugins/__init__.py", + "services/plugins/async_bridge.py", + "services/plugins/builtin/__init__.py", + "services/plugins/builtin/audit_log.py", + "services/plugins/builtin/model_alias.py", + "services/plugins/builtin/params_clamp.py", + "services/plugins/contract.py", + "services/plugins/manager.py", + "services/pnginfo.py", + "services/policy_posture.py", + "services/preflight.py", + "services/presets/__init__.py", + "services/presets/models.py", + "services/presets/storage.py", + "services/product_boundary.py", + "services/provider_errors.py", + "services/providers/__init__.py", + "services/providers/anthropic.py", + "services/providers/catalog.py", + "services/providers/keys.py", + "services/providers/openai_compat.py", + "services/queue_submit.py", + "services/rate_limit.py", + "services/reasoning_redaction.py", + "services/redaction.py", + "services/refiner.py", + "services/registry.py", + "services/registry_quarantine.py", + "services/request_contracts.py", + "services/request_ip.py", + "services/retry_after.py", + "services/retry_partition.py", + "services/rewrite_recipes.py", + "services/route_bootstrap.py", + "services/route_bootstrap_contract.py", + "services/runtime_config.py", + "services/runtime_config_policy.py", + "services/runtime_config_projection.py", + "services/runtime_config_store.py", + "services/runtime_dependency_hygiene.py", + "services/runtime_guardrails.py", + "services/runtime_lifecycle.py", + "services/runtime_profile.py", + "services/safe_io.py", + "services/scheduler/__init__.py", + "services/scheduler/delivery_contract.py", + "services/scheduler/history.py", + "services/scheduler/models.py", + "services/scheduler/runner.py", + "services/scheduler/storage.py", + "services/schema_sanitizer.py", + "services/secret_providers.py", + "services/secret_store.py", + "services/secrets_encryption.py", + "services/security_advisories.py", + "services/security_doctor.py", + "services/security_doctor_checks.py", + "services/security_doctor_connector_checks.py", + "services/security_doctor_endpoint_checks.py", + "services/security_doctor_remediation.py", + "services/security_doctor_report.py", + "services/security_doctor_runner.py", + "services/security_doctor_runtime_checks.py", + "services/security_gate.py", + "services/security_invariants.py", + "services/security_telemetry.py", + "services/settings_schema.py", + "services/sidecar/__init__.py", + "services/sidecar/auth.py", + "services/sidecar/bridge_client.py", + "services/sidecar/bridge_contract.py", + "services/sidecar/runtime.py", + "services/sidecar_secret_refs.py", + "services/startup_lifecycle.py", + "services/startup_profile_gate.py", + "services/state_dir.py", + "services/structured_logging.py", + "services/surface_guard.py", + "services/templates.py", + "services/tenant_context.py", + "services/threat_intel_gate.py", + "services/threat_intel_provider.py", + "services/tool_calling.py", + "services/tool_runner.py", + "services/trace.py", + "services/trace_store.py", + "services/transform_common.py", + "services/transform_runner.py", + "services/transform_worker.py", + "services/webhook_auth.py", + "services/webhook_mapping.py", + "services/workflow_portability.py" + ] + }, + "allowed_dependencies": { + "api": [ + "api", + "models", + "root", + "services" + ], + "connector": [ + "connector", + "services" + ], + "models": [ + "models" + ], + "nodes": [ + "models", + "nodes", + "services" + ], + "root": [ + "nodes", + "root", + "services" + ], + "scripts": [ + "api", + "connector", + "models", + "nodes", + "root", + "scripts", + "services" + ], + "services": [ + "models", + "root", + "services" + ] + }, + "compatibility_exceptions": [ + { + "importer": "models.schemas", + "imported": "services.request_contracts", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.connector_callback_contract", + "imported": "connector.config", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.connector_callback_contract", + "imported": "connector.security_profile", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.connector_callback_contract", + "imported": "connector.transport_contract", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.preflight", + "imported": "nodes", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.queue_submit", + "imported": "api.errors", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.sidecar.runtime", + "imported": "connector.config", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.sidecar.runtime", + "imported": "connector.openclaw_client", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + }, + { + "importer": "services.workflow_portability", + "imported": "nodes.portability_contract", + "owner": "architecture-maintainers", + "rationale": "Existing boundary retained for compatibility while domain ownership is reduced.", + "review_condition": "Remove when the importer no longer requires the higher-level domain." + } + ], + "accepted_cycles": [ + { + "modules": [ + "config", + "services.effective_config", + "services.runtime_config", + "services.runtime_config_policy", + "services.safe_io" + ], + "owner": "architecture-maintainers", + "rationale": "Existing runtime configuration and safe-I/O bootstrap cycle.", + "review_condition": "Remove when the cycle is decomposed into one-way contracts." + }, + { + "modules": [ + "services.access_control", + "services.audit_events", + "services.reasoning_redaction", + "services.security_telemetry" + ], + "owner": "architecture-maintainers", + "rationale": "Existing security telemetry, audit, access, and redaction cycle.", + "review_condition": "Remove when the cycle is decomposed into one-way contracts." + } + ], + "dynamic_imports": [ + { + "path": "api/routes.py", + "scope": "check_dependency", + "callee": "__import__", + "target_kind": "expression", + "target": "module_name", + "owner": "architecture-maintainers", + "rationale": "Optional dependency availability probe.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + }, + { + "path": "scripts/openclaw_smoke_import.py", + "scope": "main", + "callee": "__import__", + "target_kind": "expression", + "target": "candidate", + "owner": "architecture-maintainers", + "rationale": "Package/top-level smoke import candidate.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + }, + { + "path": "scripts/openclaw_smoke_import.py", + "scope": "test_import", + "callee": "__import__", + "target_kind": "expression", + "target": "module_name", + "owner": "architecture-maintainers", + "rationale": "Smoke-test module import boundary.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + }, + { + "path": "services/import_fallback.py", + "scope": "import_module_dual", + "callee": "importlib.import_module", + "target_kind": "expression", + "target": "absolute_module", + "owner": "architecture-maintainers", + "rationale": "Top-level fallback half of dual import compatibility.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + }, + { + "path": "services/import_fallback.py", + "scope": "import_module_dual", + "callee": "importlib.import_module", + "target_kind": "expression", + "target": "relative_module", + "owner": "architecture-maintainers", + "rationale": "Package-relative half of dual import compatibility.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + }, + { + "path": "services/jobs_read_model.py", + "scope": "_resolve_get_all_jobs", + "callee": "importlib.import_module", + "target_kind": "literal", + "target": "comfy_execution.jobs", + "owner": "architecture-maintainers", + "rationale": "Optional ComfyUI jobs host capability.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + }, + { + "path": "services/operator_doctor.py", + "scope": "check_core_imports", + "callee": "importlib.import_module", + "target_kind": "expression", + "target": "mod_name", + "owner": "architecture-maintainers", + "rationale": "Operator-doctor import health probe.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + }, + { + "path": "services/operator_doctor.py", + "scope": "check_pre_commit", + "callee": "importlib.import_module", + "target_kind": "literal", + "target": "pre_commit", + "owner": "architecture-maintainers", + "rationale": "Optional quality-tool availability probe.", + "review_condition": "Remove or update when the dynamic loading boundary changes." + } + ] +} diff --git a/tests/test_architecture_dependency_policy.py b/tests/test_architecture_dependency_policy.py new file mode 100644 index 0000000..a3ab121 --- /dev/null +++ b/tests/test_architecture_dependency_policy.py @@ -0,0 +1,337 @@ +from __future__ import annotations + +import copy +import json +import tempfile +import unittest +from pathlib import Path + +from scripts import verify_production_dependencies as dependency_policy + + +class ArchitecturePolicyFixture(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.repo_root = Path(self.temp_dir.name) + self._write("app/__init__.py", "") + self._write("app/main.py", "from core.util import VALUE\n") + self._write("core/__init__.py", "") + self._write("core/util.py", "VALUE = 1\n") + + def tearDown(self) -> None: + self.temp_dir.cleanup() + + def _write(self, relative_path: str, content: str, *, bom: bool = False) -> None: + path = self.repo_root / relative_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8-sig" if bom else "utf-8") + + def _tracked_files(self) -> tuple[str, ...]: + return tuple( + sorted( + path.relative_to(self.repo_root).as_posix() + for path in self.repo_root.rglob("*.py") + ) + ) + + @staticmethod + def _review_metadata() -> dict[str, str]: + return { + "owner": "architecture-maintainers", + "rationale": "Temporary compatibility boundary.", + "review_condition": "Remove when the importing module is moved.", + } + + def _policy(self) -> dict: + return { + "schema_version": 1, + "review": { + "owner": "architecture-maintainers", + "reviewed_at": "2026-07-31", + "next_review_by": "2026-10-31", + }, + "tracked_roots": ["app", "core"], + "domains": { + "app": ["app/__init__.py", "app/main.py"], + "core": ["core/__init__.py", "core/util.py"], + }, + "allowed_dependencies": { + "app": ["app", "core"], + "core": ["core"], + }, + "compatibility_exceptions": [], + "accepted_cycles": [], + "dynamic_imports": [], + } + + def _verify(self, policy: dict | None = None): + return dependency_policy.verify_repository( + self.repo_root, + policy or self._policy(), + tracked_files=self._tracked_files(), + ) + + def _codes(self, policy: dict | None = None) -> set[str]: + return {finding.rule_id for finding in self._verify(policy)} + + def test_allowed_direction_passes(self): + self.assertEqual(self._verify(), ()) + + def test_forbidden_direction_fails_with_stable_rule(self): + self._write("core/util.py", "from app.main import VALUE\n") + + self.assertIn("DEP_FORBIDDEN_DIRECTION", self._codes()) + + def test_exact_compatibility_exception_passes_and_stale_entry_fails(self): + self._write("app/main.py", "VALUE = 1\n") + self._write("core/util.py", "from app.main import VALUE\n") + policy = self._policy() + policy["compatibility_exceptions"] = [ + { + "importer": "core.util", + "imported": "app.main", + **self._review_metadata(), + } + ] + self.assertEqual(self._verify(policy), ()) + + self._write("core/util.py", "VALUE = 1\n") + self.assertIn("DEP_STALE_EXCEPTION", self._codes(policy)) + + def test_new_cycle_and_stale_accepted_cycle_fail(self): + self._write("core/util.py", "from app.main import VALUE\n") + policy = self._policy() + policy["allowed_dependencies"]["core"].append("app") + self.assertIn("CYCLE_NEW", self._codes(policy)) + + policy["accepted_cycles"] = [ + { + "modules": ["app.main", "core.util"], + **self._review_metadata(), + } + ] + self.assertEqual(self._verify(policy), ()) + + self._write("core/util.py", "VALUE = 1\n") + self.assertIn("CYCLE_STALE", self._codes(policy)) + + def test_literal_and_expression_dynamic_imports_require_exact_registration(self): + self._write( + "app/main.py", + "import importlib\n\n" + "def load_literal():\n" + ' return importlib.import_module("optional_plugin")\n\n' + "def load_expression(module_name):\n" + " return importlib.import_module(module_name)\n", + ) + self.assertEqual( + self._codes(), + { + "DYNAMIC_UNREGISTERED_EXPRESSION", + "DYNAMIC_UNREGISTERED_LITERAL", + }, + ) + + policy = self._policy() + policy["dynamic_imports"] = [ + { + "path": "app/main.py", + "scope": "load_expression", + "callee": "importlib.import_module", + "target_kind": "expression", + "target": "module_name", + **self._review_metadata(), + }, + { + "path": "app/main.py", + "scope": "load_literal", + "callee": "importlib.import_module", + "target_kind": "literal", + "target": "optional_plugin", + **self._review_metadata(), + }, + ] + self.assertEqual(self._verify(policy), ()) + + self._write("app/main.py", "VALUE = 1\n") + self.assertIn("DYNAMIC_STALE", self._codes(policy)) + + def test_builtin_and_importlib_aliases_cannot_bypass_dynamic_registration(self): + self._write( + "app/main.py", + "import builtins as runtime_builtins\n" + "from importlib import import_module as load_module\n\n" + "def load(module_name):\n" + " runtime_builtins.__import__(module_name)\n" + ' return load_module("optional_plugin")\n', + ) + + self.assertEqual( + self._codes(), + { + "DYNAMIC_UNREGISTERED_EXPRESSION", + "DYNAMIC_UNREGISTERED_LITERAL", + }, + ) + + def test_dual_package_and_top_level_imports_resolve_to_one_owned_module(self): + self._write( + "app/main.py", + "try:\n" + " from ..core.util import VALUE\n" + "except ImportError:\n" + " from core.util import VALUE\n", + ) + + analysis = dependency_policy.analyze_repository( + self.repo_root, + self._policy(), + tracked_files=self._tracked_files(), + ) + + self.assertEqual(analysis.static_edges, (("app.main", "core.util"),)) + self.assertEqual(analysis.findings, ()) + + def test_unowned_new_tracked_module_fails(self): + self._write("core/new_module.py", "VALUE = 2\n") + + self.assertIn("OWN_UNOWNED_MODULE", self._codes()) + + def test_policy_rejects_missing_unsafe_duplicate_and_unknown_ownership(self): + policy = self._policy() + policy["tracked_roots"].extend(["missing", "../outside"]) + policy["domains"]["core"].append("app/main.py") + policy["allowed_dependencies"]["app"].append("unknown-domain") + + codes = self._codes(policy) + + self.assertTrue( + { + "ROOT_MISSING", + "PATH_UNSAFE", + "OWN_DUPLICATE", + "DOMAIN_UNKNOWN", + }.issubset(codes) + ) + + def test_policy_rejects_unknown_keys_and_incomplete_review_metadata(self): + policy = self._policy() + policy["PRIVATE_POLICY_SENTINEL"] = True + policy["compatibility_exceptions"] = [ + { + "importer": "core.util", + "imported": "app.main", + "owner": "", + "rationale": "fixture", + "review_condition": "", + } + ] + + codes = self._codes(policy) + + self.assertIn("POLICY_UNKNOWN_KEY", codes) + self.assertIn("POLICY_REVIEW_METADATA", codes) + self.assertNotIn( + "PRIVATE_POLICY_SENTINEL", + dependency_policy.render_findings(self._verify(policy)), + ) + + def test_python_encoding_detection_accepts_utf8_bom_without_rewriting(self): + self._write("core/util.py", "VALUE = 1\n", bom=True) + before = (self.repo_root / "core" / "util.py").read_bytes() + + self.assertEqual(self._verify(), ()) + self.assertEqual((self.repo_root / "core" / "util.py").read_bytes(), before) + + def test_source_is_parsed_without_execution_or_source_content_disclosure(self): + marker = self.repo_root / "executed.txt" + secret = "PRIVATE_SOURCE_SENTINEL" + self._write( + "app/main.py", + "from pathlib import Path\n" + f'Path({str(marker)!r}).write_text("{secret}")\n' + "import importlib\n" + 'importlib.import_module("unregistered")\n', + ) + before = { + path.relative_to(self.repo_root).as_posix(): path.read_bytes() + for path in self.repo_root.rglob("*") + if path.is_file() + } + + findings = self._verify() + rendered = dependency_policy.render_findings(findings) + after = { + path.relative_to(self.repo_root).as_posix(): path.read_bytes() + for path in self.repo_root.rglob("*") + if path.is_file() + } + + self.assertFalse(marker.exists()) + self.assertEqual(after, before) + self.assertNotIn(secret, rendered) + self.assertNotIn(str(self.repo_root), rendered) + self.assertIn("DYNAMIC_UNREGISTERED_LITERAL", rendered) + self.assertIn("app/main.py", rendered) + + +class RepositoryArchitecturePolicyTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.repo_root = Path(__file__).resolve().parents[1] + cls.policy_path = ( + cls.repo_root / "tests" / "architecture_dependency_policy.json" + ) + + def test_repository_policy_is_canonical_and_current(self): + policy = json.loads(self.policy_path.read_text(encoding="utf-8")) + + analysis = dependency_policy.analyze_repository(self.repo_root, policy) + + self.assertEqual(analysis.findings, ()) + self.assertEqual(len(analysis.owned_paths), 297) + self.assertEqual(len(policy["accepted_cycles"]), 2) + self.assertEqual(len(policy["dynamic_imports"]), 8) + self.assertEqual(len(policy["compatibility_exceptions"]), 9) + + def test_policy_change_does_not_weaken_static_analysis_governance(self): + static_policy_path = self.repo_root / "tests" / "static_analysis_policy.json" + static_policy = json.loads(static_policy_path.read_text(encoding="utf-8")) + architecture_policy = json.loads(self.policy_path.read_text(encoding="utf-8")) + + self.assertEqual( + architecture_policy["tracked_roots"], + static_policy["production_roots"], + ) + self.assertEqual( + architecture_policy["review"]["static_analysis_policy_schema"], + static_policy["schema_version"], + ) + + def test_precommit_runs_the_dependency_verifier_without_runtime_dependencies(self): + config = (self.repo_root / ".pre-commit-config.yaml").read_text( + encoding="utf-8" + ) + hook = config.split("- id: production-dependency-boundary", 1)[1].split( + "# Secret detection", 1 + )[0] + + self.assertIn("scripts/verify_production_dependencies.py", hook) + self.assertIn("language: python", hook) + self.assertIn("pass_filenames: false", hook) + self.assertIn("always_run: true", hook) + self.assertNotIn("additional_dependencies", hook) + + def test_repository_verifier_is_deterministic(self): + policy = json.loads(self.policy_path.read_text(encoding="utf-8")) + + first = dependency_policy.verify_repository(self.repo_root, policy) + second = dependency_policy.verify_repository( + self.repo_root, copy.deepcopy(policy) + ) + + self.assertEqual(first, second) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_production_dependency_policy.py b/tests/test_production_dependency_policy.py new file mode 100644 index 0000000..25b21e6 --- /dev/null +++ b/tests/test_production_dependency_policy.py @@ -0,0 +1,417 @@ +import json +import subprocess +import sys +import tempfile +import unittest +from copy import deepcopy +from pathlib import Path + +from scripts import verify_production_dependencies as verifier + + +class ProductionDependencyFixture: + def __init__(self, root: Path): + self.root = root + + @staticmethod + def policy() -> dict: + return { + "schema_version": 1, + "review": { + "owner": "maintainers", + "reviewed_at": "2026-07-31", + "next_review_by": "2026-10-31", + }, + "tracked_roots": ["root.py", "alpha", "beta"], + "domains": { + "root": ["root.py"], + "alpha": [ + "alpha/__init__.py", + "alpha/api.py", + "alpha/helper.py", + ], + "beta": ["beta/__init__.py", "beta/adapter.py"], + }, + "allowed_dependencies": { + "root": ["alpha", "root"], + "alpha": ["alpha", "beta"], + "beta": ["beta"], + }, + "compatibility_exceptions": [], + "accepted_cycles": [], + "dynamic_imports": [], + } + + def write( + self, + files: dict[str, str], + *, + policy: dict | None = None, + bom_paths: set[str] | None = None, + ) -> dict: + payload = deepcopy(policy or self.policy()) + for relative, content in files.items(): + path = self.root / relative + path.parent.mkdir(parents=True, exist_ok=True) + encoding = "utf-8-sig" if relative in (bom_paths or set()) else "utf-8" + path.write_text(content, encoding=encoding) + (self.root / "policy.json").write_text( + json.dumps(payload, indent=2) + "\n", encoding="utf-8" + ) + subprocess.run( + ["git", "init", "--quiet"], + cwd=self.root, + check=True, + capture_output=True, + ) + subprocess.run( + ["git", "add", "."], + cwd=self.root, + check=True, + capture_output=True, + ) + return payload + + +def metadata(reason: str = "temporary compatibility debt") -> dict[str, str]: + return { + "owner": "maintainers", + "rationale": reason, + "review_condition": "remove after the dependency direction is corrected", + } + + +class TestProductionDependencyPolicy(unittest.TestCase): + def _base_files(self) -> dict[str, str]: + return { + "root.py": "from alpha import api\n", + "alpha/__init__.py": "", + "alpha/api.py": "from beta import adapter\n", + "alpha/helper.py": "VALUE = 1\n", + "beta/__init__.py": "", + "beta/adapter.py": "VALUE = 2\n", + } + + def _evaluate( + self, + files: dict[str, str] | None = None, + *, + configure=None, + bom_paths: set[str] | None = None, + ): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + fixture = ProductionDependencyFixture(root) + policy = fixture.policy() + if configure is not None: + configure(policy) + fixture.write( + files or self._base_files(), policy=policy, bom_paths=bom_paths + ) + findings = verifier.evaluate_repository(root, policy) + return findings + + def assertCodes(self, findings, *codes: str) -> None: + self.assertEqual([finding.rule_id for finding in findings], list(codes)) + + def test_allowed_direction_and_dual_import_mode_pass_without_importing_source(self): + files = self._base_files() + files[ + "alpha/dual.py" + ] = """ +try: + from . import helper +except ImportError: + from alpha import helper + +raise RuntimeError("production modules must never be imported by the verifier") +""" + + def own_dual(policy): + policy["domains"]["alpha"].append("alpha/dual.py") + + findings = self._evaluate(files, configure=own_dual) + + self.assertEqual(findings, []) + + def test_missing_submodule_does_not_collapse_to_an_owned_parent_package(self): + files = self._base_files() + files["alpha/api.py"] = "VALUE = 1\n" + files["alpha/probe.py"] = "import beta.missing.redaction\n" + + def exact_only(policy): + policy["domains"]["alpha"].append("alpha/probe.py") + policy["allowed_dependencies"]["alpha"] = ["alpha"] + + findings = self._evaluate(files, configure=exact_only) + + self.assertEqual(findings, []) + + def test_forbidden_direction_has_stable_path_and_rule_code(self): + files = self._base_files() + files["beta/reverse.py"] = "from alpha import api\n" + + def own_reverse(policy): + policy["domains"]["beta"].append("beta/reverse.py") + + findings = self._evaluate(files, configure=own_reverse) + + self.assertCodes(findings, "DEP_FORBIDDEN_DIRECTION") + self.assertEqual(findings[0].path, "beta/reverse.py") + self.assertEqual(findings[0].identity, "beta.reverse->alpha.api") + + def test_exact_compatibility_exception_passes_and_stale_entry_fails(self): + files = self._base_files() + files["alpha/api.py"] = "VALUE = 1\n" + files["beta/reverse.py"] = "from alpha import api\n" + + def accepted(policy): + policy["domains"]["beta"].append("beta/reverse.py") + policy["compatibility_exceptions"] = [ + { + "importer": "beta.reverse", + "imported": "alpha.api", + **metadata(), + } + ] + + self.assertEqual(self._evaluate(files, configure=accepted), []) + + files["beta/reverse.py"] = "VALUE = 3\n" + findings = self._evaluate(files, configure=accepted) + self.assertCodes(findings, "DEP_STALE_EXCEPTION") + + def test_new_cycle_and_stale_accepted_cycle_are_rejected(self): + files = self._base_files() + files["beta/adapter.py"] = "from alpha import api\n" + + def allow_reverse(policy): + policy["allowed_dependencies"]["beta"].append("alpha") + + findings = self._evaluate(files, configure=allow_reverse) + self.assertCodes(findings, "CYCLE_NEW") + + def accept_cycle(policy): + policy["allowed_dependencies"]["beta"].append("alpha") + policy["accepted_cycles"] = [ + { + "modules": ["alpha.api", "beta.adapter"], + **metadata("reviewed fixture cycle"), + } + ] + + self.assertEqual(self._evaluate(files, configure=accept_cycle), []) + files["beta/adapter.py"] = "VALUE = 2\n" + findings = self._evaluate(files, configure=accept_cycle) + self.assertCodes(findings, "CYCLE_STALE") + + def test_dynamic_literal_and_expression_require_exact_registration(self): + files = self._base_files() + files[ + "alpha/dynamic.py" + ] = """ +import importlib + +def load(name): + importlib.import_module("external.literal") + return __import__(name) +""" + + def own_dynamic(policy): + policy["domains"]["alpha"].append("alpha/dynamic.py") + + findings = self._evaluate(files, configure=own_dynamic) + self.assertCodes( + findings, + "DYNAMIC_UNREGISTERED_EXPRESSION", + "DYNAMIC_UNREGISTERED_LITERAL", + ) + + def register(policy): + policy["domains"]["alpha"].append("alpha/dynamic.py") + policy["dynamic_imports"] = [ + { + "path": "alpha/dynamic.py", + "scope": "load", + "callee": "__import__", + "target_kind": "expression", + "target": "name", + **metadata("runtime-selected module fixture"), + }, + { + "path": "alpha/dynamic.py", + "scope": "load", + "callee": "importlib.import_module", + "target_kind": "literal", + "target": "external.literal", + **metadata("optional external module fixture"), + }, + ] + + self.assertEqual(self._evaluate(files, configure=register), []) + files["alpha/dynamic.py"] = "VALUE = 1\n" + findings = self._evaluate(files, configure=register) + self.assertCodes(findings, "DYNAMIC_STALE", "DYNAMIC_STALE") + + def test_policy_validation_rejects_unsafe_missing_duplicate_and_unknown_ownership( + self, + ): + def invalid(policy): + policy["unexpected"] = True + policy["tracked_roots"].append("missing") + policy["domains"]["root"].append("../outside.py") + policy["domains"]["beta"].append("alpha/api.py") + policy["allowed_dependencies"]["alpha"].append("ghost") + + findings = self._evaluate(configure=invalid) + codes = {finding.code for finding in findings} + + self.assertTrue( + { + "POLICY_UNKNOWN_KEY", + "ROOT_MISSING", + "PATH_UNSAFE", + "OWN_DUPLICATE", + "DOMAIN_UNKNOWN", + }.issubset(codes) + ) + + def test_tracked_module_in_accepted_root_must_be_owned(self): + files = self._base_files() + files["orphan/tool.py"] = "VALUE = 1\n" + + def unowned(policy): + policy["tracked_roots"].append("orphan") + + findings = self._evaluate(files, configure=unowned) + + self.assertCodes(findings, "OWN_UNOWNED_MODULE") + self.assertEqual(findings[0].path, "orphan/tool.py") + + def test_bom_source_is_parsed_and_source_content_is_not_reported(self): + files = self._base_files() + files["beta/reverse.py"] = ( + "# PRIVATE_SOURCE_MARKER_MUST_NOT_APPEAR\nfrom alpha import api\n" + ) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + fixture = ProductionDependencyFixture(root) + policy = fixture.policy() + policy["domains"]["beta"].append("beta/reverse.py") + fixture.write( + files, + policy=policy, + bom_paths={"beta/reverse.py"}, + ) + result = subprocess.run( + [ + sys.executable, + str( + Path(__file__).resolve().parents[1] + / "scripts" + / "verify_production_dependencies.py" + ), + "--repo-root", + str(root), + "--policy", + "policy.json", + ], + cwd=root, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=False, + ) + + self.assertEqual(result.returncode, 1) + self.assertIn("DEP_FORBIDDEN_DIRECTION beta/reverse.py", result.stdout) + self.assertNotIn("PRIVATE_SOURCE_MARKER", result.stdout) + self.assertNotIn(str(root), result.stdout) + self.assertNotIn(str(root).replace("\\", "/"), result.stdout) + + def test_malformed_source_reports_only_bounded_content_free_identity(self): + files = self._base_files() + files["alpha/broken.py"] = "PRIVATE_PAYLOAD = '''unterminated\n" + + def own_broken(policy): + policy["domains"]["alpha"].append("alpha/broken.py") + + findings = self._evaluate(files, configure=own_broken) + + self.assertCodes(findings, "SOURCE_PARSE") + self.assertEqual(findings[0].path, "alpha/broken.py") + self.assertEqual(findings[0].identity, "SyntaxError") + self.assertNotIn("PRIVATE_PAYLOAD", findings[0].render()) + + def test_cli_bounds_many_findings_and_reports_truncation(self): + files = self._base_files() + for index in range(verifier.MAX_FINDINGS + 5): + files[f"beta/reverse_{index:02d}.py"] = "from alpha import api\n" + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + fixture = ProductionDependencyFixture(root) + policy = fixture.policy() + policy["domains"]["beta"].extend( + f"beta/reverse_{index:02d}.py" + for index in range(verifier.MAX_FINDINGS + 5) + ) + fixture.write(files, policy=policy) + exit_code, lines = verifier.run_cli( + root, root / "policy.json", max_findings=verifier.MAX_FINDINGS + ) + + self.assertEqual(exit_code, 1) + self.assertEqual(len(lines), verifier.MAX_FINDINGS + 1) + self.assertTrue(lines[-1].startswith("FINDINGS_TRUNCATED - ")) + + +class TestRepositoryProductionDependencyPolicy(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.repo_root = Path(__file__).resolve().parents[1] + cls.policy_path = ( + cls.repo_root / "tests" / "architecture_dependency_policy.json" + ) + + def test_repository_policy_is_complete_and_current(self): + policy = json.loads(self.policy_path.read_text(encoding="utf-8")) + + findings = verifier.evaluate_repository(self.repo_root, policy) + + self.assertEqual(findings, []) + self.assertEqual( + policy["tracked_roots"], + [ + "__init__.py", + "config.py", + "api", + "connector", + "models", + "nodes", + "services", + "scripts", + ], + ) + self.assertEqual(len(policy["accepted_cycles"]), 2) + self.assertEqual(len(policy["dynamic_imports"]), 8) + + def test_precommit_invokes_the_dependency_verifier(self): + content = (self.repo_root / ".pre-commit-config.yaml").read_text( + encoding="utf-8" + ) + hook = content.split("- id: production-dependency-boundary", 1)[1].split( + "# Secret detection", 1 + )[0] + + self.assertIn("scripts/verify_production_dependencies.py", hook) + self.assertIn("language: python", hook) + self.assertIn("pass_filenames: false", hook) + self.assertIn("always_run: true", hook) + self.assertNotIn("additional_dependencies", hook) + + +if __name__ == "__main__": + unittest.main()