Add deterministic Python reference adapter
This commit is contained in:
parent
e6523c0c00
commit
8956c5c5f5
7 changed files with 1156 additions and 0 deletions
267
tests/test_python_reference_adapter.py
Normal file
267
tests/test_python_reference_adapter.py
Normal file
|
|
@ -0,0 +1,267 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from typing import cast
|
||||
|
||||
from docforge.adapter_sdk import (
|
||||
AdapterProject,
|
||||
AdapterSource,
|
||||
AdapterSourceProjection,
|
||||
verify_adapter_conformance,
|
||||
)
|
||||
from docforge.adapters.python import (
|
||||
PYTHON_ADAPTER_ID,
|
||||
PYTHON_ADAPTER_VERSION,
|
||||
PYTHON_EXTRACTOR_VERSION,
|
||||
PythonReferenceAdapter,
|
||||
)
|
||||
from docforge.errors import DocForgeError
|
||||
from docforge.index import ProjectIndex
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
FIXTURE = ROOT / "tests" / "fixtures" / "reference-python"
|
||||
|
||||
|
||||
class RecordingPythonAdapter(PythonReferenceAdapter):
|
||||
def __init__(self, root: Path) -> None:
|
||||
super().__init__(
|
||||
root,
|
||||
source_roots=("src",),
|
||||
project_id="reference-python",
|
||||
title="Reference Python fixture",
|
||||
)
|
||||
self.extracted_paths: list[str] = []
|
||||
|
||||
def extract_source(self, source: AdapterSource) -> AdapterSourceProjection:
|
||||
self.extracted_paths.append(source.source_path)
|
||||
return super().extract_source(source)
|
||||
|
||||
|
||||
class PythonReferenceAdapterTests(unittest.TestCase):
|
||||
def copy_fixture(self, parent: Path) -> Path:
|
||||
root = parent / "reference-python"
|
||||
shutil.copytree(FIXTURE, root)
|
||||
return root.resolve()
|
||||
|
||||
def adapter(self, root: Path) -> RecordingPythonAdapter:
|
||||
return RecordingPythonAdapter(root)
|
||||
|
||||
@staticmethod
|
||||
def build_metrics(result: dict[str, object]) -> dict[str, object]:
|
||||
return cast(dict[str, object], result["build"])
|
||||
|
||||
def test_complete_assembly_is_exact_ordered_cache_independent_and_scoped(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = self.copy_fixture(Path(directory))
|
||||
adapter = self.adapter(root)
|
||||
|
||||
first = adapter.load_assembly()
|
||||
second = adapter.load_complete_assembly()
|
||||
projection = adapter.load_projection()
|
||||
|
||||
self.assertEqual(first, second)
|
||||
self.assertEqual(first.projection, projection)
|
||||
self.assertEqual(PYTHON_ADAPTER_ID, projection.adapter_id)
|
||||
self.assertEqual(PYTHON_ADAPTER_VERSION, projection.adapter_version)
|
||||
self.assertEqual(14, len(projection.nodes))
|
||||
self.assertEqual(13, len(projection.edges))
|
||||
self.assertEqual(5, len(first.logic))
|
||||
self.assertEqual(
|
||||
list(projection.nodes),
|
||||
sorted(projection.nodes, key=lambda item: item.node.node_id),
|
||||
)
|
||||
self.assertEqual(
|
||||
{"contains", "depends_on"}, {item.edge.relation for item in projection.edges}
|
||||
)
|
||||
self.assertFalse(any(item.node.title == "json" for item in projection.nodes))
|
||||
self.assertTrue(
|
||||
any(
|
||||
item.node.title == "Service" and ("kind", "class") in item.metadata
|
||||
for item in projection.nodes
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
any(
|
||||
item.node.title == "Service.run" and ("kind", "function") in item.metadata
|
||||
for item in projection.nodes
|
||||
)
|
||||
)
|
||||
|
||||
report = adapter.support_report()
|
||||
self.assertEqual(1, report["schema_version"])
|
||||
self.assertEqual(PYTHON_EXTRACTOR_VERSION, report["extractor_version"])
|
||||
self.assertFalse(report["imports_project_code"])
|
||||
self.assertFalse(report["executes_project_code"])
|
||||
self.assertEqual(
|
||||
[
|
||||
"call_resolution",
|
||||
"dynamic_import_resolution",
|
||||
"inheritance_resolution",
|
||||
"runtime_generated_facts",
|
||||
"symbol_reference_resolution",
|
||||
],
|
||||
[item.code for item in adapter.unsupported_facts()],
|
||||
)
|
||||
self.assertFalse((root / ".cache").exists())
|
||||
|
||||
conformance = verify_adapter_conformance(
|
||||
adapter,
|
||||
cache_root=root / ".cache" / "conformance",
|
||||
)
|
||||
self.assertEqual("reference-python", conformance.project_id)
|
||||
self.assertEqual(14, conformance.node_count)
|
||||
self.assertEqual(13, conformance.edge_count)
|
||||
self.assertEqual(5, conformance.logic_projection_count)
|
||||
self.assertTrue(conformance.incremental)
|
||||
|
||||
def test_incremental_cold_warm_reverse_dependencies_and_logic_equivalence(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = self.copy_fixture(Path(directory))
|
||||
adapter = self.adapter(root)
|
||||
project = AdapterProject(adapter, cache_root=root / ".cache" / "incremental")
|
||||
index = ProjectIndex(project)
|
||||
|
||||
cold = index.build()
|
||||
self.assertEqual(4, self.build_metrics(cold)["reparsed_sources"])
|
||||
self.assertEqual(
|
||||
[
|
||||
"src/sample/__init__.py",
|
||||
"src/sample/service.py",
|
||||
"src/sample/shared.py",
|
||||
"src/sample/worker.py",
|
||||
],
|
||||
sorted(adapter.extracted_paths),
|
||||
)
|
||||
|
||||
adapter.extracted_paths.clear()
|
||||
warm = index.build()
|
||||
self.assertEqual(4, self.build_metrics(warm)["cache_hits"])
|
||||
self.assertEqual(0, self.build_metrics(warm)["reparsed_sources"])
|
||||
self.assertEqual([], adapter.extracted_paths)
|
||||
|
||||
shared = root / "src" / "sample" / "shared.py"
|
||||
shared.write_text(
|
||||
shared.read_text(encoding="utf-8").replace(
|
||||
"DEFAULT_LIMIT = 3", "DEFAULT_LIMIT = 5"
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
adapter.extracted_paths.clear()
|
||||
changed = index.build()
|
||||
self.assertEqual(4, self.build_metrics(changed)["invalidated_sources"])
|
||||
self.assertEqual(
|
||||
[
|
||||
"src/sample/__init__.py",
|
||||
"src/sample/service.py",
|
||||
"src/sample/shared.py",
|
||||
"src/sample/worker.py",
|
||||
],
|
||||
sorted(adapter.extracted_paths),
|
||||
)
|
||||
|
||||
equivalent = project.verify_incremental_equivalence()
|
||||
self.assertEqual("ok", equivalent["status"])
|
||||
self.assertEqual(14, equivalent["node_count"])
|
||||
self.assertEqual(13, equivalent["edge_count"])
|
||||
self.assertEqual(5, equivalent["logic_projection_count"])
|
||||
|
||||
def test_add_rename_delete_and_corrupt_cache_recover_without_stale_facts(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = self.copy_fixture(Path(directory))
|
||||
adapter = self.adapter(root)
|
||||
cache_root = root / ".cache" / "incremental"
|
||||
project = AdapterProject(adapter, cache_root=cache_root)
|
||||
index = ProjectIndex(project)
|
||||
index.build()
|
||||
|
||||
extra = root / "src" / "sample" / "extra.py"
|
||||
extra.write_text(
|
||||
'"""Added source."""\n\n\ndef extra() -> str:\n return "extra"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
adapter.extracted_paths.clear()
|
||||
added = index.build()
|
||||
self.assertEqual(1, self.build_metrics(added)["reparsed_sources"])
|
||||
self.assertEqual(["src/sample/extra.py"], adapter.extracted_paths)
|
||||
self.assertIn(
|
||||
"src/sample/extra.py",
|
||||
{node.source_path for node in project.load().nodes},
|
||||
)
|
||||
|
||||
worker = root / "src" / "sample" / "worker.py"
|
||||
runner = root / "src" / "sample" / "runner.py"
|
||||
worker.rename(runner)
|
||||
adapter.extracted_paths.clear()
|
||||
renamed = index.build()
|
||||
self.assertEqual(1, self.build_metrics(renamed)["deleted_sources"])
|
||||
self.assertEqual(1, self.build_metrics(renamed)["reparsed_sources"])
|
||||
renamed_paths = {node.source_path for node in project.load().nodes}
|
||||
self.assertNotIn("src/sample/worker.py", renamed_paths)
|
||||
self.assertIn("src/sample/runner.py", renamed_paths)
|
||||
|
||||
extra.unlink()
|
||||
adapter.extracted_paths.clear()
|
||||
deleted = index.build()
|
||||
self.assertEqual(1, self.build_metrics(deleted)["deleted_sources"])
|
||||
self.assertNotIn(
|
||||
"src/sample/extra.py",
|
||||
{node.source_path for node in project.load().nodes},
|
||||
)
|
||||
|
||||
(cache_root / "extractions.json").write_text("{broken", encoding="utf-8")
|
||||
adapter.extracted_paths.clear()
|
||||
recovered = index.build()
|
||||
self.assertEqual(4, self.build_metrics(recovered)["reparsed_sources"])
|
||||
self.assertEqual(
|
||||
[
|
||||
"src/sample/__init__.py",
|
||||
"src/sample/runner.py",
|
||||
"src/sample/service.py",
|
||||
"src/sample/shared.py",
|
||||
],
|
||||
sorted(adapter.extracted_paths),
|
||||
)
|
||||
self.assertEqual("ok", project.verify_incremental_equivalence()["status"])
|
||||
|
||||
def test_confinement_and_no_ast_policy_reject_logic_without_importing_code(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
parent = Path(directory)
|
||||
root = self.copy_fixture(parent)
|
||||
|
||||
for source_root in (root / "src", Path("../outside")):
|
||||
with self.subTest(source_root=source_root):
|
||||
with self.assertRaises(DocForgeError) as captured:
|
||||
PythonReferenceAdapter(root, source_roots=(source_root,))
|
||||
self.assertEqual("path_escape", captured.exception.code)
|
||||
|
||||
outside = parent / "outside"
|
||||
outside.mkdir()
|
||||
linked_root = root / "linked"
|
||||
linked_root.symlink_to(outside, target_is_directory=True)
|
||||
with self.assertRaises(DocForgeError) as linked:
|
||||
PythonReferenceAdapter(root, source_roots=("linked",))
|
||||
self.assertEqual("path_escape", linked.exception.code)
|
||||
|
||||
external_source = outside / "external.py"
|
||||
external_source.write_text("def outside():\n return True\n", encoding="utf-8")
|
||||
worker = root / "src" / "sample" / "worker.py"
|
||||
worker.unlink()
|
||||
worker.symlink_to(external_source)
|
||||
adapter = self.adapter(root)
|
||||
with self.assertRaises(DocForgeError) as source_link:
|
||||
adapter.load_manifest()
|
||||
self.assertEqual("path_escape", source_link.exception.code)
|
||||
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = self.copy_fixture(Path(directory))
|
||||
adapter = self.adapter(root)
|
||||
project = AdapterProject(adapter, cache_root=root / ".cache" / "no-ast")
|
||||
|
||||
with self.assertRaises(DocForgeError) as no_ast:
|
||||
ProjectIndex(project, allow_logic=False).build()
|
||||
|
||||
self.assertEqual("adapter_policy_forbids_logic", no_ast.exception.code)
|
||||
self.assertFalse(project.descriptor.index_path.exists())
|
||||
Loading…
Add table
Add a link
Reference in a new issue