226 lines
8.7 KiB
Python
226 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
from mcp.shared.memory import create_connected_server_and_client_session
|
|
|
|
from docforge.adapter_contract import (
|
|
AdapterEdge,
|
|
AdapterNode,
|
|
AdapterProject,
|
|
AdapterProjection,
|
|
ShadowArtifact,
|
|
compare_artifacts,
|
|
validate_projection,
|
|
)
|
|
from docforge.errors import DocForgeError
|
|
from docforge.index import ProjectIndex
|
|
from docforge.mcp_server import READ_TOOLS, create_read_only_server
|
|
from docforge.models import Edge, Node
|
|
|
|
|
|
class Loader:
|
|
def __init__(self, projection: AdapterProjection) -> None:
|
|
self.projection = projection
|
|
|
|
def load_projection(self) -> AdapterProjection:
|
|
return self.projection
|
|
|
|
|
|
class AdapterContractTests(unittest.TestCase):
|
|
def projection(self, root: Path) -> AdapterProjection:
|
|
foundation = Node(
|
|
node_id="guide.foundation",
|
|
title="Foundation",
|
|
family="guide",
|
|
authority="authoritative",
|
|
status="active",
|
|
tags=("guide",),
|
|
summary="The base contract.",
|
|
content="Foundation content.",
|
|
source_path="docs/foundation.md",
|
|
source_anchor=None,
|
|
content_hash="1" * 64,
|
|
)
|
|
workflow = Node(
|
|
node_id="guide.workflow",
|
|
title="Workflow",
|
|
family="guide",
|
|
authority="approved_plan",
|
|
status="planned",
|
|
tags=("guide", "workflow"),
|
|
summary="The editing workflow.",
|
|
content="Workflow content.",
|
|
source_path="docs/workflow.md",
|
|
source_anchor=None,
|
|
content_hash="2" * 64,
|
|
)
|
|
return AdapterProjection(
|
|
project_id="adapter-fixture",
|
|
title="Adapter fixture",
|
|
adapter_id="fixture-shadow",
|
|
adapter_version="1",
|
|
root=root,
|
|
revision="fixture-revision",
|
|
source_hash="3" * 64,
|
|
nodes=(
|
|
AdapterNode(foundation, (("acceptance", "proven"),)),
|
|
AdapterNode(workflow, (("acceptance", "pending"),)),
|
|
),
|
|
edges=(
|
|
AdapterEdge(
|
|
Edge("guide.workflow", "depends_on", "guide.foundation"),
|
|
(("source", "docs/workflow.md"),),
|
|
),
|
|
),
|
|
)
|
|
|
|
def test_adapter_projection_builds_and_checks_through_standard_index(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
projection = self.projection(root)
|
|
project = AdapterProject(Loader(projection), cache_root=root / ".cache" / "shadow")
|
|
index = ProjectIndex(project)
|
|
|
|
built = index.build()
|
|
checked = index.check()
|
|
|
|
self.assertEqual("fixture-shadow@1", built["adapter"])
|
|
self.assertEqual(2, checked["node_count"])
|
|
self.assertEqual("Workflow", index.get_node("guide.workflow")["node"]["title"])
|
|
self.assertEqual(projection.identity(), projection.identity())
|
|
|
|
def test_projection_rejects_unsorted_metadata_graph_and_identity_changes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
projection = self.projection(root)
|
|
invalid_metadata = replace(
|
|
projection,
|
|
nodes=(
|
|
replace(
|
|
projection.nodes[0],
|
|
metadata=(("z", "last"), ("a", "first")),
|
|
),
|
|
projection.nodes[1],
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "metadata keys"):
|
|
validate_projection(invalid_metadata)
|
|
|
|
invalid_source = replace(
|
|
projection,
|
|
nodes=(
|
|
replace(
|
|
projection.nodes[0],
|
|
node=replace(projection.nodes[0].node, source_path="../outside.md"),
|
|
),
|
|
projection.nodes[1],
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "source path"):
|
|
validate_projection(invalid_source)
|
|
|
|
broken = replace(
|
|
projection,
|
|
edges=(AdapterEdge(Edge("guide.workflow", "depends_on", "missing.node")),),
|
|
)
|
|
with self.assertRaisesRegex(DocForgeError, "missing nodes"):
|
|
validate_projection(broken)
|
|
|
|
loader = Loader(projection)
|
|
project = AdapterProject(loader, cache_root=root / ".cache" / "shadow")
|
|
loader.projection = replace(projection, adapter_version="2")
|
|
with self.assertRaisesRegex(DocForgeError, "identity changed"):
|
|
project.load()
|
|
|
|
def test_adapter_cache_must_remain_inside_project(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
outside = root.parent / "outside-adapter-cache"
|
|
with self.assertRaisesRegex(DocForgeError, "confined"):
|
|
AdapterProject(Loader(self.projection(root)), cache_root=outside)
|
|
|
|
def test_artifact_comparison_is_complete_and_byte_exact(self) -> None:
|
|
reference = (
|
|
ShadowArtifact("manual", b"same"),
|
|
ShadowArtifact("timeline", b"old"),
|
|
)
|
|
exact = compare_artifacts(reference, reference)
|
|
self.assertEqual("ok", exact["status"])
|
|
self.assertEqual(2, exact["count"])
|
|
|
|
mismatch = compare_artifacts(
|
|
reference,
|
|
(
|
|
ShadowArtifact("timeline", b"new"),
|
|
ShadowArtifact("extra", b"extra"),
|
|
),
|
|
)
|
|
self.assertEqual("mismatch", mismatch["status"])
|
|
self.assertEqual(["manual"], mismatch["missing"])
|
|
self.assertEqual(["extra"], mismatch["unexpected"])
|
|
self.assertEqual(["timeline"], mismatch["changed"])
|
|
|
|
|
|
class AdapterReadOnlyMcpTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_adapter_project_exposes_only_read_tools_and_custom_context(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
fixture = AdapterContractTests()
|
|
project = AdapterProject(
|
|
Loader(fixture.projection(root)),
|
|
cache_root=root / ".cache" / "adapter-read-only",
|
|
)
|
|
index = ProjectIndex(project)
|
|
index.build()
|
|
calls: list[tuple[str, int | None]] = []
|
|
|
|
def context_provider(
|
|
current: ProjectIndex, profile: str, budget: int | None
|
|
) -> dict[str, object]:
|
|
checked = current.check()
|
|
calls.append((profile, budget))
|
|
return {
|
|
"status": "ok",
|
|
"project_id": checked["project_id"],
|
|
"project_root_fingerprint": checked["project_root_fingerprint"],
|
|
"revision": checked["revision"],
|
|
"source_hash": checked["source_hash"],
|
|
"adapter": checked["adapter"],
|
|
"profile": profile,
|
|
"budget": budget,
|
|
"estimated_tokens": 1,
|
|
"entries": [],
|
|
"omissions": [],
|
|
}
|
|
|
|
server = create_read_only_server(project, context_provider=context_provider)
|
|
async with create_connected_server_and_client_session(
|
|
server, raise_exceptions=True
|
|
) as session:
|
|
tools = await session.list_tools()
|
|
info = await session.call_tool("docforge_project_info", {})
|
|
contract = await session.call_tool("docforge_get_contract", {})
|
|
context = await session.call_tool(
|
|
"docforge_get_context", {"profile": "fixture", "budget": 321}
|
|
)
|
|
|
|
self.assertEqual(READ_TOOLS, tuple(tool.name for tool in tools.tools))
|
|
self.assertEqual("adapter-fixture", info.structuredContent["project_id"])
|
|
self.assertEqual(list(READ_TOOLS), contract.structuredContent["allowed_tools"])
|
|
self.assertIn(
|
|
"isolated_changeset_writes",
|
|
contract.structuredContent["excluded_operations"],
|
|
)
|
|
self.assertFalse(contract.structuredContent["proposal_access"]["enabled"])
|
|
self.assertFalse(contract.structuredContent["isolated_changeset_writes_allowed"])
|
|
self.assertEqual("fixture", context.structuredContent["profile"])
|
|
self.assertEqual([("fixture", 321)], calls)
|
|
self.assertFalse(project.descriptor.changeset_root.exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|