From 8956c5c5f5a201cf2fb556c3be43a50f19af97a3 Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 14:30:05 -0400 Subject: [PATCH] Add deterministic Python reference adapter --- src/docforge/adapters/__init__.py | 15 + src/docforge/adapters/python.py | 830 ++++++++++++++++++ .../reference-python/src/sample/__init__.py | 7 + .../reference-python/src/sample/service.py | 18 + .../reference-python/src/sample/shared.py | 11 + .../reference-python/src/sample/worker.py | 8 + tests/test_python_reference_adapter.py | 267 ++++++ 7 files changed, 1156 insertions(+) create mode 100644 src/docforge/adapters/__init__.py create mode 100644 src/docforge/adapters/python.py create mode 100644 tests/fixtures/reference-python/src/sample/__init__.py create mode 100644 tests/fixtures/reference-python/src/sample/service.py create mode 100644 tests/fixtures/reference-python/src/sample/shared.py create mode 100644 tests/fixtures/reference-python/src/sample/worker.py create mode 100644 tests/test_python_reference_adapter.py diff --git a/src/docforge/adapters/__init__.py b/src/docforge/adapters/__init__.py new file mode 100644 index 0000000..489a70b --- /dev/null +++ b/src/docforge/adapters/__init__.py @@ -0,0 +1,15 @@ +"""Repository-owned reference adapters built on the public adapter SDK.""" + +from .python import ( + PYTHON_ADAPTER_VERSION, + PYTHON_EXTRACTOR_VERSION, + PythonReferenceAdapter, + PythonUnsupportedFact, +) + +__all__ = [ + "PYTHON_ADAPTER_VERSION", + "PYTHON_EXTRACTOR_VERSION", + "PythonReferenceAdapter", + "PythonUnsupportedFact", +] diff --git a/src/docforge/adapters/python.py b/src/docforge/adapters/python.py new file mode 100644 index 0000000..ef4349c --- /dev/null +++ b/src/docforge/adapters/python.py @@ -0,0 +1,830 @@ +"""Deterministic stdlib-AST reference adapter for explicitly confined Python roots. + +The adapter parses Python source as untrusted data. It never imports or executes +project code. Its intentionally narrow dependency model publishes only imports +that resolve to another module in the same declared source inventory. +""" + +from __future__ import annotations + +import ast +import hashlib +import io +import os +import stat +import tokenize +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from pathlib import Path, PurePosixPath + +from ..adapter_sdk import ( + AdapterAssembly, + AdapterEdge, + AdapterManifest, + AdapterNode, + AdapterProjection, + AdapterSource, + AdapterSourceProjection, + Edge, + Node, +) +from ..errors import DocForgeError +from ..python_logic import PythonLogicOwner, analyze_python_source + +PYTHON_ADAPTER_ID = "docforge.reference.python" +PYTHON_ADAPTER_VERSION = "1" +PYTHON_EXTRACTOR_VERSION = "stdlib-ast@1" +PYTHON_IDENTITY_VERSION = "python-reference-id@1" +PYTHON_SUPPORT_SCHEMA_VERSION = 1 + +_SUPPORTED_FACTS = ( + "python_file", + "python_module", + "python_class", + "python_function", + "lexical_containment", + "local_import_dependency", + "function_logic", +) + + +@dataclass(frozen=True) +class PythonUnsupportedFact: + """One semantic fact this syntax-only reference adapter does not claim.""" + + code: str + description: str + + def as_dict(self) -> dict[str, str]: + return {"code": self.code, "description": self.description} + + +_UNSUPPORTED_FACTS = ( + PythonUnsupportedFact( + "call_resolution", + "Calls are represented only inside function Logic and are not resolved to symbols.", + ), + PythonUnsupportedFact( + "dynamic_import_resolution", + "Imports performed through runtime calls are not dependency evidence.", + ), + PythonUnsupportedFact( + "inheritance_resolution", + "Class bases are syntax only and are not resolved to local or external types.", + ), + PythonUnsupportedFact( + "runtime_generated_facts", + "Decorators, metaclasses, descriptors, and executed module code are never evaluated.", + ), + PythonUnsupportedFact( + "symbol_reference_resolution", + "Imported names, variable references, types, overloads, and re-exports are not resolved.", + ), +) + + +@dataclass(frozen=True) +class _SourceRecord: + source_id: str + source_path: str + module_name: str + fingerprint: str + raw: bytes + text: str + tree: ast.Module + dependencies: tuple[str, ...] = () + + +@dataclass(frozen=True) +class _Definition: + kind: str + qualified_name: str + parent_node_id: str + node: ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef + node_id: str + + +class PythonReferenceAdapter: + """Reference Python adapter over explicit, non-overlapping source roots.""" + + def __init__( + self, + root: Path, + *, + source_roots: Iterable[str | Path], + project_id: str = "python-reference", + title: str = "Python reference project", + max_sources: int = 4_096, + max_source_bytes: int = 1_000_000, + max_logic_nodes_per_function: int = 2_000, + ) -> None: + resolved_root = root.resolve(strict=True) + if not resolved_root.is_dir(): + raise DocForgeError("invalid_adapter", "Python adapter root must be a directory") + if max_sources < 1 or max_source_bytes < 1 or max_logic_nodes_per_function < 2: + raise ValueError("Python adapter limits must be positive") + self.root = resolved_root + self.project_id = project_id + self.title = title + self.max_sources = max_sources + self.max_source_bytes = max_source_bytes + self.max_logic_nodes_per_function = max_logic_nodes_per_function + self.source_roots = self._normalize_source_roots(source_roots) + + def support_report(self) -> dict[str, object]: + """Return deterministic, machine-readable scope and limitation evidence.""" + + return { + "schema_version": PYTHON_SUPPORT_SCHEMA_VERSION, + "adapter_id": PYTHON_ADAPTER_ID, + "adapter_version": PYTHON_ADAPTER_VERSION, + "extractor_version": PYTHON_EXTRACTOR_VERSION, + "identity_version": PYTHON_IDENTITY_VERSION, + "frontend": "python-stdlib-ast", + "imports_project_code": False, + "executes_project_code": False, + "source_roots": list(self.source_roots), + "supported_facts": list(_SUPPORTED_FACTS), + "unsupported_facts": [fact.as_dict() for fact in _UNSUPPORTED_FACTS], + } + + def unsupported_facts(self) -> tuple[PythonUnsupportedFact, ...]: + """Return the adapter's fixed unsupported semantic fact inventory.""" + + return _UNSUPPORTED_FACTS + + def load_manifest(self) -> AdapterManifest: + records = self._inventory() + module_sources = {record.module_name: record.source_id for record in records} + if len(module_sources) != len(records): + raise DocForgeError( + "ambiguous_python_module", + "Declared Python source roots produce duplicate module names", + ) + sources = tuple( + sorted( + ( + AdapterSource( + source_id=record.source_id, + source_path=record.source_path, + fingerprint=record.fingerprint, + extractor_version=PYTHON_EXTRACTOR_VERSION, + dependencies=self._local_dependencies(record, module_sources), + ) + for record in records + ), + key=lambda source: source.source_id, + ) + ) + source_hash = self._source_hash(sources) + return AdapterManifest( + project_id=self.project_id, + title=self.title, + adapter_id=PYTHON_ADAPTER_ID, + adapter_version=PYTHON_ADAPTER_VERSION, + root=self.root, + revision=source_hash[:12], + source_hash=source_hash, + families=("code",), + allowed_relations=("contains", "depends_on"), + sources=sources, + estimated_nodes=max(1, len(sources) * 8), + ) + + def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: + record = self._read_source(source.source_path) + if record.source_id != source.source_id or record.fingerprint != source.fingerprint: + raise DocForgeError( + "stale_adapter_source", + "Python source changed after its manifest was captured", + source=source.source_path, + ) + return self._extract_record(record, source) + + def assemble_projection( + self, + manifest: AdapterManifest, + contributions: tuple[AdapterSourceProjection, ...], + ) -> AdapterAssembly: + expected = tuple(source.source_id for source in manifest.sources) + actual = tuple(sorted(contribution.source_id for contribution in contributions)) + if actual != expected: + raise DocForgeError( + "invalid_adapter", + "Python assembly contributions do not match the current manifest", + ) + nodes = tuple( + sorted( + (node for contribution in contributions for node in contribution.nodes), + key=lambda item: item.node.node_id, + ) + ) + edges = tuple( + sorted( + (edge for contribution in contributions for edge in contribution.edges), + key=lambda item: ( + item.edge.source_id, + item.edge.relation, + item.edge.target_id, + ), + ) + ) + logic = tuple( + sorted( + (projection for contribution in contributions for projection in contribution.logic), + key=lambda projection: projection.owner_node_id, + ) + ) + return AdapterAssembly( + projection=AdapterProjection( + project_id=manifest.project_id, + title=manifest.title, + adapter_id=manifest.adapter_id, + adapter_version=manifest.adapter_version, + root=manifest.root, + revision=manifest.revision, + source_hash=manifest.source_hash, + nodes=nodes, + edges=edges, + ), + logic=logic, + ) + + def load_assembly(self) -> AdapterAssembly: + """Load one cache-independent complete graph and Logic assembly.""" + + manifest = self.load_manifest() + contributions = tuple(self.extract_source(source) for source in manifest.sources) + stable = self.load_manifest() + if stable != manifest: + raise DocForgeError( + "source_changed", + "Python sources changed during complete adapter extraction", + ) + return self.assemble_projection(manifest, contributions) + + def load_complete_assembly(self) -> AdapterAssembly: + """Implement the SDK's independent complete-assembly oracle.""" + + return self.load_assembly() + + def load_projection(self) -> AdapterProjection: + """Preserve the first-class legacy complete-projection contract.""" + + return self.load_assembly().projection + + def _normalize_source_roots(self, source_roots: Iterable[str | Path]) -> tuple[str, ...]: + normalized: list[str] = [] + for value in source_roots: + candidate = PurePosixPath(Path(value).as_posix()) + if ( + candidate.is_absolute() + or not candidate.parts + or ".." in candidate.parts + or str(candidate) in {"", "."} + ): + raise DocForgeError( + "path_escape", + "Python source roots must be explicit project-relative directories", + ) + relative = candidate.as_posix() + absolute = self.root.joinpath(*candidate.parts) + if absolute.is_symlink(): + raise DocForgeError("path_escape", "Python source root cannot be a symbolic link") + try: + resolved = absolute.resolve(strict=True) + except OSError as error: + raise DocForgeError( + "invalid_adapter", + "Python source root does not exist", + source_root=relative, + ) from error + if ( + resolved != absolute + or not resolved.is_dir() + or not resolved.is_relative_to(self.root) + ): + raise DocForgeError( + "path_escape", + "Python source root must be a confined real directory", + source_root=relative, + ) + normalized.append(relative) + result = tuple(sorted(set(normalized))) + if not result: + raise DocForgeError("invalid_adapter", "At least one Python source root is required") + if len(result) != len(normalized): + raise DocForgeError("invalid_adapter", "Python source roots must be unique") + paths = [PurePosixPath(value) for value in result] + for index, left in enumerate(paths): + for right in paths[index + 1 :]: + if left in right.parents or right in left.parents: + raise DocForgeError( + "invalid_adapter", + "Python source roots must not overlap", + ) + return result + + def _inventory(self) -> tuple[_SourceRecord, ...]: + paths: list[str] = [] + for source_root in self.source_roots: + absolute_root = self.root.joinpath(*PurePosixPath(source_root).parts) + if ( + absolute_root.is_symlink() + or not absolute_root.is_dir() + or absolute_root.resolve(strict=True) != absolute_root + ): + raise DocForgeError( + "path_escape", + "Python source root changed after adapter binding", + source_root=source_root, + ) + for directory, directory_names, file_names in os.walk( + absolute_root, + topdown=True, + followlinks=False, + ): + current = Path(directory) + for name in tuple(directory_names): + child = current / name + if child.is_symlink(): + raise DocForgeError( + "path_escape", + "Python source inventory contains a symbolic-link directory", + source=child.relative_to(self.root).as_posix(), + ) + directory_names[:] = sorted( + name for name in directory_names if name != "__pycache__" + ) + for name in sorted(file_names): + if not name.endswith(".py"): + continue + child = current / name + relative = child.relative_to(self.root).as_posix() + if child.is_symlink(): + raise DocForgeError( + "path_escape", + "Python source inventory contains a symbolic-link file", + source=relative, + ) + paths.append(relative) + if len(paths) > self.max_sources: + raise DocForgeError( + "adapter_too_large", + "Python source inventory exceeds the configured limit", + maximum=self.max_sources, + ) + records = tuple(self._read_source(path) for path in sorted(paths)) + modules = [record.module_name for record in records] + if len(modules) != len(set(modules)): + duplicates = sorted(module for module in set(modules) if modules.count(module) > 1) + raise DocForgeError( + "ambiguous_python_module", + "Declared Python source roots produce duplicate module names", + modules=duplicates, + ) + return records + + def _read_source(self, relative: str) -> _SourceRecord: + path = PurePosixPath(relative) + if path.is_absolute() or ".." in path.parts or path.suffix != ".py": + raise DocForgeError("path_escape", "Python source path is unsafe") + source_root = self._source_root_for(path) + candidate = self.root.joinpath(*path.parts) + try: + before = candidate.lstat() + except OSError as error: + raise DocForgeError( + "stale_adapter_source", + "Python source is unavailable", + source=relative, + ) from error + if ( + stat.S_ISLNK(before.st_mode) + or not stat.S_ISREG(before.st_mode) + or candidate.resolve(strict=True) != candidate + or not candidate.is_relative_to(self.root) + ): + raise DocForgeError( + "path_escape", + "Python source must be a confined regular file", + source=relative, + ) + if before.st_size > self.max_source_bytes: + raise DocForgeError( + "source_too_large", + "Python source exceeds the configured adapter limit", + source=relative, + maximum=self.max_source_bytes, + ) + descriptor = -1 + try: + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(candidate, flags) + opened = os.fstat(descriptor) + if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != ( + before.st_dev, + before.st_ino, + ): + raise DocForgeError( + "path_escape", + "Python source identity changed before it was opened", + source=relative, + ) + with os.fdopen(descriptor, "rb") as handle: + descriptor = -1 + raw = handle.read(self.max_source_bytes + 1) + after_open = os.fstat(handle.fileno()) + after = candidate.lstat() + except DocForgeError: + raise + except OSError as error: + raise DocForgeError( + "stale_adapter_source", + "Python source changed while being read", + source=relative, + ) from error + finally: + if descriptor >= 0: + os.close(descriptor) + identity_before = ( + before.st_dev, + before.st_ino, + before.st_size, + before.st_mtime_ns, + ) + identity_after = ( + after.st_dev, + after.st_ino, + after.st_size, + after.st_mtime_ns, + ) + identity_opened = ( + after_open.st_dev, + after_open.st_ino, + after_open.st_size, + after_open.st_mtime_ns, + ) + if ( + len(raw) > self.max_source_bytes + or identity_before != identity_opened + or identity_before != identity_after + or len(raw) != after.st_size + ): + raise DocForgeError( + "stale_adapter_source", + "Python source changed while being read", + source=relative, + ) + text = self._decode_source(raw, relative) + try: + tree = ast.parse(text, filename=relative, type_comments=True) + except SyntaxError as error: + raise DocForgeError( + "invalid_python_source", + "Python source cannot be parsed by the reference adapter", + source=relative, + line=error.lineno, + ) from error + module_name = self._module_name(path, PurePosixPath(source_root)) + return _SourceRecord( + source_id=self._source_id(relative), + source_path=relative, + module_name=module_name, + fingerprint=hashlib.sha256(raw).hexdigest(), + raw=raw, + text=text, + tree=tree, + ) + + def _source_root_for(self, path: PurePosixPath) -> str: + matches = [ + source_root + for source_root in self.source_roots + if path == PurePosixPath(source_root) or PurePosixPath(source_root) in path.parents + ] + if len(matches) != 1: + raise DocForgeError( + "path_escape", + "Python source is outside the declared source roots", + source=path.as_posix(), + ) + return matches[0] + + @staticmethod + def _decode_source(raw: bytes, source_path: str) -> str: + try: + encoding, _ = tokenize.detect_encoding(io.BytesIO(raw).readline) + return raw.decode(encoding) + except (LookupError, SyntaxError, UnicodeDecodeError) as error: + raise DocForgeError( + "invalid_python_source", + "Python source encoding is invalid", + source=source_path, + ) from error + + @staticmethod + def _module_name(path: PurePosixPath, source_root: PurePosixPath) -> str: + local = path.relative_to(source_root) + parts = list(local.with_suffix("").parts) + if parts and parts[-1] == "__init__": + parts.pop() + if not parts: + parts = list(source_root.parts) + return ".".join(parts) + + @staticmethod + def _source_id(relative: str) -> str: + digest = hashlib.sha256( + f"{PYTHON_IDENTITY_VERSION}\0source\0{relative}".encode() + ).hexdigest()[:24] + return f"python.source.{digest}" + + @staticmethod + def _file_node_id(source_id: str) -> str: + return source_id.replace("python.source.", "python.file.", 1) + + @staticmethod + def _module_node_id(source_id: str) -> str: + return source_id.replace("python.source.", "python.module.", 1) + + @staticmethod + def _symbol_node_id(kind: str, source_id: str, qualified_name: str) -> str: + digest = hashlib.sha256( + (f"{PYTHON_IDENTITY_VERSION}\0{kind}\0{source_id}\0{qualified_name}").encode() + ).hexdigest()[:24] + return f"python.{kind}.{digest}" + + @staticmethod + def _source_hash(sources: Sequence[AdapterSource]) -> str: + digest = hashlib.sha256() + digest.update(PYTHON_ADAPTER_VERSION.encode()) + digest.update(PYTHON_EXTRACTOR_VERSION.encode()) + digest.update(PYTHON_IDENTITY_VERSION.encode()) + for source in sources: + digest.update(b"\0source\0") + digest.update(source.source_id.encode()) + digest.update(b"\0path\0") + digest.update(source.source_path.encode()) + digest.update(b"\0fingerprint\0") + digest.update(source.fingerprint.encode()) + for dependency in source.dependencies: + digest.update(b"\0dependency\0") + digest.update(dependency.encode()) + return digest.hexdigest() + + def _local_dependencies( + self, + record: _SourceRecord, + module_sources: dict[str, str], + ) -> tuple[str, ...]: + dependencies: set[str] = set() + package = ( + record.module_name + if record.source_path.endswith("/__init__.py") + else record.module_name.rpartition(".")[0] + ) + for node in ast.walk(record.tree): + candidates: list[str] = [] + if isinstance(node, ast.Import): + candidates.extend(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + base = self._import_from_base(package, node.level, node.module) + if base: + candidates.append(base) + if node.module is None and base: + candidates.extend( + f"{base}.{alias.name}" for alias in node.names if alias.name != "*" + ) + for module_name in candidates: + target = module_sources.get(module_name) + if target is not None and target != record.source_id: + dependencies.add(target) + return tuple(sorted(dependencies)) + + @staticmethod + def _import_from_base(package: str, level: int, module: str | None) -> str: + if level == 0: + return module or "" + package_parts = package.split(".") if package else [] + keep = len(package_parts) - (level - 1) + if keep < 0: + return "" + prefix = package_parts[:keep] + if module: + prefix.extend(module.split(".")) + return ".".join(prefix) + + def _extract_record( + self, + record: _SourceRecord, + source: AdapterSource, + ) -> AdapterSourceProjection: + file_id = self._file_node_id(source.source_id) + module_id = self._module_node_id(source.source_id) + nodes: list[AdapterNode] = [ + self._node( + node_id=file_id, + title=record.source_path, + kind="file", + qualified_name=record.source_path, + content=record.text, + source_path=record.source_path, + anchor="L1", + ), + self._node( + node_id=module_id, + title=record.module_name, + kind="module", + qualified_name=record.module_name, + content=( + ast.get_docstring(record.tree, clean=False) + or f"Python module {record.module_name}." + ), + source_path=record.source_path, + anchor="L1", + ), + ] + edges: list[AdapterEdge] = [self._edge(file_id, "contains", module_id, "syntax")] + definitions = self._definitions(record, module_id) + owners: list[PythonLogicOwner] = [] + for definition in definitions: + content = ast.get_source_segment(record.text, definition.node) + if content is None or not content.strip(): + content = f"Python {definition.kind} {definition.qualified_name}." + anchor = f"L{definition.node.lineno}" + nodes.append( + self._node( + node_id=definition.node_id, + title=definition.qualified_name, + kind=definition.kind, + qualified_name=f"{record.module_name}.{definition.qualified_name}", + content=content, + source_path=record.source_path, + anchor=anchor, + asynchronous=isinstance(definition.node, ast.AsyncFunctionDef), + ) + ) + edges.append( + self._edge( + definition.parent_node_id, + "contains", + definition.node_id, + "syntax", + anchor=anchor, + ) + ) + if definition.kind == "function": + owners.append( + PythonLogicOwner( + owner_node_id=definition.node_id, + qualified_name=definition.qualified_name, + line=definition.node.lineno, + ) + ) + for dependency in source.dependencies: + edges.append( + self._edge( + module_id, + "depends_on", + self._module_node_id(dependency), + "local_import", + ) + ) + logic = analyze_python_source( + record.text, + source_id=source.source_id, + owners=owners, + filename=record.source_path, + max_nodes_per_function=self.max_logic_nodes_per_function, + ) + return AdapterSourceProjection( + source_id=source.source_id, + fingerprint=source.fingerprint, + nodes=tuple(sorted(nodes, key=lambda item: item.node.node_id)), + edges=tuple( + sorted( + edges, + key=lambda item: ( + item.edge.source_id, + item.edge.relation, + item.edge.target_id, + ), + ) + ), + logic=logic, + ) + + def _definitions( + self, + record: _SourceRecord, + module_node_id: str, + ) -> tuple[_Definition, ...]: + definitions: list[_Definition] = [] + qualified_names: set[str] = set() + adapter = self + + class Collector(ast.NodeVisitor): + def __init__(self) -> None: + self.scope_names: list[str] = [] + self.scope_node_ids: list[str] = [module_node_id] + + def _definition( + self, + kind: str, + node: ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + qualified_name = ".".join((*self.scope_names, node.name)) + if qualified_name in qualified_names: + raise DocForgeError( + "ambiguous_python_symbol", + "Python source repeats a class or function identity", + source=record.source_path, + qualified_name=qualified_name, + ) + qualified_names.add(qualified_name) + node_id = adapter._symbol_node_id( + kind, + record.source_id, + qualified_name, + ) + definitions.append( + _Definition( + kind=kind, + qualified_name=qualified_name, + parent_node_id=self.scope_node_ids[-1], + node=node, + node_id=node_id, + ) + ) + self.scope_names.append(node.name) + self.scope_node_ids.append(node_id) + self.generic_visit(node) + self.scope_node_ids.pop() + self.scope_names.pop() + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self._definition("class", node) + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + self._definition("function", node) + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + self._definition("function", node) + + Collector().visit(record.tree) + return tuple(sorted(definitions, key=lambda item: item.node_id)) + + @staticmethod + def _node( + *, + node_id: str, + title: str, + kind: str, + qualified_name: str, + content: str, + source_path: str, + anchor: str, + asynchronous: bool = False, + ) -> AdapterNode: + normalized = content.strip() or f"Python {kind} {qualified_name}." + tags = tuple(sorted({"python", kind, *(("async",) if asynchronous else ())})) + return AdapterNode( + node=Node( + node_id=node_id, + title=title, + family="code", + authority="derived", + status="active", + tags=tags, + summary=f"Python {kind} fact for {qualified_name}.", + content=normalized, + source_path=source_path, + source_anchor=anchor, + content_hash=hashlib.sha256(normalized.encode()).hexdigest(), + ), + metadata=( + ("extractor", PYTHON_EXTRACTOR_VERSION), + ("identity", PYTHON_IDENTITY_VERSION), + ("kind", kind), + ("qualified_name", qualified_name), + ), + ) + + @staticmethod + def _edge( + source_id: str, + relation: str, + target_id: str, + evidence: str, + *, + anchor: str | None = None, + ) -> AdapterEdge: + metadata = [("evidence", evidence)] + if anchor is not None: + metadata.append(("source_anchor", anchor)) + return AdapterEdge( + edge=Edge(source_id, relation, target_id), + metadata=tuple(metadata), + ) diff --git a/tests/fixtures/reference-python/src/sample/__init__.py b/tests/fixtures/reference-python/src/sample/__init__.py new file mode 100644 index 0000000..cc3b58a --- /dev/null +++ b/tests/fixtures/reference-python/src/sample/__init__.py @@ -0,0 +1,7 @@ +"""Small multi-file package for the Python reference adapter.""" + +from .service import Service + + +def package_name() -> str: + return Service.__name__ diff --git a/tests/fixtures/reference-python/src/sample/service.py b/tests/fixtures/reference-python/src/sample/service.py new file mode 100644 index 0000000..b559183 --- /dev/null +++ b/tests/fixtures/reference-python/src/sample/service.py @@ -0,0 +1,18 @@ +"""Service behavior with class, method, function, and local import evidence.""" + +import json + +from .shared import clamp + +raise RuntimeError("The reference adapter must never execute project code") + + +class Service: + def run(self, value: int) -> int: + return clamp(value) if json.__name__ else value + + +async def build_service(enabled: bool) -> Service | None: + if enabled: + return Service() + return None diff --git a/tests/fixtures/reference-python/src/sample/shared.py b/tests/fixtures/reference-python/src/sample/shared.py new file mode 100644 index 0000000..42a1fe9 --- /dev/null +++ b/tests/fixtures/reference-python/src/sample/shared.py @@ -0,0 +1,11 @@ +"""Shared values used by dependent modules.""" + +DEFAULT_LIMIT = 3 + + +def clamp(value: int, limit: int = DEFAULT_LIMIT) -> int: + if value < 0: + return 0 + if value > limit: + return limit + return value diff --git a/tests/fixtures/reference-python/src/sample/worker.py b/tests/fixtures/reference-python/src/sample/worker.py new file mode 100644 index 0000000..7d1962d --- /dev/null +++ b/tests/fixtures/reference-python/src/sample/worker.py @@ -0,0 +1,8 @@ +"""A transitive dependent used to prove reverse invalidation.""" + +from sample.service import Service + + +def execute(value: int) -> int: + worker = Service() + return worker.run(value) diff --git a/tests/test_python_reference_adapter.py b/tests/test_python_reference_adapter.py new file mode 100644 index 0000000..8887458 --- /dev/null +++ b/tests/test_python_reference_adapter.py @@ -0,0 +1,267 @@ +from __future__ import annotations + +import shutil +import tempfile +import unittest +from pathlib import Path +from typing import cast + +from docforge.adapter_sdk import ( + AdapterProject, + AdapterSource, + AdapterSourceProjection, + verify_adapter_conformance, +) +from docforge.adapters.python import ( + PYTHON_ADAPTER_ID, + PYTHON_ADAPTER_VERSION, + PYTHON_EXTRACTOR_VERSION, + PythonReferenceAdapter, +) +from docforge.errors import DocForgeError +from docforge.index import ProjectIndex + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = ROOT / "tests" / "fixtures" / "reference-python" + + +class RecordingPythonAdapter(PythonReferenceAdapter): + def __init__(self, root: Path) -> None: + super().__init__( + root, + source_roots=("src",), + project_id="reference-python", + title="Reference Python fixture", + ) + self.extracted_paths: list[str] = [] + + def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: + self.extracted_paths.append(source.source_path) + return super().extract_source(source) + + +class PythonReferenceAdapterTests(unittest.TestCase): + def copy_fixture(self, parent: Path) -> Path: + root = parent / "reference-python" + shutil.copytree(FIXTURE, root) + return root.resolve() + + def adapter(self, root: Path) -> RecordingPythonAdapter: + return RecordingPythonAdapter(root) + + @staticmethod + def build_metrics(result: dict[str, object]) -> dict[str, object]: + return cast(dict[str, object], result["build"]) + + def test_complete_assembly_is_exact_ordered_cache_independent_and_scoped(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + adapter = self.adapter(root) + + first = adapter.load_assembly() + second = adapter.load_complete_assembly() + projection = adapter.load_projection() + + self.assertEqual(first, second) + self.assertEqual(first.projection, projection) + self.assertEqual(PYTHON_ADAPTER_ID, projection.adapter_id) + self.assertEqual(PYTHON_ADAPTER_VERSION, projection.adapter_version) + self.assertEqual(14, len(projection.nodes)) + self.assertEqual(13, len(projection.edges)) + self.assertEqual(5, len(first.logic)) + self.assertEqual( + list(projection.nodes), + sorted(projection.nodes, key=lambda item: item.node.node_id), + ) + self.assertEqual( + {"contains", "depends_on"}, {item.edge.relation for item in projection.edges} + ) + self.assertFalse(any(item.node.title == "json" for item in projection.nodes)) + self.assertTrue( + any( + item.node.title == "Service" and ("kind", "class") in item.metadata + for item in projection.nodes + ) + ) + self.assertTrue( + any( + item.node.title == "Service.run" and ("kind", "function") in item.metadata + for item in projection.nodes + ) + ) + + report = adapter.support_report() + self.assertEqual(1, report["schema_version"]) + self.assertEqual(PYTHON_EXTRACTOR_VERSION, report["extractor_version"]) + self.assertFalse(report["imports_project_code"]) + self.assertFalse(report["executes_project_code"]) + self.assertEqual( + [ + "call_resolution", + "dynamic_import_resolution", + "inheritance_resolution", + "runtime_generated_facts", + "symbol_reference_resolution", + ], + [item.code for item in adapter.unsupported_facts()], + ) + self.assertFalse((root / ".cache").exists()) + + conformance = verify_adapter_conformance( + adapter, + cache_root=root / ".cache" / "conformance", + ) + self.assertEqual("reference-python", conformance.project_id) + self.assertEqual(14, conformance.node_count) + self.assertEqual(13, conformance.edge_count) + self.assertEqual(5, conformance.logic_projection_count) + self.assertTrue(conformance.incremental) + + def test_incremental_cold_warm_reverse_dependencies_and_logic_equivalence(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + adapter = self.adapter(root) + project = AdapterProject(adapter, cache_root=root / ".cache" / "incremental") + index = ProjectIndex(project) + + cold = index.build() + self.assertEqual(4, self.build_metrics(cold)["reparsed_sources"]) + self.assertEqual( + [ + "src/sample/__init__.py", + "src/sample/service.py", + "src/sample/shared.py", + "src/sample/worker.py", + ], + sorted(adapter.extracted_paths), + ) + + adapter.extracted_paths.clear() + warm = index.build() + self.assertEqual(4, self.build_metrics(warm)["cache_hits"]) + self.assertEqual(0, self.build_metrics(warm)["reparsed_sources"]) + self.assertEqual([], adapter.extracted_paths) + + shared = root / "src" / "sample" / "shared.py" + shared.write_text( + shared.read_text(encoding="utf-8").replace( + "DEFAULT_LIMIT = 3", "DEFAULT_LIMIT = 5" + ), + encoding="utf-8", + ) + adapter.extracted_paths.clear() + changed = index.build() + self.assertEqual(4, self.build_metrics(changed)["invalidated_sources"]) + self.assertEqual( + [ + "src/sample/__init__.py", + "src/sample/service.py", + "src/sample/shared.py", + "src/sample/worker.py", + ], + sorted(adapter.extracted_paths), + ) + + equivalent = project.verify_incremental_equivalence() + self.assertEqual("ok", equivalent["status"]) + self.assertEqual(14, equivalent["node_count"]) + self.assertEqual(13, equivalent["edge_count"]) + self.assertEqual(5, equivalent["logic_projection_count"]) + + def test_add_rename_delete_and_corrupt_cache_recover_without_stale_facts(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + adapter = self.adapter(root) + cache_root = root / ".cache" / "incremental" + project = AdapterProject(adapter, cache_root=cache_root) + index = ProjectIndex(project) + index.build() + + extra = root / "src" / "sample" / "extra.py" + extra.write_text( + '"""Added source."""\n\n\ndef extra() -> str:\n return "extra"\n', + encoding="utf-8", + ) + adapter.extracted_paths.clear() + added = index.build() + self.assertEqual(1, self.build_metrics(added)["reparsed_sources"]) + self.assertEqual(["src/sample/extra.py"], adapter.extracted_paths) + self.assertIn( + "src/sample/extra.py", + {node.source_path for node in project.load().nodes}, + ) + + worker = root / "src" / "sample" / "worker.py" + runner = root / "src" / "sample" / "runner.py" + worker.rename(runner) + adapter.extracted_paths.clear() + renamed = index.build() + self.assertEqual(1, self.build_metrics(renamed)["deleted_sources"]) + self.assertEqual(1, self.build_metrics(renamed)["reparsed_sources"]) + renamed_paths = {node.source_path for node in project.load().nodes} + self.assertNotIn("src/sample/worker.py", renamed_paths) + self.assertIn("src/sample/runner.py", renamed_paths) + + extra.unlink() + adapter.extracted_paths.clear() + deleted = index.build() + self.assertEqual(1, self.build_metrics(deleted)["deleted_sources"]) + self.assertNotIn( + "src/sample/extra.py", + {node.source_path for node in project.load().nodes}, + ) + + (cache_root / "extractions.json").write_text("{broken", encoding="utf-8") + adapter.extracted_paths.clear() + recovered = index.build() + self.assertEqual(4, self.build_metrics(recovered)["reparsed_sources"]) + self.assertEqual( + [ + "src/sample/__init__.py", + "src/sample/runner.py", + "src/sample/service.py", + "src/sample/shared.py", + ], + sorted(adapter.extracted_paths), + ) + self.assertEqual("ok", project.verify_incremental_equivalence()["status"]) + + def test_confinement_and_no_ast_policy_reject_logic_without_importing_code(self) -> None: + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + root = self.copy_fixture(parent) + + for source_root in (root / "src", Path("../outside")): + with self.subTest(source_root=source_root): + with self.assertRaises(DocForgeError) as captured: + PythonReferenceAdapter(root, source_roots=(source_root,)) + self.assertEqual("path_escape", captured.exception.code) + + outside = parent / "outside" + outside.mkdir() + linked_root = root / "linked" + linked_root.symlink_to(outside, target_is_directory=True) + with self.assertRaises(DocForgeError) as linked: + PythonReferenceAdapter(root, source_roots=("linked",)) + self.assertEqual("path_escape", linked.exception.code) + + external_source = outside / "external.py" + external_source.write_text("def outside():\n return True\n", encoding="utf-8") + worker = root / "src" / "sample" / "worker.py" + worker.unlink() + worker.symlink_to(external_source) + adapter = self.adapter(root) + with self.assertRaises(DocForgeError) as source_link: + adapter.load_manifest() + self.assertEqual("path_escape", source_link.exception.code) + + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + adapter = self.adapter(root) + project = AdapterProject(adapter, cache_root=root / ".cache" / "no-ast") + + with self.assertRaises(DocForgeError) as no_ast: + ProjectIndex(project, allow_logic=False).build() + + self.assertEqual("adapter_policy_forbids_logic", no_ast.exception.code) + self.assertFalse(project.descriptor.index_path.exists())