DocForge2/tests/test_adapter_contract.py

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()