307 lines
14 KiB
Python
307 lines
14 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import hashlib
|
||
|
|
import re
|
||
|
|
import shutil
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from contextlib import AsyncExitStack
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from mcp import ClientSession, StdioServerParameters
|
||
|
|
from mcp.client.stdio import stdio_client
|
||
|
|
|
||
|
|
from docforge.changesets import ChangesetStore
|
||
|
|
from docforge.context import compile_context
|
||
|
|
from docforge.index import ProjectIndex
|
||
|
|
from docforge.project import Project
|
||
|
|
from docforge.rendering import RenderService
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
FIXTURES = ROOT / "tests" / "fixtures"
|
||
|
|
|
||
|
|
|
||
|
|
def copy_fixture(name: str, destination: Path) -> Path:
|
||
|
|
root = destination / name
|
||
|
|
shutil.copytree(FIXTURES / name, root)
|
||
|
|
return root
|
||
|
|
|
||
|
|
|
||
|
|
def canonical_digest(root: Path) -> str:
|
||
|
|
paths = [
|
||
|
|
root / ".docforge/project.toml",
|
||
|
|
*(path for path in root.rglob("*") if path.is_file() and ".docforge" not in path.parts),
|
||
|
|
]
|
||
|
|
digest = hashlib.sha256()
|
||
|
|
for path in sorted(set(paths), key=lambda item: item.relative_to(root).as_posix()):
|
||
|
|
digest.update(path.relative_to(root).as_posix().encode())
|
||
|
|
digest.update(path.read_bytes())
|
||
|
|
return digest.hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def node_hash(project: Project, node_id: str) -> str:
|
||
|
|
return next(node.content_hash for node in project.load().nodes if node.node_id == node_id)
|
||
|
|
|
||
|
|
|
||
|
|
class UnrelatedProjectProofTests(unittest.TestCase):
|
||
|
|
def test_generic_ski_project_completes_the_full_workflow(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
root = copy_fixture("awesome-ski-game", Path(directory))
|
||
|
|
before = canonical_digest(root)
|
||
|
|
project = Project.open(root)
|
||
|
|
snapshot = project.load()
|
||
|
|
|
||
|
|
self.assertEqual("generic", project.descriptor.adapter)
|
||
|
|
self.assertEqual("awesome-ski-game", snapshot.descriptor.project_id)
|
||
|
|
self.assertEqual(
|
||
|
|
{"proof", "riding", "safety", "session", "trail"},
|
||
|
|
{node.family for node in snapshot.nodes},
|
||
|
|
)
|
||
|
|
self.assertEqual(5, len(snapshot.nodes))
|
||
|
|
self.assertEqual(6, len(snapshot.edges))
|
||
|
|
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
built = index.build()
|
||
|
|
self.assertEqual(built, index.check())
|
||
|
|
self.assertEqual(
|
||
|
|
["proof.carve-playtest", "riding.edge-control"],
|
||
|
|
[item["node_id"] for item in index.search("grip carve", limit=5)["results"]],
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
["safety.avalanche-window"],
|
||
|
|
[
|
||
|
|
item["node_id"]
|
||
|
|
for item in index.filter_nodes(family="safety", status="current")["results"]
|
||
|
|
],
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
{
|
||
|
|
"riding.edge-control",
|
||
|
|
"safety.avalanche-window",
|
||
|
|
"trail.glacier-bowl",
|
||
|
|
},
|
||
|
|
{
|
||
|
|
item["node_id"]
|
||
|
|
for item in index.dependencies("session.first-descent", depth=3)["results"]
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
first_context = compile_context(index, "ride-day", budget=300)
|
||
|
|
second_context = compile_context(index, "ride-day", budget=300)
|
||
|
|
self.assertEqual(first_context, second_context)
|
||
|
|
self.assertLessEqual(first_context["estimated_tokens"], 300)
|
||
|
|
self.assertEqual("session.first-descent", first_context["entries"][0]["node_id"])
|
||
|
|
self.assertTrue(first_context["omissions"])
|
||
|
|
|
||
|
|
changesets = ChangesetStore(project, "ski-editor")
|
||
|
|
created = changesets.create("edge-tuning")
|
||
|
|
updated = changesets.propose_update(
|
||
|
|
changeset_id="edge-tuning",
|
||
|
|
expected_changeset_hash=created["changeset_hash"],
|
||
|
|
node_id="riding.edge-control",
|
||
|
|
expected_content_hash=node_hash(project, "riding.edge-control"),
|
||
|
|
metadata={"summary": "Proposes a narrower grip transition for review."},
|
||
|
|
content="<script>unsafe()</script>\n\n**Previewed edge-tuning proposal.**",
|
||
|
|
relationship_changes=[],
|
||
|
|
rationale="Exercise an unrelated project's isolated proposal path.",
|
||
|
|
)
|
||
|
|
validated = changesets.validate("edge-tuning")
|
||
|
|
diff = changesets.diff("edge-tuning")
|
||
|
|
self.assertTrue(validated["valid"])
|
||
|
|
self.assertEqual(updated["changeset_hash"], validated["changeset_hash"])
|
||
|
|
self.assertEqual("update", diff["changes"][0]["operation"])
|
||
|
|
|
||
|
|
rendering = RenderService(project, changesets)
|
||
|
|
first_preview = rendering.preview("edge-tuning", "field-guide")
|
||
|
|
preview_path = root / ".docforge/previews/edge-tuning/field-guide.html"
|
||
|
|
preview_bytes = preview_path.read_bytes()
|
||
|
|
second_preview = rendering.preview("edge-tuning", "field-guide")
|
||
|
|
self.assertEqual(first_preview["preview_identity"], second_preview["preview_identity"])
|
||
|
|
self.assertEqual(preview_bytes, preview_path.read_bytes())
|
||
|
|
preview = preview_bytes.decode()
|
||
|
|
self.assertIn("<script>unsafe()</script>", preview)
|
||
|
|
self.assertIn("<strong>Previewed edge-tuning proposal.</strong>", preview)
|
||
|
|
self.assertFalse((root / ".docforge/rendered/field-guide.html").exists())
|
||
|
|
self.assertEqual(before, canonical_digest(root))
|
||
|
|
|
||
|
|
def test_generic_core_contains_no_project_specific_vocabulary(self) -> None:
|
||
|
|
banned = re.compile(r"\b(worldforge|assetforge|phase|villager)\b", re.IGNORECASE)
|
||
|
|
leaks = []
|
||
|
|
for path in sorted((ROOT / "src/docforge").rglob("*.py")):
|
||
|
|
for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||
|
|
if banned.search(line):
|
||
|
|
leaks.append(f"{path.name}:{line_number}:{line.strip()}")
|
||
|
|
self.assertEqual([], leaks)
|
||
|
|
|
||
|
|
|
||
|
|
class UnrelatedProjectIsolationTests(unittest.IsolatedAsyncioTestCase):
|
||
|
|
async def test_two_live_servers_remain_project_isolated(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
parent = Path(directory)
|
||
|
|
ski_root = copy_fixture("awesome-ski-game", parent / "ski")
|
||
|
|
alpha_root = copy_fixture("alpha", parent / "alpha")
|
||
|
|
ski_before = canonical_digest(ski_root)
|
||
|
|
alpha_before = canonical_digest(alpha_root)
|
||
|
|
ski_project = Project.open(ski_root)
|
||
|
|
alpha_project = Project.open(alpha_root)
|
||
|
|
ProjectIndex(ski_project).build()
|
||
|
|
ProjectIndex(alpha_project).build()
|
||
|
|
|
||
|
|
ski_parameters = StdioServerParameters(
|
||
|
|
command=sys.executable,
|
||
|
|
args=[
|
||
|
|
"-m",
|
||
|
|
"docforge.mcp_server",
|
||
|
|
"--project-root",
|
||
|
|
str(ski_root),
|
||
|
|
"--proposal-writer",
|
||
|
|
"ski-editor",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
alpha_parameters = StdioServerParameters(
|
||
|
|
command=sys.executable,
|
||
|
|
args=[
|
||
|
|
"-m",
|
||
|
|
"docforge.mcp_server",
|
||
|
|
"--project-root",
|
||
|
|
str(alpha_root),
|
||
|
|
"--proposal-writer",
|
||
|
|
"alpha-editor",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
|
||
|
|
async with AsyncExitStack() as stack:
|
||
|
|
ski_read, ski_write = await stack.enter_async_context(stdio_client(ski_parameters))
|
||
|
|
alpha_read, alpha_write = await stack.enter_async_context(
|
||
|
|
stdio_client(alpha_parameters)
|
||
|
|
)
|
||
|
|
ski_session = await stack.enter_async_context(ClientSession(ski_read, ski_write))
|
||
|
|
alpha_session = await stack.enter_async_context(
|
||
|
|
ClientSession(alpha_read, alpha_write)
|
||
|
|
)
|
||
|
|
await asyncio.gather(ski_session.initialize(), alpha_session.initialize())
|
||
|
|
|
||
|
|
ski_info, alpha_info = await asyncio.gather(
|
||
|
|
ski_session.call_tool("docforge_project_info", {}),
|
||
|
|
alpha_session.call_tool("docforge_project_info", {}),
|
||
|
|
)
|
||
|
|
ski_own, alpha_own, ski_cross, alpha_cross = await asyncio.gather(
|
||
|
|
ski_session.call_tool(
|
||
|
|
"docforge_get_node", {"node_id": "session.first-descent"}
|
||
|
|
),
|
||
|
|
alpha_session.call_tool("docforge_get_node", {"node_id": "guide.workflow"}),
|
||
|
|
ski_session.call_tool("docforge_get_node", {"node_id": "guide.workflow"}),
|
||
|
|
alpha_session.call_tool(
|
||
|
|
"docforge_get_node", {"node_id": "session.first-descent"}
|
||
|
|
),
|
||
|
|
)
|
||
|
|
ski_result, alpha_result = await asyncio.gather(
|
||
|
|
self._propose_and_preview(
|
||
|
|
ski_session,
|
||
|
|
node_id="riding.edge-control",
|
||
|
|
content_hash=node_hash(ski_project, "riding.edge-control"),
|
||
|
|
view_id="field-guide",
|
||
|
|
summary="A ski-only proposal from its bound server.",
|
||
|
|
),
|
||
|
|
self._propose_and_preview(
|
||
|
|
alpha_session,
|
||
|
|
node_id="guide.workflow",
|
||
|
|
content_hash=node_hash(alpha_project, "guide.workflow"),
|
||
|
|
view_id="manual",
|
||
|
|
summary="An Alpha-only proposal from its bound server.",
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual("awesome-ski-game", ski_info.structuredContent["project_id"])
|
||
|
|
self.assertEqual("alpha-docs", alpha_info.structuredContent["project_id"])
|
||
|
|
self.assertEqual("current", ski_info.structuredContent["index_health"]["state"])
|
||
|
|
self.assertEqual("current", alpha_info.structuredContent["index_health"]["state"])
|
||
|
|
self.assertNotEqual(
|
||
|
|
ski_info.structuredContent["project_root_fingerprint"],
|
||
|
|
alpha_info.structuredContent["project_root_fingerprint"],
|
||
|
|
)
|
||
|
|
self.assertNotEqual(
|
||
|
|
ski_info.structuredContent["source_hash"],
|
||
|
|
alpha_info.structuredContent["source_hash"],
|
||
|
|
)
|
||
|
|
self.assertEqual("session.first-descent", ski_own.structuredContent["node"]["node_id"])
|
||
|
|
self.assertEqual("guide.workflow", alpha_own.structuredContent["node"]["node_id"])
|
||
|
|
self.assertEqual("missing_node", ski_cross.structuredContent["error"]["code"])
|
||
|
|
self.assertEqual("awesome-ski-game", ski_cross.structuredContent["project_id"])
|
||
|
|
self.assertEqual("missing_node", alpha_cross.structuredContent["error"]["code"])
|
||
|
|
self.assertEqual("alpha-docs", alpha_cross.structuredContent["project_id"])
|
||
|
|
self.assertEqual("ski-editor", ski_result["creator"])
|
||
|
|
self.assertEqual("alpha-editor", alpha_result["creator"])
|
||
|
|
|
||
|
|
ski_changeset = ski_root / ".docforge/changesets/parallel-edit.json"
|
||
|
|
alpha_changeset = alpha_root / ".docforge/changesets/parallel-edit.json"
|
||
|
|
ski_preview = ski_root / ".docforge/previews/parallel-edit/field-guide.html"
|
||
|
|
alpha_preview = alpha_root / ".docforge/previews/parallel-edit/manual.html"
|
||
|
|
ski_index = ski_root / ".docforge/cache/index.sqlite3"
|
||
|
|
alpha_index = alpha_root / ".docforge/cache/index.sqlite3"
|
||
|
|
self.assertTrue(ski_index.is_file())
|
||
|
|
self.assertTrue(alpha_index.is_file())
|
||
|
|
self.assertTrue(ski_index.resolve().is_relative_to(ski_root))
|
||
|
|
self.assertTrue(alpha_index.resolve().is_relative_to(alpha_root))
|
||
|
|
self.assertTrue(ski_changeset.is_file())
|
||
|
|
self.assertTrue(alpha_changeset.is_file())
|
||
|
|
self.assertTrue(ski_preview.is_file())
|
||
|
|
self.assertTrue(alpha_preview.is_file())
|
||
|
|
ski_html = ski_preview.read_text(encoding="utf-8")
|
||
|
|
alpha_html = alpha_preview.read_text(encoding="utf-8")
|
||
|
|
self.assertIn('data-project="awesome-ski-game"', ski_html)
|
||
|
|
self.assertNotIn('data-project="alpha-docs"', ski_html)
|
||
|
|
self.assertIn('data-project="alpha-docs"', alpha_html)
|
||
|
|
self.assertNotIn('data-project="awesome-ski-game"', alpha_html)
|
||
|
|
self.assertFalse((ski_root / ".docforge/rendered/field-guide.html").exists())
|
||
|
|
self.assertFalse((alpha_root / ".docforge/rendered/manual.html").exists())
|
||
|
|
self.assertEqual(ski_before, canonical_digest(ski_root))
|
||
|
|
self.assertEqual(alpha_before, canonical_digest(alpha_root))
|
||
|
|
|
||
|
|
async def _propose_and_preview(
|
||
|
|
self,
|
||
|
|
session: ClientSession,
|
||
|
|
*,
|
||
|
|
node_id: str,
|
||
|
|
content_hash: str,
|
||
|
|
view_id: str,
|
||
|
|
summary: str,
|
||
|
|
) -> dict[str, object]:
|
||
|
|
created = await session.call_tool(
|
||
|
|
"docforge_create_changeset", {"changeset_id": "parallel-edit"}
|
||
|
|
)
|
||
|
|
updated = await session.call_tool(
|
||
|
|
"docforge_propose_node_update",
|
||
|
|
{
|
||
|
|
"changeset_id": "parallel-edit",
|
||
|
|
"expected_changeset_hash": created.structuredContent["changeset_hash"],
|
||
|
|
"node_id": node_id,
|
||
|
|
"expected_content_hash": content_hash,
|
||
|
|
"metadata": {"summary": summary},
|
||
|
|
"content": None,
|
||
|
|
"relationship_changes": [],
|
||
|
|
"rationale": "Prove simultaneous project-bound proposal isolation.",
|
||
|
|
},
|
||
|
|
)
|
||
|
|
validated = await session.call_tool(
|
||
|
|
"docforge_validate_changeset", {"changeset_id": "parallel-edit"}
|
||
|
|
)
|
||
|
|
preview = await session.call_tool(
|
||
|
|
"docforge_preview_changeset",
|
||
|
|
{"changeset_id": "parallel-edit", "view_id": view_id},
|
||
|
|
)
|
||
|
|
self.assertEqual("ok", updated.structuredContent["status"])
|
||
|
|
self.assertTrue(validated.structuredContent["valid"])
|
||
|
|
self.assertEqual("current", preview.structuredContent["state"])
|
||
|
|
return {
|
||
|
|
"creator": created.structuredContent["creator"],
|
||
|
|
"changeset_hash": updated.structuredContent["changeset_hash"],
|
||
|
|
"preview_identity": preview.structuredContent["preview_identity"],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|