344 lines
14 KiB
Python
344 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from collections.abc import Mapping
|
|
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,
|
|
AdapterProjectSettings,
|
|
ShadowArtifact,
|
|
compare_artifacts,
|
|
validate_projection,
|
|
)
|
|
from docforge.errors import DocForgeError
|
|
from docforge.index import ProjectIndex
|
|
from docforge.mcp_server import (
|
|
ALL_TOOLS,
|
|
READ_TOOLS,
|
|
create_project_server,
|
|
create_read_only_server,
|
|
)
|
|
from docforge.models import (
|
|
Edge,
|
|
Node,
|
|
ProjectSnapshot,
|
|
ProposalWriter,
|
|
RenderConfig,
|
|
RenderView,
|
|
)
|
|
|
|
|
|
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())
|
|
|
|
async def test_adapter_project_proposals_require_explicit_policy_and_stay_isolated(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory).resolve()
|
|
(root / ".docforge").mkdir()
|
|
descriptor = root / ".docforge" / "project.toml"
|
|
descriptor.write_text("adapter fixture\n", encoding="utf-8")
|
|
content_root = root / "docs"
|
|
content_root.mkdir()
|
|
sources = (content_root / "foundation.md", content_root / "workflow.md")
|
|
for source in sources:
|
|
source.write_text(f"canonical {source.stem}\n", encoding="utf-8")
|
|
template_root = root / "templates"
|
|
template_root.mkdir()
|
|
template = template_root / "review.html"
|
|
template.write_text(
|
|
"<html><head><title>{{ docforge_title }}</title></head>"
|
|
"<body>{{ docforge_content }}</body></html>\n",
|
|
encoding="utf-8",
|
|
)
|
|
validation_calls: list[tuple[Mapping[str, object], ...]] = []
|
|
|
|
def validate_proposal(
|
|
base: ProjectSnapshot,
|
|
projected: ProjectSnapshot,
|
|
operations: tuple[Mapping[str, object], ...],
|
|
) -> None:
|
|
self.assertEqual(base.edges, projected.edges)
|
|
validation_calls.append(operations)
|
|
|
|
settings = AdapterProjectSettings(
|
|
descriptor_path=descriptor,
|
|
content_roots=(content_root,),
|
|
canonical_sources=sources,
|
|
changeset_root=root / ".cache" / "changesets",
|
|
proposal_writers=(ProposalWriter("fixture-writer", ("guide",), ("update",)),),
|
|
render=RenderConfig(
|
|
template_root=template_root,
|
|
preview_root=root / ".cache" / "previews",
|
|
views=(
|
|
RenderView(
|
|
"review",
|
|
"generic_html",
|
|
template,
|
|
root / ".cache" / "rendered" / "review.html",
|
|
"Adapter review",
|
|
("guide",),
|
|
),
|
|
),
|
|
),
|
|
proposal_validator=validate_proposal,
|
|
)
|
|
project = AdapterProject(
|
|
Loader(AdapterContractTests().projection(root)),
|
|
cache_root=root / ".cache" / "adapter",
|
|
settings=settings,
|
|
)
|
|
ProjectIndex(project).build()
|
|
before = {source: source.read_bytes() for source in sources}
|
|
server = create_project_server(project, proposal_writer="fixture-writer")
|
|
async with create_connected_server_and_client_session(
|
|
server, raise_exceptions=True
|
|
) as session:
|
|
tools = await session.list_tools()
|
|
created = await session.call_tool(
|
|
"docforge_create_changeset", {"changeset_id": "adapter-update"}
|
|
)
|
|
node = await session.call_tool("docforge_get_node", {"node_id": "guide.workflow"})
|
|
proposed = await session.call_tool(
|
|
"docforge_propose_node_update",
|
|
{
|
|
"changeset_id": "adapter-update",
|
|
"expected_changeset_hash": created.structuredContent["changeset_hash"],
|
|
"node_id": "guide.workflow",
|
|
"expected_content_hash": node.structuredContent["node"]["content_hash"],
|
|
"metadata": None,
|
|
"content": "Updated workflow content.",
|
|
"relationship_changes": [],
|
|
"rationale": "Prove adapter proposal policy.",
|
|
},
|
|
)
|
|
validated = await session.call_tool(
|
|
"docforge_validate_changeset",
|
|
{"changeset_id": "adapter-update"},
|
|
)
|
|
diff = await session.call_tool(
|
|
"docforge_get_changeset_diff",
|
|
{"changeset_id": "adapter-update"},
|
|
)
|
|
preview = await session.call_tool(
|
|
"docforge_preview_changeset",
|
|
{"changeset_id": "adapter-update", "view_id": "review"},
|
|
)
|
|
|
|
self.assertEqual(ALL_TOOLS, tuple(tool.name for tool in tools.tools))
|
|
self.assertEqual("ok", proposed.structuredContent["status"])
|
|
self.assertTrue(validated.structuredContent["valid"])
|
|
self.assertEqual("update", diff.structuredContent["changes"][0]["operation"])
|
|
preview_path = root / preview.structuredContent["preview"]["path"]
|
|
self.assertTrue(preview_path.is_file())
|
|
self.assertTrue(validation_calls)
|
|
self.assertEqual(before, {source: source.read_bytes() for source in sources})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|