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"])