1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages
DocForge2/tests/test_cpp_reference_adapter.py

487 lines
20 KiB
Python
Raw Permalink Normal View History

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