from __future__ import annotations import io import json import shutil import sys import tempfile import unittest from contextlib import redirect_stderr from pathlib import Path from unittest import mock from jsonschema import Draft202012Validator from mcp.shared.memory import create_connected_server_and_client_session from docforge.adapter_launcher import AdapterLauncherV1 from docforge.client_config import generate_adapter_client_configuration from docforge.errors import DocForgeError from docforge.index import ProjectIndex from docforge.mcp_server import READ_TOOLS from docforge.reference_config import ( REFERENCE_ADAPTER_CONFIG, load_reference_adapter_config, ) from docforge.reference_mcp import ( REFERENCE_MCP_MODULE, create_reference_project, create_reference_server, main, ) ROOT = Path(__file__).resolve().parents[1] FIXTURE = ROOT / "tests" / "fixtures" / "reference-python" CPP_FIXTURE = ROOT / "tests" / "fixtures" / "reference-cpp" SCHEMA = json.loads( (ROOT / "schemas" / "reference-adapter.schema.json").read_text(encoding="utf-8") ) class ReferenceMcpTests(unittest.IsolatedAsyncioTestCase): def copy_fixture(self, parent: Path) -> Path: root = parent / "reference-python" shutil.copytree(FIXTURE, root) return root.resolve() def copy_cpp_fixture(self, parent: Path) -> Path: root = parent / "reference-cpp" shutil.copytree(CPP_FIXTURE, root) return root.resolve() @staticmethod def write_config( root: Path, *, language: str = "python", source_roots: tuple[str, ...] = ("src",), extra: str = "", ) -> Path: config = root.joinpath(*REFERENCE_ADAPTER_CONFIG.parts) config.parent.mkdir(parents=True, exist_ok=True) quoted_roots = ", ".join(json.dumps(item) for item in source_roots) config.write_text( "\n".join( ( "schema_version = 1", 'project_id = "reference-python"', 'title = "Runnable Python reference"', f"language = {json.dumps(language)}", f"source_roots = [{quoted_roots}]", extra, "", ) ), encoding="utf-8", ) return config async def test_python_config_build_check_and_real_mcp_retrieval(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) self.write_config(root) config = load_reference_adapter_config(root) Draft202012Validator(SCHEMA).validate(config.as_dict()) self.assertEqual("python", config.language) self.assertEqual((root / "src",), config.source_roots) self.assertIsNone(config.compilation_database) self.assertEqual(64, len(config.config_hash)) project = create_reference_project(root) index = ProjectIndex(project) built = index.build() checked = index.check() self.assertEqual("ok", built["status"]) self.assertEqual("ok", checked["status"]) self.assertTrue(project.descriptor.index_path.is_relative_to(root / ".docforge")) async with create_connected_server_and_client_session( create_reference_server(root, capability_mode="read"), raise_exceptions=True, ) as session: tools = tuple(tool.name for tool in (await session.list_tools()).tools) bootstrap = await session.call_tool("docforge_bootstrap", {}) search = await session.call_tool( "docforge_search", {"query": "Service", "limit": 5}, ) node_id = search.structuredContent["results"][0]["node_id"] node = await session.call_tool("docforge_get_node", {"node_id": node_id}) self.assertEqual(READ_TOOLS, tools) self.assertEqual("ok", bootstrap.structuredContent["status"]) self.assertEqual( REFERENCE_MCP_MODULE, bootstrap.structuredContent["binding"]["server_module"], ) self.assertEqual( "python", bootstrap.structuredContent["binding"]["reference_language"], ) self.assertGreater(search.structuredContent["count"], 0) self.assertEqual(node_id, node.structuredContent["node"]["node_id"]) async def test_cpp_config_build_check_and_real_mcp_retrieval(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_cpp_fixture(Path(directory)) self.write_config( root, language="cpp", source_roots=("include", "src"), extra='compilation_database = "compile_commands.json"', ) config = load_reference_adapter_config(root) Draft202012Validator(SCHEMA).validate(config.as_dict()) project = create_reference_project(root) index = ProjectIndex(project) self.assertEqual("ok", index.build()["status"]) self.assertEqual("ok", index.check()["status"]) async with create_connected_server_and_client_session( create_reference_server(root, capability_mode="read"), raise_exceptions=True, ) as session: bootstrap = await session.call_tool("docforge_bootstrap", {}) search = await session.call_tool( "docforge_search", {"query": "worker", "limit": 5}, ) node_id = search.structuredContent["results"][0]["node_id"] node = await session.call_tool("docforge_get_node", {"node_id": node_id}) self.assertEqual("cpp", config.language) self.assertEqual(root / "compile_commands.json", config.compilation_database) self.assertEqual( "cpp", bootstrap.structuredContent["binding"]["reference_language"], ) self.assertEqual(node_id, node.structuredContent["node"]["node_id"]) def test_launcher_generates_exact_isolated_module_invocation(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) self.write_config(root) project = create_reference_project(root) launcher = AdapterLauncherV1.for_project(project, module=REFERENCE_MCP_MODULE) with mock.patch("docforge.client_config.subprocess.run") as executed: result = generate_adapter_client_configuration( project, launcher, "codex", capability_mode="read", ) self.assertEqual( [ "-I", "-m", REFERENCE_MCP_MODULE, "--project-root", str(root), "--capability-mode", "read", ], result["binding"]["args"], ) self.assertEqual({}, result["binding"]["environment"]) self.assertNotIn("cwd", result["binding"]) executed.assert_not_called() def test_config_drift_requires_runtime_restart(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) config_path = self.write_config(root) project = create_reference_project(root) project.validate_runtime() config_path.write_text( config_path.read_text(encoding="utf-8").replace( "Runnable Python reference", "Changed Python reference", ), encoding="utf-8", ) with self.assertRaises(DocForgeError) as captured: project.validate_runtime() self.assertEqual("adapter_restart_required", captured.exception.code) self.assertEqual( [REFERENCE_ADAPTER_CONFIG.as_posix()], captured.exception.details["changed"], ) def test_config_rejects_unknown_fields_languages_and_language_mismatch(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) for language, extra, code in ( ("ruby", "", "unsupported_reference_language"), ("javascript", "", "reference_language_mismatch"), ("python", 'command = "python project.py"', "invalid_reference_config"), ( "python", 'compilation_database = "compile_commands.json"', "invalid_reference_config", ), ("cpp", "", "invalid_reference_config"), ): with self.subTest(language=language, extra=extra): self.write_config(root, language=language, extra=extra) with self.assertRaises(DocForgeError) as captured: load_reference_adapter_config(root) self.assertEqual(code, captured.exception.code) def test_config_rejects_path_escape_and_source_symlinks(self) -> None: with tempfile.TemporaryDirectory() as directory: parent = Path(directory) root = self.copy_fixture(parent) outside = parent / "outside" outside.mkdir() self.write_config(root, source_roots=("../outside",)) with self.assertRaises(DocForgeError) as escaped: load_reference_adapter_config(root) self.assertEqual("path_escape", escaped.exception.code) linked = root / "linked" linked.symlink_to(outside, target_is_directory=True) self.write_config(root, source_roots=("linked",)) with self.assertRaises(DocForgeError) as linked_root: load_reference_adapter_config(root) self.assertEqual("path_escape", linked_root.exception.code) self.write_config(root) external_source = outside / "external.py" external_source.write_text("def external(): pass\n", encoding="utf-8") source_link = root / "src" / "external.py" source_link.symlink_to(external_source) with self.assertRaises(DocForgeError) as linked_source: load_reference_adapter_config(root) self.assertEqual("path_escape", linked_source.exception.code) def test_config_file_must_be_a_regular_non_symlink(self) -> None: with tempfile.TemporaryDirectory() as directory: parent = Path(directory) root = self.copy_fixture(parent) config_path = self.write_config(root) outside = parent / "reference-adapter.toml" shutil.copyfile(config_path, outside) config_path.unlink() config_path.symlink_to(outside) with self.assertRaises(DocForgeError) as captured: load_reference_adapter_config(root) self.assertEqual("path_escape", captured.exception.code) def test_reference_module_cli_advertises_read_mode_only(self) -> None: with ( redirect_stderr(io.StringIO()), mock.patch.object( sys, "argv", [ REFERENCE_MCP_MODULE, "--project-root", "/irrelevant", "--capability-mode", "proposal", ], ), mock.patch("docforge.reference_mcp.create_reference_server") as create_server, self.assertRaises(SystemExit) as captured, ): main() self.assertEqual(2, captured.exception.code) create_server.assert_not_called()