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())