486 lines
20 KiB
Python
486 lines
20 KiB
Python
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"])
|