297 lines
12 KiB
Python
297 lines
12 KiB
Python
|
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()
|