from __future__ import annotations import shutil import sys import tempfile import unittest from pathlib import Path from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.shared.memory import create_connected_server_and_client_session from docforge.index import ProjectIndex from docforge.mcp_server import ALL_TOOLS, CONTENT_WARNING, PROPOSAL_TOOLS, create_server from docforge.project import Project ROOT = Path(__file__).resolve().parents[1] FIXTURES = ROOT / "tests" / "fixtures" class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase): def copy_fixture(self, name: str, destination: Path) -> Path: root = destination / name shutil.copytree(FIXTURES / name, root) return root async def test_protocol_lists_only_the_fixed_safe_surface(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: response = await session.list_tools() names = tuple(tool.name for tool in response.tools) self.assertEqual(ALL_TOOLS, names) self.assertEqual(10, len(PROPOSAL_TOOLS)) self.assertFalse( any( token in name for name in names for token in ("apply", "commit", "push", "deploy", "publish", "shell") ) ) async def test_every_read_tool_returns_scoped_structured_results(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() calls = ( ("docforge_project_info", {}), ("docforge_get_contract", {}), ("docforge_get_node", {"node_id": "guide.workflow"}), ("docforge_search", {"query": "canonical nodes", "limit": 5}), ("docforge_filter_nodes", {"family": "proof", "tag": "validation"}), ("docforge_backlinks", {"node_id": "guide.workflow"}), ("docforge_dependencies", {"node_id": "guide.workflow", "depth": 2}), ("docforge_impact", {"node_id": "guide.foundation", "depth": 2}), ("docforge_get_context", {"profile": "active", "budget": 180}), ("docforge_validate_project", {}), ("docforge_render_status", {}), ("docforge_visualize", {"node_id": "guide.workflow", "depth": 1}), ) async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: results = [await session.call_tool(name, arguments) for name, arguments in calls] for result in results: self.assertFalse(result.isError) self.assertIsNotNone(result.structuredContent) payload = result.structuredContent self.assertEqual("ok", payload["status"]) self.assertEqual("alpha-docs", payload["project_id"]) self.assertEqual(CONTENT_WARNING, payload["content_warning"]) self.assertTrue(payload["project_root_fingerprint"]) self.assertEqual("current", payload["staleness"]) contract = results[1].structuredContent self.assertFalse(contract["canonical_writes_allowed"]) self.assertFalse(contract["project_switching_allowed"]) self.assertIn("canonical_writes", contract["excluded_operations"]) self.assertIn("arbitrary_renderer_execution", contract["excluded_operations"]) self.assertFalse(contract["isolated_changeset_writes_allowed"]) self.assertFalse(contract["proposal_access"]["enabled"]) self.assertTrue(results[10].structuredContent["configured"]) self.assertEqual("stale", results[10].structuredContent["state"]) visualization = results[11].structuredContent["visualization"] self.assertTrue(visualization["read_only"]) self.assertTrue(visualization["project_bound"]) self.assertEqual("graph-browser@1", visualization["template"]) self.assertTrue(visualization["url"].startswith("http://127.0.0.1:")) context = results[8].structuredContent self.assertLessEqual(context["estimated_tokens"], 180) self.assertTrue(context["omissions"]) async def test_missing_node_and_stale_index_are_structured_failures(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() server = create_server(root) async with create_connected_server_and_client_session( server, raise_exceptions=True ) as session: missing = await session.call_tool( "docforge_get_node", {"node_id": "research.question"} ) workflow = root / "docs" / "content" / "workflow.md" workflow.write_text( workflow.read_text(encoding="utf-8") + "\nChanged after startup.\n", encoding="utf-8", ) stale = await session.call_tool("docforge_get_node", {"node_id": "guide.workflow"}) self.assertEqual("missing_node", missing.structuredContent["error"]["code"]) self.assertEqual("stale_index", stale.structuredContent["error"]["code"]) self.assertEqual("current", missing.structuredContent["staleness"]) self.assertEqual("stale", stale.structuredContent["staleness"]) self.assertTrue(missing.structuredContent["source_hash"]) self.assertTrue(stale.structuredContent["source_hash"]) self.assertFalse(missing.isError) self.assertFalse(stale.isError) async def test_output_limit_fails_without_returning_partial_content(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) descriptor = root / ".docforge" / "project.toml" descriptor.write_text( descriptor.read_text(encoding="utf-8").replace( "max_context_tokens = 2000", "max_context_tokens = 2000\nmax_tool_output_chars = 700", ), encoding="utf-8", ) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: result = await session.call_tool("docforge_get_contract", {}) payload = result.structuredContent self.assertEqual("error", payload["status"]) self.assertEqual("result_too_large", payload["error"]["code"]) self.assertNotIn("canonical_paths", payload) async def test_proposal_tools_use_fixed_writer_and_never_change_canonical_content(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) project = Project.open(root) ProjectIndex(project).build() canonical_before = { path.relative_to(root).as_posix(): path.read_bytes() for path in (root / "docs/content").glob("*") if path.is_file() } node_hashes = {node.node_id: node.content_hash for node in project.load().nodes} async with create_connected_server_and_client_session( create_server(root, "alpha-editor"), raise_exceptions=True ) as session: contract = await session.call_tool("docforge_get_contract", {}) created = await session.call_tool( "docforge_create_changeset", {"changeset_id": "mcp-update"} ) updated = await session.call_tool( "docforge_propose_node_update", { "changeset_id": "mcp-update", "expected_changeset_hash": created.structuredContent["changeset_hash"], "node_id": "guide.workflow", "expected_content_hash": node_hashes["guide.workflow"], "metadata": {"summary": "A proposal written through MCP."}, "content": None, "relationship_changes": [], "rationale": "Prove fixed-writer isolated proposal access.", }, ) created_node = await session.call_tool( "docforge_propose_node_create", { "changeset_id": "mcp-update", "expected_changeset_hash": updated.structuredContent["changeset_hash"], "node_id": "guide.mcp-node", "target_source": "docs/content/mcp-node.md", "metadata": { "title": "MCP proposal node", "family": "guide", "authority": "proposal", "status": "active", "tags": ["mcp", "proposal"], "summary": "A node creation proposed through MCP.", }, "content": "This node is not canonical until external integration.", "relationship_changes": [], "rationale": "Prove isolated MCP creation.", }, ) moved = await session.call_tool( "docforge_propose_node_move", { "changeset_id": "mcp-update", "expected_changeset_hash": created_node.structuredContent["changeset_hash"], "node_id": "guide.foundation", "expected_content_hash": node_hashes["guide.foundation"], "target_source": "docs/content/foundation-moved.md", "rationale": "Prove isolated MCP movement.", }, ) deleted = await session.call_tool( "docforge_propose_node_delete", { "changeset_id": "mcp-update", "expected_changeset_hash": moved.structuredContent["changeset_hash"], "node_id": "proof.validation", "expected_content_hash": node_hashes["proof.validation"], "relationship_changes": [ { "action": "remove", "source_id": "proof.validation", "relation": "proves", "target_id": "guide.workflow", } ], "rationale": "Prove isolated MCP deletion.", }, ) validated = await session.call_tool( "docforge_validate_changeset", {"changeset_id": "mcp-update"} ) listed = await session.call_tool("docforge_list_changesets", {}) inspected = await session.call_tool( "docforge_get_changeset", {"changeset_id": "mcp-update"} ) diff = await session.call_tool( "docforge_get_changeset_diff", {"changeset_id": "mcp-update"} ) preview = await session.call_tool( "docforge_preview_changeset", {"changeset_id": "mcp-update", "view_id": "manual"}, ) undeclared = await session.call_tool( "docforge_preview_changeset", {"changeset_id": "mcp-update", "view_id": "not-declared"}, ) self.assertTrue(contract.structuredContent["proposal_access"]["enabled"]) self.assertEqual( "alpha-editor", contract.structuredContent["proposal_access"]["writer"] ) self.assertTrue(contract.structuredContent["isolated_changeset_writes_allowed"]) self.assertFalse(contract.structuredContent["canonical_writes_allowed"]) self.assertEqual("alpha-editor", created.structuredContent["creator"]) self.assertEqual(1, updated.structuredContent["operation_count"]) self.assertEqual(4, deleted.structuredContent["operation_count"]) self.assertTrue(validated.structuredContent["valid"]) self.assertEqual(1, listed.structuredContent["count"]) self.assertEqual( "mcp-update", listed.structuredContent["changesets"][0]["changeset_id"] ) self.assertEqual("current", inspected.structuredContent["base_state"]) self.assertEqual(4, inspected.structuredContent["operation_count"]) self.assertEqual( ["update", "create", "move", "delete"], [change["operation"] for change in diff.structuredContent["changes"]], ) self.assertEqual("current", preview.structuredContent["state"]) self.assertEqual( ".docforge/previews/mcp-update/manual.html", preview.structuredContent["preview"]["path"], ) self.assertTrue(preview.structuredContent["preview_identity"]) self.assertEqual("error", undeclared.structuredContent["status"]) self.assertEqual("unknown_render_view", undeclared.structuredContent["error"]["code"]) canonical_after = { path.relative_to(root).as_posix(): path.read_bytes() for path in (root / "docs/content").glob("*") if path.is_file() } self.assertEqual(canonical_before, canonical_after) self.assertFalse((root / "docs/content/mcp-node.md").exists()) self.assertTrue((root / ".docforge/changesets/mcp-update.json").is_file()) self.assertTrue((root / ".docforge/previews/mcp-update/manual.html").is_file()) self.assertFalse((root / ".docforge/rendered/manual.html").exists()) async def test_server_without_writer_rejects_proposal_mutation_structurally(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: result = await session.call_tool( "docforge_create_changeset", {"changeset_id": "disabled"} ) self.assertFalse(result.isError) self.assertEqual("error", result.structuredContent["status"]) self.assertEqual("proposal_access_disabled", result.structuredContent["error"]["code"]) self.assertFalse((root / ".docforge/changesets/disabled.json").exists()) async def test_stdio_transport_serves_the_same_project_bound_contract(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("beta", Path(directory)) ProjectIndex(Project.open(root)).build() parameters = StdioServerParameters( command=sys.executable, args=["-m", "docforge.mcp_server", "--project-root", str(root)], ) async with ( stdio_client(parameters) as (read, write), ClientSession(read, write) as session, ): await session.initialize() result = await session.call_tool("docforge_project_info", {}) self.assertFalse(result.isError) self.assertEqual("beta-notes", result.structuredContent["project_id"]) self.assertEqual(1, result.structuredContent["node_count"]) if __name__ == "__main__": unittest.main()