282 lines
12 KiB
Python
282 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import cast
|
|
from unittest import mock
|
|
|
|
import docforge.adapters.python as python_adapter
|
|
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()
|
|
original_parse = python_adapter.ast.parse
|
|
with mock.patch.object(
|
|
python_adapter.ast,
|
|
"parse",
|
|
wraps=original_parse,
|
|
) as parsed:
|
|
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)
|
|
parsed.assert_not_called()
|
|
|
|
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()
|
|
with mock.patch.object(
|
|
python_adapter.ast,
|
|
"parse",
|
|
wraps=original_parse,
|
|
) as parsed:
|
|
changed = index.build()
|
|
self.assertEqual(4, self.build_metrics(changed)["invalidated_sources"])
|
|
self.assertGreater(parsed.call_count, 0)
|
|
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())
|