diff --git a/src/docforge/adapters/cpp.py b/src/docforge/adapters/cpp.py new file mode 100644 index 0000000..11e6425 --- /dev/null +++ b/src/docforge/adapters/cpp.py @@ -0,0 +1,1349 @@ +"""Deterministic, syntax-only C++ reference adapter. + +The adapter accepts one explicit, project-confined ``compile_commands.json`` as +the authoritative translation-unit inventory. The database, compiler command, +arguments, and C++ sources are parsed only as inert data. No compiler, build +tool, project binary, or project code is ever executed. + +Only quoted includes that resolve directly to a real, project-local header are +published as dependencies. Tree-sitter proves lexical declarations and +function control flow; compiler-dependent facts are deliberately omitted. +""" + +from __future__ import annotations + +import hashlib +import importlib +import json +import os +import stat +from collections.abc import Iterable, Sequence +from dataclasses import dataclass, replace +from functools import lru_cache +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Protocol, cast + +from ..adapter_sdk import ( + AdapterAssembly, + AdapterEdge, + AdapterManifest, + AdapterNode, + AdapterProjection, + AdapterSource, + AdapterSourceProjection, + Edge, +) +from ..adapter_sdk import Node as GraphNode +from ..errors import DocForgeError +from ..treesitter_logic import ( + TreeSitterLogicOwner, + analyze_cpp_source, + discover_cpp_functions, +) + +if TYPE_CHECKING: + from tree_sitter import Language + +CPP_ADAPTER_ID = "docforge.reference.cpp" +CPP_ADAPTER_VERSION = "1" +CPP_EXTRACTOR_VERSION = "tree-sitter-cpp@1" +CPP_IDENTITY_VERSION = "cpp-reference-id@1" +CPP_SUPPORT_SCHEMA_VERSION = 1 + +_TRANSLATION_UNIT_SUFFIXES = frozenset({".c++", ".cc", ".cpp", ".cxx"}) +_HEADER_SUFFIXES = frozenset({".h", ".hh", ".hpp", ".hxx", ".inc", ".ipp", ".tpp"}) +_SCOPE_TYPES = { + "namespace_definition": "namespace", + "class_specifier": "class", + "struct_specifier": "struct", +} +_COMPILE_ENTRY_KEYS = frozenset({"arguments", "command", "directory", "file", "output"}) +_SUPPORTED_FACTS = ( + "cpp_file", + "cpp_translation_unit", + "cpp_namespace", + "cpp_class", + "cpp_struct", + "cpp_function", + "lexical_containment", + "project_local_quoted_include_dependency", + "function_logic", +) + + +@dataclass(frozen=True) +class CppUnsupportedFact: + """One compiler-dependent fact this syntax-only adapter does not claim.""" + + code: str + description: str + + def as_dict(self) -> dict[str, str]: + return {"code": self.code, "description": self.description} + + +_UNSUPPORTED_FACTS = ( + CppUnsupportedFact( + "call_resolution", + "Calls appear only inside function Logic and are not resolved to symbols.", + ), + CppUnsupportedFact( + "type_resolution", + "Types, templates, aliases, concepts, variables, and references are not resolved.", + ), + CppUnsupportedFact( + "inheritance_resolution", + "Base-specifier syntax is not resolved to local or external types.", + ), + CppUnsupportedFact( + "overload_ownership", + "Overload sets and out-of-line semantic ownership are not compiler-resolved.", + ), + CppUnsupportedFact( + "macro_semantics", + "Conditional compilation and macro expansion are not evaluated.", + ), + CppUnsupportedFact( + "compiler_command_execution", + "Compilation database commands and arguments are retained as fingerprints only.", + ), + CppUnsupportedFact( + "nonlocal_include_resolution", + "Angle-bracket, include-path, framework, generated, and compiler-provided " + "headers are omitted.", + ), +) + + +class _Point(Protocol): + row: int + + +class _TreeNode(Protocol): + type: str + has_error: bool + is_error: bool + is_missing: bool + start_byte: int + end_byte: int + start_point: _Point + named_children: list[_TreeNode] + + def child_by_field_name(self, name: str, /) -> _TreeNode | None: ... + + +@dataclass(frozen=True) +class _CompilationEntry: + source_path: str + directory: str + command: str | None + arguments: tuple[str, ...] | None + output: str | None + + def canonical_bytes(self) -> bytes: + value: dict[str, object] = { + "directory": self.directory, + "file": self.source_path, + } + if self.command is not None: + value["command"] = self.command + if self.arguments is not None: + value["arguments"] = list(self.arguments) + if self.output is not None: + value["output"] = self.output + return json.dumps( + value, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + + +@dataclass(frozen=True) +class _SourceRecord: + source_id: str + source_path: str + translation_unit: bool + fingerprint: str + text: str + raw: bytes + root: _TreeNode + compilation_entry: _CompilationEntry | None = None + dependencies: tuple[str, ...] = () + + +@dataclass(frozen=True) +class _Definition: + kind: str + qualified_name: str + logic_qualified_name: str | None + line: int + parent_qualified_name: str | None + node: _TreeNode + node_id: str + + +@lru_cache(maxsize=1) +def _cpp_language() -> Language: + try: + tree_sitter = importlib.import_module("tree_sitter") + grammar = importlib.import_module("tree_sitter_cpp") + except ModuleNotFoundError as error: + raise DocForgeError( + "optional_dependency_missing", + "The requested language frontend is not installed", + extra="cpp", + install="docforge[cpp]", + missing_module=error.name, + ) from error + return cast("Language", tree_sitter.Language(grammar.language())) + + +class CppReferenceAdapter: + """Reference C++ adapter bound to one explicit compilation database.""" + + def __init__( + self, + root: Path, + *, + source_roots: Iterable[str | Path], + compilation_database: str | Path, + project_id: str = "cpp-reference", + title: str = "C++ reference project", + max_translation_units: int = 4_096, + max_headers: int = 8_192, + max_database_bytes: int = 8_000_000, + max_source_bytes: int = 2_000_000, + max_command_chars: int = 64_000, + max_arguments: int = 8_192, + 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", "C++ adapter root must be a directory") + if ( + max_translation_units < 1 + or max_headers < 1 + or max_database_bytes < 1 + or max_source_bytes < 1 + or max_command_chars < 1 + or max_arguments < 1 + or max_logic_nodes_per_function < 2 + ): + raise ValueError("C++ adapter limits must be positive") + self.root = resolved_root + self.project_id = project_id + self.title = title + self.max_translation_units = max_translation_units + self.max_headers = max_headers + self.max_database_bytes = max_database_bytes + self.max_source_bytes = max_source_bytes + self.max_command_chars = max_command_chars + self.max_arguments = max_arguments + self.max_logic_nodes_per_function = max_logic_nodes_per_function + self.source_roots = self._normalize_source_roots(source_roots) + self.compilation_database = self._normalize_database_path(compilation_database) + + def support_report(self) -> dict[str, object]: + """Return deterministic scope, frontend, and limitation evidence.""" + + return { + "schema_version": CPP_SUPPORT_SCHEMA_VERSION, + "adapter_id": CPP_ADAPTER_ID, + "adapter_version": CPP_ADAPTER_VERSION, + "extractor_version": CPP_EXTRACTOR_VERSION, + "identity_version": CPP_IDENTITY_VERSION, + "frontend": "tree-sitter-cpp", + "compilation_database": self.compilation_database, + "compilation_database_is_authoritative": True, + "source_roots": list(self.source_roots), + "executes_compiler": False, + "executes_project_code": False, + "translation_unit_suffixes": sorted(_TRANSLATION_UNIT_SUFFIXES), + "header_suffixes": sorted(_HEADER_SUFFIXES), + "supported_facts": list(_SUPPORTED_FACTS), + "unsupported_facts": [fact.as_dict() for fact in _UNSUPPORTED_FACTS], + } + + def unsupported_facts(self) -> tuple[CppUnsupportedFact, ...]: + return _UNSUPPORTED_FACTS + + def load_manifest(self) -> AdapterManifest: + records = self._inventory() + path_sources = {record.source_path: record.source_id for record in records} + sources = tuple( + sorted( + ( + AdapterSource( + source_id=record.source_id, + source_path=record.source_path, + fingerprint=record.fingerprint, + extractor_version=CPP_EXTRACTOR_VERSION, + dependencies=tuple( + sorted( + path_sources[path] + for path in record.dependencies + if path_sources[path] != record.source_id + ) + ), + ) + 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=CPP_ADAPTER_ID, + adapter_version=CPP_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) * 10), + ) + + def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: + entry: _CompilationEntry | None = None + translation_unit = source.source_id.startswith("cpp.tu.") + if translation_unit: + entries = {item.source_path: item for item in self._compilation_entries()} + entry = entries.get(source.source_path) + if entry is None: + raise DocForgeError( + "stale_adapter_source", + "C++ translation unit is no longer in the compilation database", + source=source.source_path, + ) + record = self._read_source(source.source_path, compilation_entry=entry) + if ( + record.source_id != source.source_id + or record.fingerprint != source.fingerprint + or record.translation_unit != translation_unit + ): + raise DocForgeError( + "stale_adapter_source", + "C++ source or compilation entry 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", + "C++ 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) + if self.load_manifest() != manifest: + raise DocForgeError( + "source_changed", + "C++ inputs 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 complete-projection compatibility contract.""" + + return self.load_assembly().projection + + def _normalize_database_path(self, value: str | Path) -> str: + supplied = Path(value) + candidate = supplied if supplied.is_absolute() else self.root / supplied + normalized = Path(os.path.abspath(candidate)) + if normalized.name != "compile_commands.json" or not normalized.is_relative_to(self.root): + raise DocForgeError( + "path_escape", + "C++ compilation database must be an explicit project-confined " + "compile_commands.json", + ) + relative = normalized.relative_to(self.root).as_posix() + self._validate_real_file( + normalized, + relative, + unavailable_code="invalid_compilation_database", + label="C++ compilation database", + ) + return relative + + 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", + "C++ 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", "C++ source root cannot be a symbolic link") + try: + resolved = absolute.resolve(strict=True) + except OSError as error: + raise DocForgeError( + "invalid_adapter", + "C++ 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", + "C++ 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 C++ source root is required") + if len(result) != len(normalized): + raise DocForgeError("invalid_adapter", "C++ 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", + "C++ source roots must not overlap", + ) + return result + + 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", + "C++ source is outside the declared source roots", + source=path.as_posix(), + ) + return matches[0] + + def _compilation_entries(self) -> tuple[_CompilationEntry, ...]: + database = self.root.joinpath(*PurePosixPath(self.compilation_database).parts) + raw = self._read_real_file( + database, + self.compilation_database, + maximum=self.max_database_bytes, + unavailable_code="invalid_compilation_database", + label="C++ compilation database", + ) + try: + text = raw.decode("utf-8") + except UnicodeDecodeError as error: + raise DocForgeError( + "invalid_compilation_database", + "C++ compilation database must use UTF-8", + ) from error + try: + parsed = cast( + object, + json.loads(text, object_pairs_hook=self._unique_json_object), + ) + except (TypeError, ValueError, json.JSONDecodeError) as error: + raise DocForgeError( + "invalid_compilation_database", + "C++ compilation database is not valid duplicate-free JSON", + ) from error + if not isinstance(parsed, list) or not parsed: + raise DocForgeError( + "invalid_compilation_database", + "C++ compilation database must be a non-empty JSON array", + ) + parsed_items = cast("list[object]", parsed) + if len(parsed_items) > self.max_translation_units: + raise DocForgeError( + "adapter_too_large", + "C++ translation-unit inventory exceeds the configured limit", + maximum=self.max_translation_units, + ) + entries = tuple( + self._compilation_entry(value, index) for index, value in enumerate(parsed_items) + ) + paths = [entry.source_path for entry in entries] + if len(paths) != len(set(paths)): + duplicates = sorted(path for path in set(paths) if paths.count(path) > 1) + raise DocForgeError( + "duplicate_translation_unit", + "C++ compilation database repeats a normalized translation unit", + sources=duplicates, + ) + return tuple(sorted(entries, key=lambda entry: entry.source_path)) + + def _compilation_entry(self, value: object, index: int) -> _CompilationEntry: + if not isinstance(value, dict): + raise DocForgeError( + "invalid_compilation_database", + "Each C++ compilation entry must be a JSON object", + entry=index, + ) + raw_document = cast("dict[object, object]", value) + if any(not isinstance(key, str) for key in raw_document): + raise DocForgeError( + "invalid_compilation_database", + "Each C++ compilation entry must have string keys", + entry=index, + ) + document = cast("dict[str, object]", raw_document) + keys = set(document) + if ( + not keys <= _COMPILE_ENTRY_KEYS + or not {"directory", "file"} <= keys + or (("command" in keys) == ("arguments" in keys)) + ): + raise DocForgeError( + "invalid_compilation_database", + "Each C++ compilation entry requires directory, file, and exactly " + "one of command or arguments", + entry=index, + ) + directory_value = self._bounded_string( + document["directory"], + "directory", + index, + maximum=self.max_command_chars, + ) + file_value = self._bounded_string( + document["file"], + "file", + index, + maximum=self.max_command_chars, + ) + directory_path = Path(directory_value) + directory_candidate = ( + directory_path if directory_path.is_absolute() else self.root / directory_path + ) + directory = Path(os.path.abspath(directory_candidate)) + if not directory.is_relative_to(self.root): + raise DocForgeError( + "path_escape", + "C++ compilation entry directory escapes the project root", + entry=index, + ) + self._validate_real_directory(directory, index) + file_path = Path(file_value) + file_candidate = file_path if file_path.is_absolute() else directory / file_path + source = Path(os.path.abspath(file_candidate)) + if ( + not source.is_relative_to(self.root) + or source.suffix.casefold() not in _TRANSLATION_UNIT_SUFFIXES + ): + raise DocForgeError( + "path_escape", + "C++ compilation entry file must be a confined C++ translation unit", + entry=index, + ) + relative = source.relative_to(self.root).as_posix() + self._source_root_for(PurePosixPath(relative)) + self._validate_real_file( + source, + relative, + unavailable_code="invalid_compilation_database", + label="C++ translation unit", + ) + command: str | None = None + arguments: tuple[str, ...] | None = None + if "command" in value: + command = self._bounded_string( + document["command"], + "command", + index, + maximum=self.max_command_chars, + ) + else: + raw_arguments = document["arguments"] + if not isinstance(raw_arguments, list): + raise DocForgeError( + "invalid_compilation_database", + "C++ compilation arguments must be a non-empty bounded string array", + entry=index, + ) + argument_items = cast("list[object]", raw_arguments) + if not argument_items or len(argument_items) > self.max_arguments: + raise DocForgeError( + "invalid_compilation_database", + "C++ compilation arguments must be a non-empty bounded string array", + entry=index, + ) + arguments = tuple( + self._bounded_string( + argument, + "arguments", + index, + maximum=self.max_command_chars, + ) + for argument in argument_items + ) + output = ( + self._bounded_string( + document["output"], + "output", + index, + maximum=self.max_command_chars, + ) + if "output" in value + else None + ) + directory_relative = directory.relative_to(self.root).as_posix() or "." + return _CompilationEntry( + source_path=relative, + directory=directory_relative, + command=command, + arguments=arguments, + output=output, + ) + + @staticmethod + def _unique_json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate JSON key: {key}") + result[key] = value + return result + + @staticmethod + def _bounded_string( + value: object, + field: str, + index: int, + *, + maximum: int, + ) -> str: + if ( + not isinstance(value, str) + or not value + or len(value) > maximum + or any(ord(character) < 32 or ord(character) == 127 for character in value) + ): + raise DocForgeError( + "invalid_compilation_database", + f"C++ compilation entry {field} must be a non-empty bounded control-free string", + entry=index, + ) + return value + + def _validate_real_directory(self, path: Path, index: int) -> None: + try: + before = path.lstat() + resolved = path.resolve(strict=True) + except OSError as error: + raise DocForgeError( + "invalid_compilation_database", + "C++ compilation entry directory does not exist", + entry=index, + ) from error + if ( + stat.S_ISLNK(before.st_mode) + or not stat.S_ISDIR(before.st_mode) + or resolved != path + or not resolved.is_relative_to(self.root) + ): + raise DocForgeError( + "path_escape", + "C++ compilation entry directory must be a confined real directory", + entry=index, + ) + + def _inventory(self) -> tuple[_SourceRecord, ...]: + entries = self._compilation_entries() + records: dict[str, _SourceRecord] = {} + queue: list[tuple[str, _CompilationEntry | None]] = [ + (entry.source_path, entry) for entry in entries + ] + header_count = 0 + while queue: + source_path, compilation_entry = queue.pop(0) + existing = records.get(source_path) + if existing is not None: + if ( + compilation_entry is not None + and existing.compilation_entry != compilation_entry + ): + raise DocForgeError( + "duplicate_translation_unit", + "C++ source has conflicting compilation entries", + source=source_path, + ) + continue + record = self._read_source(source_path, compilation_entry=compilation_entry) + dependencies: list[str] = [] + for include in self._quoted_includes(record): + resolved = self._resolve_quoted_header(source_path, include) + if resolved is None or resolved == source_path: + continue + dependencies.append(resolved) + if resolved not in records and all(item[0] != resolved for item in queue): + header_count += 1 + if header_count > self.max_headers: + raise DocForgeError( + "adapter_too_large", + "C++ project-local header inventory exceeds the configured limit", + maximum=self.max_headers, + ) + queue.append((resolved, None)) + records[source_path] = replace( + record, + dependencies=tuple(sorted(set(dependencies))), + ) + queue.sort(key=lambda item: item[0]) + return tuple(sorted(records.values(), key=lambda record: record.source_path)) + + def _read_source( + self, + relative: str, + *, + compilation_entry: _CompilationEntry | None, + ) -> _SourceRecord: + path = PurePosixPath(relative) + translation_unit = compilation_entry is not None + allowed_suffixes = _TRANSLATION_UNIT_SUFFIXES if translation_unit else _HEADER_SUFFIXES + if ( + path.is_absolute() + or ".." in path.parts + or path.suffix.casefold() not in allowed_suffixes + ): + raise DocForgeError("path_escape", "C++ source path is unsafe", source=relative) + self._source_root_for(path) + candidate = self.root.joinpath(*path.parts) + raw = self._read_real_file( + candidate, + relative, + maximum=self.max_source_bytes, + unavailable_code="stale_adapter_source", + label="C++ source", + ) + try: + text = raw.decode("utf-8-sig") + except UnicodeDecodeError as error: + raise DocForgeError( + "invalid_cpp_source", + "C++ source must use UTF-8", + source=relative, + ) from error + parse_raw, root = self._parse(text) + if root.has_error: + raise DocForgeError( + "invalid_cpp_source", + "C++ source cannot be parsed by the reference adapter", + source=relative, + line=self._first_error_line(root) or 1, + ) + source_id = self._source_id(relative, translation_unit=translation_unit) + digest = hashlib.sha256() + digest.update(CPP_EXTRACTOR_VERSION.encode()) + digest.update(b"\0source\0") + digest.update(raw) + if compilation_entry is not None: + digest.update(b"\0compilation-entry\0") + digest.update(compilation_entry.canonical_bytes()) + return _SourceRecord( + source_id=source_id, + source_path=relative, + translation_unit=translation_unit, + fingerprint=digest.hexdigest(), + text=text, + raw=parse_raw, + root=root, + compilation_entry=compilation_entry, + ) + + @staticmethod + def _parse(source: str) -> tuple[bytes, _TreeNode]: + try: + tree_sitter = importlib.import_module("tree_sitter") + except ModuleNotFoundError as error: + raise DocForgeError( + "optional_dependency_missing", + "The requested language frontend is not installed", + extra="cpp", + install="docforge[cpp]", + missing_module=error.name, + ) from error + raw = source.encode("utf-8") + tree = tree_sitter.Parser(_cpp_language()).parse(raw) + return raw, cast("_TreeNode", tree.root_node) + + @classmethod + def _first_error_line(cls, node: _TreeNode) -> int | None: + if node.is_error or node.is_missing: + return node.start_point.row + 1 + for child in node.named_children: + line = cls._first_error_line(child) + if line is not None: + return line + return None + + def _resolve_quoted_header(self, source_path: str, include: str) -> str | None: + if ( + not include + or "\\" in include + or "\x00" in include + or PurePosixPath(include).is_absolute() + ): + return None + source_parent = self.root.joinpath(*PurePosixPath(source_path).parent.parts) + candidate = Path(os.path.abspath(source_parent / include)) + if not candidate.is_relative_to(self.root): + raise DocForgeError( + "path_escape", + "Quoted C++ include escapes the project root", + source=source_path, + include=include, + ) + if candidate.is_symlink(): + raise DocForgeError( + "path_escape", + "Quoted C++ include resolves through a symbolic link", + source=source_path, + include=include, + ) + if not candidate.exists(): + return None + if candidate.suffix.casefold() not in _HEADER_SUFFIXES: + return None + relative = candidate.relative_to(self.root).as_posix() + self._source_root_for(PurePosixPath(relative)) + self._validate_real_file( + candidate, + relative, + unavailable_code="invalid_cpp_source", + label="Quoted C++ header", + ) + return relative + + @classmethod + def _quoted_includes(cls, record: _SourceRecord) -> tuple[str, ...]: + result: set[str] = set() + stack = [record.root] + while stack: + node = stack.pop() + if node.type == "preproc_include": + path = node.child_by_field_name("path") + if path is not None: + value = cls._text(path, record.raw) + if ( + len(value) >= 2 + and value.startswith('"') + and value.endswith('"') + and "\\" not in value + and "\x00" not in value + ): + result.add(value[1:-1]) + stack.extend(node.named_children) + return tuple(sorted(result)) + + def _extract_record( + self, + record: _SourceRecord, + source: AdapterSource, + ) -> AdapterSourceProjection: + file_id = self._file_node_id(source.source_id) + root_id = file_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", + ) + ] + edges: list[AdapterEdge] = [] + if record.translation_unit: + root_id = self._translation_unit_node_id(source.source_id) + nodes.append( + self._node( + node_id=root_id, + title=record.source_path, + kind="translation_unit", + qualified_name=record.source_path, + content=f"C++ translation unit {record.source_path}.", + source_path=record.source_path, + anchor="L1", + ) + ) + edges.append(self._edge(root_id, "contains", file_id, "compilation_database")) + definitions = self._definitions(record) + definition_ids = { + definition.qualified_name: definition.node_id for definition in definitions + } + for definition in definitions: + parent_id = ( + definition_ids.get(definition.parent_qualified_name, file_id) + if definition.parent_qualified_name is not None + else file_id + ) + anchor = f"L{definition.line}" + nodes.append( + self._node( + node_id=definition.node_id, + title=definition.qualified_name, + kind=definition.kind, + qualified_name=definition.qualified_name, + content=self._text(definition.node, record.raw), + source_path=record.source_path, + anchor=anchor, + ) + ) + edges.append( + self._edge( + parent_id, + "contains", + definition.node_id, + "tree_sitter_syntax", + anchor=anchor, + ) + ) + for dependency in source.dependencies: + edges.append( + self._edge( + file_id, + "depends_on", + self._file_node_id(dependency), + "quoted_project_header", + ) + ) + owners = tuple( + TreeSitterLogicOwner( + owner_node_id=definition.node_id, + qualified_name=cast(str, definition.logic_qualified_name), + line=definition.line, + ) + for definition in definitions + if definition.kind == "function" + ) + logic = analyze_cpp_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) -> tuple[_Definition, ...]: + definitions: list[_Definition] = [] + identities: set[tuple[str, int, str]] = set() + + def visit(node: _TreeNode, scopes: tuple[str, ...]) -> None: + definition = self._definition(record, node, scopes) + next_scopes = scopes + if definition is not None: + identity = ( + definition.qualified_name, + definition.line, + definition.kind, + ) + if identity in identities: + raise DocForgeError( + "ambiguous_cpp_symbol", + "C++ source repeats a syntax-local symbol identity", + source=record.source_path, + qualified_name=definition.qualified_name, + line=definition.line, + ) + identities.add(identity) + definitions.append(definition) + if definition.kind != "function": + next_scopes = tuple(definition.qualified_name.split(".")) + for child in node.named_children: + visit(child, next_scopes) + + visit(record.root, ()) + ordered = tuple(sorted(definitions, key=lambda item: item.node_id)) + public_functions = { + (function.qualified_name, function.line) + for function in discover_cpp_functions(record.text) + } + extracted_functions = { + (cast(str, definition.logic_qualified_name), definition.line) + for definition in ordered + if definition.kind == "function" + } + if extracted_functions != public_functions: + raise DocForgeError( + "invalid_adapter", + "C++ function discovery drifted from the Logic frontend", + source=record.source_path, + ) + if len(public_functions) != len( + [definition for definition in ordered if definition.kind == "function"] + ): + raise DocForgeError( + "ambiguous_cpp_symbol", + "C++ source has functions that cannot receive unique syntax-local Logic owners", + source=record.source_path, + ) + return ordered + + def _definition( + self, + record: _SourceRecord, + node: _TreeNode, + scopes: tuple[str, ...], + ) -> _Definition | None: + kind = _SCOPE_TYPES.get(node.type) + name: str | None = None + logic_name: str | None = None + if kind is not None: + name = self._field_text(node, "name", record.raw) + elif node.type == "function_definition": + declarator = node.child_by_field_name("declarator") + logic_name = self._declarator_name(declarator, record.raw) + name = logic_name + kind = "function" if name is not None else None + if kind is None or name is None: + return None + normalized_name = name.replace("::", ".") + if "::" in name: + qualified_name = normalized_name + else: + qualified_name = ".".join((*scopes, normalized_name)) + if kind == "function": + logic_name = ".".join((*scopes, name)) + parent = ".".join(scopes) or None + line = node.start_point.row + 1 + return _Definition( + kind=kind, + qualified_name=qualified_name, + logic_qualified_name=logic_name if kind == "function" else None, + line=line, + parent_qualified_name=parent, + node=node, + node_id=self._symbol_node_id( + kind, + record.source_id, + qualified_name, + line=line, + ), + ) + + @classmethod + def _declarator_name(cls, node: _TreeNode | None, raw: bytes) -> str | None: + if node is None: + return None + if node.type in { + "identifier", + "field_identifier", + "operator_name", + "destructor_name", + "qualified_identifier", + }: + return cls._text(node, raw) + for field in ("declarator", "name"): + child = node.child_by_field_name(field) + result = cls._declarator_name(child, raw) + if result: + return result + for child in node.named_children: + result = cls._declarator_name(child, raw) + if result: + return result + return None + + @staticmethod + def _source_id(relative: str, *, translation_unit: bool) -> str: + source_kind = "tu" if translation_unit else "header" + digest = hashlib.sha256( + f"{CPP_IDENTITY_VERSION}\0{source_kind}\0{relative}".encode() + ).hexdigest()[:24] + return f"cpp.{source_kind}.{digest}" + + @staticmethod + def _file_node_id(source_id: str) -> str: + if source_id.startswith("cpp.tu."): + return source_id.replace("cpp.tu.", "cpp.file.", 1) + return source_id.replace("cpp.header.", "cpp.file.", 1) + + @staticmethod + def _translation_unit_node_id(source_id: str) -> str: + return source_id.replace("cpp.tu.", "cpp.translation-unit.", 1) + + @staticmethod + def _symbol_node_id( + kind: str, + source_id: str, + qualified_name: str, + *, + line: int, + ) -> str: + digest = hashlib.sha256( + (f"{CPP_IDENTITY_VERSION}\0{kind}\0{source_id}\0{qualified_name}\0{line}").encode() + ).hexdigest()[:24] + return f"cpp.{kind}.{digest}" + + @staticmethod + def _source_hash(sources: Sequence[AdapterSource]) -> str: + digest = hashlib.sha256() + digest.update(CPP_ADAPTER_VERSION.encode()) + digest.update(CPP_EXTRACTOR_VERSION.encode()) + digest.update(CPP_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() + + @staticmethod + def _text(node: _TreeNode, raw: bytes) -> str: + return raw[node.start_byte : node.end_byte].decode("utf-8") + + @classmethod + def _field_text(cls, node: _TreeNode, field: str, raw: bytes) -> str | None: + child = node.child_by_field_name(field) + return cls._text(child, raw) if child is not None else None + + @staticmethod + def _node( + *, + node_id: str, + title: str, + kind: str, + qualified_name: str, + content: str, + source_path: str, + anchor: str, + ) -> AdapterNode: + normalized = content.strip() or f"C++ {kind} {qualified_name}." + return AdapterNode( + node=GraphNode( + node_id=node_id, + title=title, + family="code", + authority="derived", + status="active", + tags=tuple(sorted({"cpp", kind})), + summary=f"C++ {kind} syntax fact for {qualified_name}.", + content=normalized, + source_path=source_path, + source_anchor=anchor, + content_hash=hashlib.sha256(normalized.encode()).hexdigest(), + ), + metadata=( + ("evidence", "tree-sitter-syntax"), + ("extractor", CPP_EXTRACTOR_VERSION), + ("identity", CPP_IDENTITY_VERSION), + ("kind", kind), + ("language", "cpp"), + ("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), + ) + + def _validate_real_file( + self, + path: Path, + relative: str, + *, + unavailable_code: str, + label: str, + ) -> os.stat_result: + try: + before = path.lstat() + resolved = path.resolve(strict=True) + except OSError as error: + raise DocForgeError( + unavailable_code, + f"{label} is unavailable", + source=relative, + ) from error + if ( + stat.S_ISLNK(before.st_mode) + or not stat.S_ISREG(before.st_mode) + or resolved != path + or not resolved.is_relative_to(self.root) + ): + raise DocForgeError( + "path_escape", + f"{label} must be a confined real regular file", + source=relative, + ) + return before + + def _read_real_file( + self, + path: Path, + relative: str, + *, + maximum: int, + unavailable_code: str, + label: str, + ) -> bytes: + before = self._validate_real_file( + path, + relative, + unavailable_code=unavailable_code, + label=label, + ) + if before.st_size > maximum: + raise DocForgeError( + "source_too_large", + f"{label} exceeds the configured adapter limit", + source=relative, + maximum=maximum, + ) + descriptor = -1 + try: + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, 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", + f"{label} identity changed before it was opened", + source=relative, + ) + with os.fdopen(descriptor, "rb") as handle: + descriptor = -1 + raw = handle.read(maximum + 1) + after_open = os.fstat(handle.fileno()) + after = path.lstat() + except DocForgeError: + raise + except OSError as error: + raise DocForgeError( + unavailable_code, + f"{label} changed while being read", + source=relative, + ) from error + finally: + if descriptor >= 0: + os.close(descriptor) + identities = ( + (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns), + ( + after_open.st_dev, + after_open.st_ino, + after_open.st_size, + after_open.st_mtime_ns, + ), + (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns), + ) + if ( + len(raw) > maximum + or identities[0] != identities[1] + or identities[0] != identities[2] + or len(raw) != after.st_size + ): + raise DocForgeError( + unavailable_code, + f"{label} changed while being read", + source=relative, + ) + return raw diff --git a/tests/fixtures/reference-cpp/compile_commands.json b/tests/fixtures/reference-cpp/compile_commands.json new file mode 100644 index 0000000..a34b931 --- /dev/null +++ b/tests/fixtures/reference-cpp/compile_commands.json @@ -0,0 +1,19 @@ +[ + { + "arguments": [ + "c++", + "-std=c++20", + "-Iinclude", + "-c", + "src/main.cpp" + ], + "directory": ".", + "file": "src/main.cpp" + }, + { + "command": "c++ -std=c++20 -Iinclude -c src/worker.cpp", + "directory": ".", + "file": "src/worker.cpp", + "output": "worker.o" + } +] diff --git a/tests/fixtures/reference-cpp/include/config.hpp b/tests/fixtures/reference-cpp/include/config.hpp new file mode 100644 index 0000000..7e57679 --- /dev/null +++ b/tests/fixtures/reference-cpp/include/config.hpp @@ -0,0 +1,15 @@ +#pragma once + +#include "detail.hpp" + +namespace demo { + +struct Config { + int limit; +}; + +inline int effective_limit(const Config& config) { + return normalize(config.limit); +} + +} // namespace demo diff --git a/tests/fixtures/reference-cpp/include/detail.hpp b/tests/fixtures/reference-cpp/include/detail.hpp new file mode 100644 index 0000000..63a76f6 --- /dev/null +++ b/tests/fixtures/reference-cpp/include/detail.hpp @@ -0,0 +1,12 @@ +#pragma once + +namespace demo { + +inline int normalize(int value) { + if (value < 0) { + return 0; + } + return value; +} + +} // namespace demo diff --git a/tests/fixtures/reference-cpp/src/main.cpp b/tests/fixtures/reference-cpp/src/main.cpp new file mode 100644 index 0000000..6f298ae --- /dev/null +++ b/tests/fixtures/reference-cpp/src/main.cpp @@ -0,0 +1,17 @@ +#include "../include/config.hpp" + +namespace demo { + +int execute(int value) { + Config config{value}; + if (value == 0) { + return effective_limit(config); + } + return normalize(value); +} + +} // namespace demo + +int main() { + return demo::execute(3); +} diff --git a/tests/fixtures/reference-cpp/src/worker.cpp b/tests/fixtures/reference-cpp/src/worker.cpp new file mode 100644 index 0000000..1d1cbcb --- /dev/null +++ b/tests/fixtures/reference-cpp/src/worker.cpp @@ -0,0 +1,17 @@ +#include "../include/config.hpp" + +namespace demo { + +class Worker { +public: + int run(int value) { + for (int attempt = 0; attempt < 2; ++attempt) { + if (value > attempt) { + return effective_limit(Config{value}); + } + } + return 0; + } +}; + +} // namespace demo diff --git a/tests/test_cpp_reference_adapter.py b/tests/test_cpp_reference_adapter.py new file mode 100644 index 0000000..70c3bdf --- /dev/null +++ b/tests/test_cpp_reference_adapter.py @@ -0,0 +1,486 @@ +from __future__ import annotations + +import json +import shutil +import tempfile +import unittest +from pathlib import Path +from typing import cast +from unittest.mock import patch + +from docforge.adapter_sdk import ( + AdapterProject, + AdapterSource, + AdapterSourceProjection, + verify_adapter_conformance, +) +from docforge.adapters.cpp import ( + CPP_ADAPTER_ID, + CPP_ADAPTER_VERSION, + CPP_EXTRACTOR_VERSION, + CppReferenceAdapter, +) +from docforge.errors import DocForgeError +from docforge.index import ProjectIndex + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = ROOT / "tests" / "fixtures" / "reference-cpp" + + +class RecordingCppAdapter(CppReferenceAdapter): + def __init__(self, root: Path) -> None: + super().__init__( + root, + source_roots=("include", "src"), + compilation_database="compile_commands.json", + project_id="reference-cpp", + title="Reference C++ 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 CppReferenceAdapterTests(unittest.TestCase): + def copy_fixture(self, parent: Path) -> Path: + root = parent / "reference-cpp" + shutil.copytree(FIXTURE, root) + return root.resolve() + + def adapter(self, root: Path) -> RecordingCppAdapter: + return RecordingCppAdapter(root) + + @staticmethod + def build_metrics(result: dict[str, object]) -> dict[str, object]: + return cast(dict[str, object], result["build"]) + + @staticmethod + def read_database(root: Path) -> list[dict[str, object]]: + return cast( + list[dict[str, object]], + json.loads((root / "compile_commands.json").read_text(encoding="utf-8")), + ) + + @staticmethod + def write_database(root: Path, entries: list[dict[str, object]]) -> None: + (root / "compile_commands.json").write_text( + json.dumps(entries, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + def test_complete_assembly_is_deterministic_conformant_and_syntax_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(CPP_ADAPTER_ID, projection.adapter_id) + self.assertEqual(CPP_ADAPTER_VERSION, projection.adapter_version) + self.assertEqual(17, len(projection.nodes)) + self.assertEqual(16, 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.assertTrue( + any( + item.node.title == "demo.Worker" and ("kind", "class") in item.metadata + for item in projection.nodes + ) + ) + self.assertTrue( + any( + item.node.title == "demo.Config" and ("kind", "struct") in item.metadata + for item in projection.nodes + ) + ) + self.assertTrue( + any( + item.node.title == "demo.Worker.run" and ("kind", "function") in item.metadata + for item in projection.nodes + ) + ) + self.assertEqual( + {"include/config.hpp", "include/detail.hpp", "src/main.cpp", "src/worker.cpp"}, + {item.node.source_path for item in projection.nodes}, + ) + self.assertFalse( + {"calls", "inherits", "type_of"} & {item.edge.relation for item in projection.edges} + ) + + report = adapter.support_report() + self.assertEqual(1, report["schema_version"]) + self.assertEqual(CPP_EXTRACTOR_VERSION, report["extractor_version"]) + self.assertEqual(["include", "src"], report["source_roots"]) + self.assertTrue(report["compilation_database_is_authoritative"]) + self.assertFalse(report["executes_compiler"]) + self.assertFalse(report["executes_project_code"]) + self.assertEqual( + [ + "call_resolution", + "type_resolution", + "inheritance_resolution", + "overload_ownership", + "macro_semantics", + "compiler_command_execution", + "nonlocal_include_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-cpp", conformance.project_id) + self.assertEqual(17, conformance.node_count) + self.assertEqual(16, conformance.edge_count) + self.assertEqual(5, conformance.logic_projection_count) + self.assertTrue(conformance.incremental) + + def test_incremental_cold_warm_source_header_and_flag_invalidation(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( + [ + "include/config.hpp", + "include/detail.hpp", + "src/main.cpp", + "src/worker.cpp", + ], + 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) + + main = root / "src" / "main.cpp" + main.write_text( + main.read_text(encoding="utf-8").replace( + "return demo::execute(3);", + "return demo::execute(4);", + ), + encoding="utf-8", + ) + adapter.extracted_paths.clear() + source_changed = index.build() + self.assertEqual(1, self.build_metrics(source_changed)["invalidated_sources"]) + self.assertEqual(["src/main.cpp"], adapter.extracted_paths) + + detail = root / "include" / "detail.hpp" + detail.write_text( + detail.read_text(encoding="utf-8").replace( + "return value;", + "return value + 0;", + ), + encoding="utf-8", + ) + adapter.extracted_paths.clear() + header_changed = index.build() + self.assertEqual(4, self.build_metrics(header_changed)["invalidated_sources"]) + self.assertEqual( + [ + "include/config.hpp", + "include/detail.hpp", + "src/main.cpp", + "src/worker.cpp", + ], + sorted(adapter.extracted_paths), + ) + + entries = self.read_database(root) + arguments = cast(list[str], entries[0]["arguments"]) + arguments.append("-DREFERENCE_FLAG=1") + self.write_database(root, entries) + adapter.extracted_paths.clear() + flag_changed = index.build() + self.assertEqual(1, self.build_metrics(flag_changed)["invalidated_sources"]) + self.assertEqual(["src/main.cpp"], adapter.extracted_paths) + + equivalent = project.verify_incremental_equivalence() + self.assertEqual("ok", equivalent["status"]) + self.assertEqual(17, equivalent["node_count"]) + self.assertEqual(16, equivalent["edge_count"]) + self.assertEqual(5, equivalent["logic_projection_count"]) + + def test_add_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" / "extra.cpp" + extra.write_text( + '#include "../include/config.hpp"\n\n' + "int extra(int value) {\n" + " return demo::normalize(value);\n" + "}\n", + encoding="utf-8", + ) + entries = self.read_database(root) + entries.append( + { + "arguments": ["c++", "-std=c++20", "-c", "src/extra.cpp"], + "directory": ".", + "file": "src/extra.cpp", + } + ) + self.write_database(root, entries) + adapter.extracted_paths.clear() + added = index.build() + self.assertEqual(1, self.build_metrics(added)["reparsed_sources"]) + self.assertEqual(["src/extra.cpp"], adapter.extracted_paths) + self.assertIn( + "src/extra.cpp", + {node.source_path for node in project.load().nodes}, + ) + + worker = root / "src" / "worker.cpp" + entries = [ + entry for entry in self.read_database(root) if entry["file"] != "src/worker.cpp" + ] + self.write_database(root, entries) + worker.unlink() + adapter.extracted_paths.clear() + deleted = index.build() + self.assertEqual(1, self.build_metrics(deleted)["deleted_sources"]) + self.assertNotIn( + "src/worker.cpp", + {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( + [ + "include/config.hpp", + "include/detail.hpp", + "src/extra.cpp", + "src/main.cpp", + ], + sorted(adapter.extracted_paths), + ) + self.assertEqual("ok", project.verify_incremental_equivalence()["status"]) + + def test_database_shape_bounds_duplicates_and_paths_fail_closed(self) -> None: + invalid_documents: list[tuple[str, list[dict[str, object]], str]] = [] + baseline = cast( + list[dict[str, object]], + json.loads((FIXTURE / "compile_commands.json").read_text(encoding="utf-8")), + ) + + both = json.loads(json.dumps(baseline)) + both[0]["command"] = "c++ -c src/main.cpp" + invalid_documents.append(("both command forms", both, "invalid_compilation_database")) + + neither = json.loads(json.dumps(baseline)) + del neither[1]["command"] + invalid_documents.append(("neither command form", neither, "invalid_compilation_database")) + + bad_argument = json.loads(json.dumps(baseline)) + bad_argument[0]["arguments"] = ["c++", 7] + invalid_documents.append( + ("non-string argument", bad_argument, "invalid_compilation_database") + ) + + control_command = json.loads(json.dumps(baseline)) + control_command[1]["command"] = "c++ -c src/worker.cpp\nmalicious" + invalid_documents.append( + ("control command", control_command, "invalid_compilation_database") + ) + + unknown_key = json.loads(json.dumps(baseline)) + unknown_key[0]["environment"] = {"PATH": "/tmp"} + invalid_documents.append(("unknown key", unknown_key, "invalid_compilation_database")) + + duplicate = json.loads(json.dumps(baseline)) + duplicate.append(json.loads(json.dumps(duplicate[0]))) + invalid_documents.append(("duplicate TU", duplicate, "duplicate_translation_unit")) + + escaped_directory = json.loads(json.dumps(baseline)) + escaped_directory[0]["directory"] = ".." + invalid_documents.append(("escaped directory", escaped_directory, "path_escape")) + + escaped_file = json.loads(json.dumps(baseline)) + escaped_file[0]["file"] = "../outside.cpp" + invalid_documents.append(("escaped file", escaped_file, "path_escape")) + + for name, document, code in invalid_documents: + with self.subTest(name=name), tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + self.write_database(root, document) + adapter = self.adapter(root) + with self.assertRaises(DocForgeError) as captured: + adapter.load_manifest() + self.assertEqual(code, captured.exception.code) + + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + root = self.copy_fixture(parent) + outside_database = parent / "compile_commands.json" + outside_database.write_text("[]\n", encoding="utf-8") + database = root / "compile_commands.json" + database.unlink() + database.symlink_to(outside_database) + with self.assertRaises(DocForgeError) as database_link: + self.adapter(root) + self.assertEqual("path_escape", database_link.exception.code) + + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + root = self.copy_fixture(parent) + outside_source = parent / "worker.cpp" + outside_source.write_text("int outside() { return 1; }\n", encoding="utf-8") + worker = root / "src" / "worker.cpp" + worker.unlink() + worker.symlink_to(outside_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)) + bounded_adapters = ( + ( + CppReferenceAdapter( + root, + source_roots=("include", "src"), + compilation_database="compile_commands.json", + max_database_bytes=1, + ), + "source_too_large", + ), + ( + CppReferenceAdapter( + root, + source_roots=("include", "src"), + compilation_database="compile_commands.json", + max_source_bytes=1, + ), + "source_too_large", + ), + ( + CppReferenceAdapter( + root, + source_roots=("include", "src"), + compilation_database="compile_commands.json", + max_command_chars=32, + ), + "invalid_compilation_database", + ), + ( + CppReferenceAdapter( + root, + source_roots=("include", "src"), + compilation_database="compile_commands.json", + max_arguments=2, + ), + "invalid_compilation_database", + ), + ( + CppReferenceAdapter( + root, + source_roots=("include", "src"), + compilation_database="compile_commands.json", + max_translation_units=1, + ), + "adapter_too_large", + ), + ( + CppReferenceAdapter( + root, + source_roots=("include", "src"), + compilation_database="compile_commands.json", + max_headers=1, + ), + "adapter_too_large", + ), + ) + for adapter, code in bounded_adapters: + with self.subTest(limit=code), self.assertRaises(DocForgeError) as bounded: + adapter.load_manifest() + self.assertEqual(code, bounded.exception.code) + + def test_source_roots_no_ast_and_optional_extra_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + with self.assertRaises(DocForgeError) as translation_unit_outside: + CppReferenceAdapter( + root, + source_roots=("include",), + compilation_database="compile_commands.json", + ).load_manifest() + self.assertEqual("path_escape", translation_unit_outside.exception.code) + + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + other = root / "other" + other.mkdir() + (other / "outside.hpp").write_text( + "#pragma once\nstruct Outside {};\n", + encoding="utf-8", + ) + main = root / "src" / "main.cpp" + main.write_text( + '#include "../other/outside.hpp"\n' + main.read_text(encoding="utf-8"), + encoding="utf-8", + ) + with self.assertRaises(DocForgeError) as header_outside: + self.adapter(root).load_manifest() + self.assertEqual("path_escape", header_outside.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()) + + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + adapter = self.adapter(root) + missing = ModuleNotFoundError( + "No module named 'tree_sitter'", + name="tree_sitter", + ) + with ( + patch( + "docforge.adapters.cpp.importlib.import_module", + side_effect=missing, + ), + self.assertRaises(DocForgeError) as optional, + ): + adapter.load_manifest() + self.assertEqual("optional_dependency_missing", optional.exception.code) + self.assertEqual("cpp", optional.exception.details["extra"]) + self.assertEqual("docforge[cpp]", optional.exception.details["install"])