138 lines
5.4 KiB
Python
138 lines
5.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from mcp.shared.memory import create_connected_server_and_client_session
|
|
from mcp.types import Tool
|
|
|
|
from docforge.cli import _parser
|
|
from docforge.command_reference import (
|
|
cli_command_references,
|
|
mcp_tool_references,
|
|
render_cli_reference_markdown,
|
|
render_command_reference_markdown,
|
|
render_mcp_reference_markdown,
|
|
)
|
|
from docforge.index import ProjectIndex
|
|
from docforge.mcp_server import ALL_TOOLS, APPLICATION_TOOLS, create_server
|
|
from docforge.project import Project
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIXTURES = ROOT / "tests" / "fixtures"
|
|
|
|
|
|
class CommandReferenceTests(unittest.IsolatedAsyncioTestCase):
|
|
def copy_fixture(self, destination: Path) -> Path:
|
|
root = destination / "alpha"
|
|
shutil.copytree(FIXTURES / "alpha", root)
|
|
return root
|
|
|
|
def test_cli_reference_is_derived_from_the_real_parser(self) -> None:
|
|
parser = _parser()
|
|
references = cli_command_references(parser)
|
|
subparsers = next(
|
|
action
|
|
for action in parser._actions # pyright: ignore[reportPrivateUsage]
|
|
if isinstance(
|
|
action,
|
|
argparse._SubParsersAction, # pyright: ignore[reportPrivateUsage]
|
|
)
|
|
)
|
|
expected = {
|
|
name: " ".join(command.format_usage().removeprefix("usage: ").split())
|
|
for name, command in subparsers.choices.items()
|
|
}
|
|
|
|
self.assertEqual(tuple(sorted(expected)), tuple(item.name for item in references))
|
|
self.assertEqual(expected, {item.name: item.invocation for item in references})
|
|
rendered = render_cli_reference_markdown(references)
|
|
self.assertEqual(rendered, render_cli_reference_markdown(reversed(references)))
|
|
self.assertIn("| `search` | `docforge search [-h] [--limit LIMIT] query` |", rendered)
|
|
self.assertIn(
|
|
"| `graph-render-status` | `docforge graph-render-status [-h] [view_id]` |",
|
|
rendered,
|
|
)
|
|
|
|
async def test_mcp_reference_is_derived_from_registered_tool_metadata(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = self.copy_fixture(Path(directory))
|
|
ProjectIndex(Project.open(root)).build()
|
|
async with create_connected_server_and_client_session(
|
|
create_server(
|
|
root,
|
|
"alpha-editor",
|
|
canonical_applier_id="alpha-editor",
|
|
),
|
|
raise_exceptions=True,
|
|
) as session:
|
|
tools = (await session.list_tools()).tools
|
|
|
|
expected_names = (*ALL_TOOLS, *APPLICATION_TOOLS)
|
|
references = mcp_tool_references(tools, expected_names=expected_names)
|
|
by_name = {item.name: item for item in references}
|
|
tools_by_name = {tool.name: tool for tool in tools}
|
|
|
|
self.assertEqual(set(expected_names), set(by_name))
|
|
for name, tool in tools_by_name.items():
|
|
with self.subTest(tool=name):
|
|
schema_hash = hashlib.sha256(
|
|
json.dumps(
|
|
tool.inputSchema,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode()
|
|
).hexdigest()
|
|
properties = set(tool.inputSchema.get("properties", {}))
|
|
required = set(tool.inputSchema.get("required", []))
|
|
reference = by_name[name]
|
|
self.assertEqual(" ".join((tool.description or "").split()), reference.description)
|
|
self.assertEqual(tuple(sorted(required)), reference.required_arguments)
|
|
self.assertEqual(
|
|
tuple(sorted(properties - required)),
|
|
reference.optional_arguments,
|
|
)
|
|
self.assertEqual(schema_hash, reference.input_schema_hash)
|
|
|
|
rendered = render_mcp_reference_markdown(references)
|
|
self.assertEqual(rendered, render_mcp_reference_markdown(reversed(references)))
|
|
self.assertIn("| read | `docforge_bootstrap` | — | — |", rendered)
|
|
self.assertIn(
|
|
"| application | `docforge_apply_changeset` | "
|
|
"`changeset_id`, `expected_changeset_hash` | — |",
|
|
rendered,
|
|
)
|
|
complete = render_command_reference_markdown(
|
|
cli_command_references(),
|
|
references,
|
|
)
|
|
self.assertEqual(1, complete.count("## CLI commands"))
|
|
self.assertEqual(1, complete.count("## MCP tools"))
|
|
|
|
def test_mcp_reference_fails_closed_on_registration_drift(self) -> None:
|
|
known = Tool(
|
|
name="docforge_bootstrap",
|
|
description="Bootstrap.",
|
|
inputSchema={"type": "object", "properties": {}},
|
|
)
|
|
unknown = Tool(
|
|
name="docforge_unregistered",
|
|
description="Unknown.",
|
|
inputSchema={"type": "object", "properties": {}},
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "no declared capability surface"):
|
|
mcp_tool_references((unknown,))
|
|
with self.assertRaisesRegex(ValueError, "missing="):
|
|
mcp_tool_references((known,), expected_names=ALL_TOOLS)
|
|
with self.assertRaisesRegex(ValueError, "repeats"):
|
|
mcp_tool_references((known, known))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|