GEWU/tests/api/test_mcp_read.py

import asyncio
import hashlib
import json
from pathlib import Path

import httpx2
import yaml
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
from mcp.types import TextContent

from gewu.adapters.git_repository import GitCanonicalRepository
from gewu.adapters.sqlite_identity import SQLiteIdentityStore
from gewu.application.reading import CanonicalNodeReader
from gewu.bootstrap.app import create_app
from tests.domain.proposal_fixtures import NODE, SOURCE_REVISION


def test_mcp_exposes_only_authenticated_canonical_read(tmp_path: Path) -> None:
    async def exercise() -> None:
        repository = GitCanonicalRepository(tmp_path / "knowledge.git")
        project_key = hashlib.sha256(NODE.project_id.encode()).hexdigest()
        node_key = hashlib.sha256(NODE.node_id.encode()).hexdigest()
        source_key = hashlib.sha256(SOURCE_REVISION.source_id.encode()).hexdigest()
        source_revision_key = hashlib.sha256(SOURCE_REVISION.revision_id.encode()).hexdigest()
        repository.publish_files(
            expected_head=repository.head(),
            files={
                f"projects/{project_key}/nodes/{node_key}.yaml": yaml.safe_dump(
                    NODE.model_dump(mode="json")
                ).encode(),
                f"sources/{source_key}/{source_revision_key}.yaml": yaml.safe_dump(
                    SOURCE_REVISION.model_dump(mode="json")
                ).encode(),
            },
            message="Publish test node",
            index_directory=tmp_path / "publishing",
        )
        identities = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
        token = identities.provision(
            credential_id="reader-key",
            principal_id="research-agent",
            roles=("agent",),
            project_ids=(NODE.project_id,),
        )
        app = create_app(
            identity_provider=identities,
            node_reader=CanonicalNodeReader(repository),
        )
        async with app.router.lifespan_context(app):
            async with httpx2.AsyncClient(
                transport=httpx2.ASGITransport(app=app),
                base_url="http://localhost",
                headers={"Authorization": f"Bearer {token}"},
            ) as http:
                async with Client(
                    streamable_http_client("http://localhost/mcp", http_client=http)
                ) as mcp:
                    tools = await mcp.list_tools()
                    assert {tool.name for tool in tools.tools} == {
                        "diagnostics",
                        "get_node",
                        "node_history",
                        "get_source",
                    }
                    diagnostic_result = await mcp.call_tool(
                        "diagnostics", {"projects": [NODE.project_id]}
                    )
                    assert not diagnostic_result.is_error
                    diagnostic_text = diagnostic_result.content[0]
                    assert isinstance(diagnostic_text, TextContent)
                    diagnostics = json.loads(diagnostic_text.text)
                    assert diagnostics["transport"] == "mcp"
                    assert diagnostics["selected_scope"]["projects"] == [NODE.project_id]
                    assert diagnostics["index_state"] == "not_configured"
                    assert "get_node" in diagnostics["authorized_operations"]
                    assert "proposal_submit" not in diagnostics["authorized_operations"]
                    assert token not in diagnostic_text.text
                    denied_diagnostics = await mcp.call_tool(
                        "diagnostics", {"projects": ["interactive-video"]}
                    )
                    assert denied_diagnostics.is_error
                    result = await mcp.call_tool(
                        "get_node",
                        {"project_id": NODE.project_id, "node_id": NODE.node_id},
                    )
                    assert not result.is_error
                    content = result.content[0]
                    assert content.type == "text"
                    envelope = json.loads(content.text)
                    assert envelope["canonical_revision"] == repository.head()
                    assert envelope["data"][0]["node_id"] == NODE.node_id
                    assert envelope["citations"][0]["quote"] == NODE.claims[0].evidence[0].quote
                    history = await mcp.call_tool(
                        "node_history",
                        {"project_id": NODE.project_id, "node_id": NODE.node_id},
                    )
                    assert not history.is_error
                    assert isinstance(history.content[0], TextContent)
                    assert len(json.loads(history.content[0].text)["data"]) == 1
                    source = await mcp.call_tool(
                        "get_source",
                        {
                            "project_id": NODE.project_id,
                            "revision_id": SOURCE_REVISION.revision_id,
                        },
                    )
                    assert not source.is_error
                    assert isinstance(source.content[0], TextContent)
                    assert json.loads(source.content[0].text)["data"][0]["revision_id"] == (
                        SOURCE_REVISION.revision_id
                    )
                    assert identities.revoke("reader-key")
                    revoked = await mcp.call_tool(
                        "get_node",
                        {"project_id": NODE.project_id, "node_id": NODE.node_id},
                    )
                    assert revoked.is_error
                    revoked_message = revoked.content[0]
                    assert isinstance(revoked_message, TextContent)
                    assert NODE.node_id not in revoked_message.text

            async with httpx2.AsyncClient(
                transport=httpx2.ASGITransport(app=app),
                base_url="http://localhost",
            ) as http:
                async with Client(
                    streamable_http_client("http://localhost/mcp", http_client=http)
                ) as mcp:
                    unauthenticated = await mcp.call_tool(
                        "get_node",
                        {"project_id": NODE.project_id, "node_id": NODE.node_id},
                    )
                    assert unauthenticated.is_error
                    error_message = unauthenticated.content[0]
                    assert isinstance(error_message, TextContent)
                    assert NODE.node_id not in error_message.text

    asyncio.run(exercise())