administrator/DocForge2
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages

Manage graph viewers with a supervised local service

Browse source
This commit is contained in:
Andraxion 2026-07-25 00:17:21 -04:00
parent eb48ba1a51
commit 7c87536167
14 changed files with 1027 additions and 103 deletions
Download patch file Download diff file Expand all files Collapse all files

51
tests/test_mcp_server.py
View file

@ -1,9 +1,13 @@
from __future__ import annotations
import os
import shutil
import sys
import tempfile
import threading
import time
import unittest
from contextlib import contextmanager
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
@ -20,6 +24,7 @@ from docforge.mcp_server import (
create_server,
)
from docforge.project import Project
from docforge.viewer_manager import ViewerManager
ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"
@ -31,6 +36,27 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
shutil.copytree(FIXTURES / name, root)
return root
@contextmanager
def running_manager(self, state_path: Path):
manager = ViewerManager(state_path, check_interval_seconds=0.02)
thread = threading.Thread(target=manager.serve_forever, daemon=True)
previous = os.environ.get("DOCFORGE_VIEWER_MANAGER_STATE")
os.environ["DOCFORGE_VIEWER_MANAGER_STATE"] = str(state_path)
thread.start()
deadline = time.monotonic() + 2
while not state_path.exists() and time.monotonic() < deadline:
time.sleep(0.01)
self.assertTrue(state_path.exists())
try:
yield
finally:
manager.shutdown()
thread.join(timeout=2)
if previous is None:
os.environ.pop("DOCFORGE_VIEWER_MANAGER_STATE", None)
else:
os.environ["DOCFORGE_VIEWER_MANAGER_STATE"] = previous
async def test_protocol_lists_only_the_fixed_safe_surface(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("alpha", Path(directory))
@ -69,17 +95,19 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
("docforge_render_status", {}),
("docforge_visualize", {"node_id": "guide.workflow", "depth": 1}),
("docforge_stop_visualization", {}),
("docforge_visualization_status", {}),
)
service = DocForgeService(Project.open(root))
try:
async with create_connected_server_and_client_session(
_create_bound_server(service, read_only=True), raise_exceptions=True
) as session:
results = [
await session.call_tool(name, arguments) for name, arguments in calls
]
finally:
service.visualization.stop()
with self.running_manager(Path(directory) / "viewer-manager.json"):
service = DocForgeService(Project.open(root))
try:
async with create_connected_server_and_client_session(
_create_bound_server(service, read_only=True), raise_exceptions=True
) as session:
results = [
await session.call_tool(name, arguments) for name, arguments in calls
]
finally:
service.visualization.stop()
for result in results:
self.assertFalse(result.isError)
@ -103,10 +131,11 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
self.assertTrue(visualization["read_only"])
self.assertTrue(visualization["project_bound"])
self.assertEqual("graph-browser@8", visualization["template"])
self.assertEqual("explicit_stop", visualization["lifetime"]["policy"])
self.assertEqual("managed_idle", visualization["lifetime"]["policy"])
self.assertEqual("docforge_stop_visualization", visualization["lifetime"]["stop_tool"])
self.assertTrue(visualization["url"].startswith("http://127.0.0.1:"))
self.assertEqual("stopped", results[12].structuredContent["state"])
self.assertEqual("not_running", results[13].structuredContent["state"])
context = results[8].structuredContent
self.assertLessEqual(context["estimated_tokens"], 180)
self.assertTrue(context["omissions"])