chemenu/tools/chemenu/tests/test_mcp_server.py

"""Tests for the MCP server (Gitea #19, plus `submit` from #32).

The acceptance criteria that are not about a value coming back:

- the wire format **is** the CLI's `--json` form, held together by a golden
  test rather than by intention;
- no path of the five original tools writes into `kb/`, `reports/` or git;
- five tools have no write path, because nothing under `chemenu.commands` is
  importable from the server - structural, not filtered;
- every response carries the commit it was computed from;
- telemetry cannot land inside the served checkout;
- the sixth tool, `submit`, exists only when `.wikitool-upload.json` opts a
  checkout in, writes only under `mcp-upload/`, and never resolves an
  identity from anywhere but the request's own headers.
"""
from __future__ import annotations

import asyncio
import base64
import json
import shutil
import subprocess
import sys
from pathlib import Path

import pytest

from chemenu import config, upload
from chemenu.errors import ValidationError

pytest.importorskip("mcp", reason="the MCP server's dependency is optional; see "
                                  "tools/requirements-mcp.txt")

from mcp.server.context import ServerRequestContext  # noqa: E402
from mcp.server.mcpserver import Context as ToolContext  # noqa: E402

from chemenu.mcp.server import (  # noqa: E402  - after the skip guard
    TRANSPORTS,
    TraceWouldWriteIntoCorpus,
    build_server,
    check_trace_destination,
    serve,
)

_UPLOAD_CFG = {
    "schema": 1,
    "identity_header": "X-Forwarded-User",
    "max_bytes": 1024,
    "allowed_extensions": [".md"],
    "quota": {"submissions_per_day": 10, "bytes_per_day": 100_000},
}


class _FakeRequest:
    def __init__(self, headers):
        self.headers = headers


def _context(server, headers: dict | None = None) -> ToolContext:
    """A `Context` carrying `headers` the way a real HTTP transport would -
    `None` simulates stdio, where the SDK's own docstring says `ctx.headers`
    is `None`."""
    request = _FakeRequest(headers) if headers is not None else None
    request_context = ServerRequestContext(
        session=None, lifespan_context={}, protocol_version="2025-06-18",
        method="tools/call", request=request,
    )
    return ToolContext(mcp_server=server, request_context=request_context)


@pytest.fixture
def corpus(tmp_path: Path) -> Path:
    """A committed instance that is not this checkout.

    Carries its own `types/`, because a served corpus is a whole instance: the
    page kinds and `describe_type` are answered from the instance's own schema,
    not from whichever checkout the package happens to sit in. Copied rather
    than linked - a link would let a test that writes to `<root>/types/...`
    write into this repository, which is not hypothetical.
    """
    root = tmp_path / "served"
    entities = root / "kb" / "entities"
    entities.mkdir(parents=True)
    shutil.copytree(config._PACKAGE_ROOT / "types", root / "types")
    (root / "kb" / "entities" / "COLLECTION.md").write_text("# entities\n", encoding="utf-8")
    (entities / "Kingfisher.md").write_text(
        "---\ntype: types/entity.md\nentity_type: system\n"
        "summary: The fixture's own system.\n---\n\n"
        "# Kingfisher\n\nKingfisher is the system this fixture is about.\n",
        encoding="utf-8",
    )
    subprocess.run(["git", "init", "-b", "main"], cwd=root, check=True, capture_output=True)
    for key, value in (("user.name", "Fixture"), ("user.email", "f@example.com")):
        subprocess.run(["git", "config", key, value], cwd=root, check=True, capture_output=True)
    subprocess.run(["git", "add", "-A"], cwd=root, check=True, capture_output=True)
    subprocess.run(["git", "commit", "-m", "corpus"], cwd=root, check=True, capture_output=True)
    return root


def _call(server, name: str, arguments: dict | None = None) -> dict:
    result = asyncio.run(server.call_tool(name, arguments or {}))
    if isinstance(result, tuple):
        result = result[1]
    assert not result.is_error, result.content
    return result.structured_content


def _tree(root: Path) -> dict[str, tuple[int, bytes]]:
    """Every file under `root` with its size and contents, for a before/after
    comparison that catches a rewrite with the same length."""
    return {
        str(path.relative_to(root)): (path.stat().st_size, path.read_bytes())
        for path in sorted(root.rglob("*"))
        if path.is_file()
    }


def test_the_four_tools_are_there_and_nothing_that_writes(corpus):
    server = build_server(corpus, check_trace=False)
    names = {tool.name for tool in asyncio.run(server.list_tools())}
    assert names == {"search", "types", "describe_type", "lint", "status"}
    # Named rather than pattern-matched: the point is that these are absent
    # because the functions are unreachable, and a test that only looked for
    # "no tool called publish" would pass on a filtered list too.
    assert names.isdisjoint({"new", "touch", "xref", "cite", "publish", "migrate", "rm"})


def test_all_three_adapters_offer_the_same_default_limit(corpus):
    """The CLI's `--limit`, `api.search(limit=...)` and this tool are three
    knobs on one search, and three literals is how they start disagreeing about
    what "the default search" returns. They read one constant."""
    import inspect

    from chemenu.api import Corpus
    from chemenu.commands.search import search_command
    from chemenu.search.types import DEFAULT_LIMIT

    server = build_server(corpus, check_trace=False)
    tool = next(t for t in asyncio.run(server.list_tools()) if t.name == "search")

    assert tool.input_schema["properties"]["limit"]["default"] == DEFAULT_LIMIT
    assert inspect.signature(Corpus.search).parameters["limit"].default == DEFAULT_LIMIT
    assert inspect.signature(search_command).parameters["limit"].default.default == DEFAULT_LIMIT


def test_no_tool_writes_anything_into_the_corpus_or_git(corpus):
    server = build_server(corpus, check_trace=False)
    before = _tree(corpus)
    head_before = subprocess.run(
        ["git", "rev-parse", "HEAD"], cwd=corpus, capture_output=True, text=True, check=True
    ).stdout

    _call(server, "search", {"query": "Kingfisher"})
    _call(server, "search", {"predicates": ["entity_type=system"]})
    _call(server, "types")
    _call(server, "describe_type", {"name": "entity"})
    _call(server, "lint")
    _call(server, "status")

    assert _tree(corpus) == before
    head_after = subprocess.run(
        ["git", "rev-parse", "HEAD"], cwd=corpus, capture_output=True, text=True, check=True
    ).stdout
    assert head_after == head_before
    # `lint` without `--json` writes a report into `reports/`; the server must
    # only ever reach the JSON form.
    assert not (corpus / "reports").exists()
    assert subprocess.run(
        ["git", "status", "--porcelain"], cwd=corpus, capture_output=True, text=True, check=True
    ).stdout == ""


def test_the_wire_format_is_the_clis_json_form(corpus):
    """The golden test. One contract, with the CLI as its executable
    specification - so the two cannot drift while both look correct.

    The CLI is run as a subprocess against the same tree, pointed at it with
    `CHEMENU_ROOT` - which is also an end-to-end check that the root resolution
    works from outside the process.
    """
    server = build_server(corpus, check_trace=False)
    served = _call(server, "search", {"query": "Kingfisher", "limit": 5})

    import os

    environment = {**os.environ, config.ENV_ROOT: str(corpus), "WIKI_TRACE": "0"}
    completed = subprocess.run(
        [str(config._PACKAGE_ROOT / "tools" / "wikitool"),
         "search", "Kingfisher", "--limit", "5", "--json"],
        capture_output=True, text=True, check=True, env=environment,
    )
    from_cli = json.loads(completed.stdout)

    # `generated` is the CLI's date stamp and `commit`/`as_of` are the server's
    # revision stamp - two answers to "when", neither of them a finding. What
    # has to match is everything that describes the *corpus*.
    shared = ("query", "predicates", "backend", "count", "total", "truncated", "limit",
              "results", "unreadable")
    assert {key: served[key] for key in shared} == {key: from_cli[key] for key in shared}


def test_types_and_describe_match_the_cli_too(corpus):
    import os

    server = build_server(corpus, check_trace=False)
    environment = {**os.environ, config.ENV_ROOT: str(corpus), "WIKI_TRACE": "0"}
    wikitool = str(config._PACKAGE_ROOT / "tools" / "wikitool")

    from_cli = json.loads(subprocess.run(
        [wikitool, "types", "list", "--json"],
        capture_output=True, text=True, check=True, env=environment,
    ).stdout)
    assert _call(server, "types")["types"] == from_cli

    from_cli = json.loads(subprocess.run(
        [wikitool, "types", "describe", "entity", "--json"],
        capture_output=True, text=True, check=True, env=environment,
    ).stdout)
    served = _call(server, "describe_type", {"name": "entity"})
    assert {k: v for k, v in served.items() if k not in ("commit", "as_of")} == from_cli


def test_every_response_carries_the_commit_it_was_computed_from(corpus):
    server = build_server(corpus, check_trace=False)
    head = subprocess.run(
        ["git", "rev-parse", "HEAD"], cwd=corpus, capture_output=True, text=True, check=True
    ).stdout.strip()
    for name in ("types", "lint", "status"):
        answer = _call(server, name)
        assert answer["commit"] == head
        assert answer["as_of"]


def test_a_dirty_tree_is_stamped_with_no_commit_rather_than_a_wrong_one(corpus):
    """The answer no longer corresponds to any revision, and says so instead of
    naming the commit it is no longer about."""
    (corpus / "kb" / "entities" / "Latecomer.md").write_text(
        "---\ntype: types/entity.md\n---\n\n# Latecomer\n", encoding="utf-8"
    )
    assert _call(build_server(corpus, check_trace=False), "status")["commit"] is None


def test_a_bad_predicate_comes_back_as_a_tool_error_not_a_traceback(corpus):
    """The SDK draws the line this depends on: a `ToolError` is a deliberate
    refusal whose text reaches the caller, while any other exception is a crash
    whose text stays on the server. A bad predicate is the caller's argument, so
    the message that says what to write instead has to travel."""
    from mcp.server.mcpserver.exceptions import ToolError, UnexpectedToolError

    server = build_server(corpus, check_trace=False)
    with pytest.raises(ToolError) as excinfo:
        asyncio.run(server.call_tool("search", {"predicates": ["not a predicate"]}))
    assert not isinstance(excinfo.value, UnexpectedToolError)
    assert "field=value" in str(excinfo.value)


def test_the_server_module_cannot_reach_a_write_command():
    """Structural: import the server in a clean interpreter and nothing under
    `chemenu.commands` is loaded, so there is no `publish` to expose."""
    result = subprocess.run(
        [sys.executable, "-c",
         "import sys, chemenu.mcp.server;"
         "print([m for m in sys.modules if m.startswith('chemenu.commands')])"],
        cwd=config._PACKAGE_ROOT / "tools", capture_output=True, text=True, check=True,
    )
    assert result.stdout.strip() == "[]"


def test_tracing_into_the_served_checkout_is_refused(corpus, monkeypatch):
    """The sync that keeps this checkout current is `git reset --hard`, which is
    entitled to wipe `reports/`. A per-request trace written there is both lost
    work and a silent way to dirty the tree the cache keys on."""
    monkeypatch.setenv("WIKI_TRACE", "1")
    monkeypatch.delenv("WIKI_TRACE_DIR", raising=False)
    with pytest.raises(TraceWouldWriteIntoCorpus):
        check_trace_destination(corpus)

    monkeypatch.setenv("WIKI_TRACE_DIR", str(corpus / "reports" / "telemetry"))
    with pytest.raises(TraceWouldWriteIntoCorpus):
        check_trace_destination(corpus)


def test_tracing_outside_the_corpus_or_switched_off_is_accepted(corpus, tmp_path, monkeypatch):
    monkeypatch.setenv("WIKI_TRACE", "0")
    check_trace_destination(corpus)

    monkeypatch.setenv("WIKI_TRACE", "1")
    monkeypatch.setenv("WIKI_TRACE_DIR", str(tmp_path / "traces"))
    check_trace_destination(corpus)


def test_a_dev_checkout_like_corpus_refuses_to_start_with_no_override_at_all(corpus, monkeypatch):
    """The `corpus` fixture carries no `.wikitool-release.json`, so the guard
    checks the same policy `wikitool doctor` and the writer would: installation-
    form default on, and the server must refuse even with nothing set."""
    monkeypatch.delenv("WIKI_TRACE", raising=False)
    monkeypatch.delenv("WIKI_TRACE_DIR", raising=False)
    with pytest.raises(TraceWouldWriteIntoCorpus):
        check_trace_destination(corpus)


def test_a_stamped_distribution_starts_with_no_override_at_all(corpus, monkeypatch):
    """A distributed instance defaults telemetry off - the server must not need
    `WIKI_TRACE=0` set for it, unlike a dev checkout."""
    (corpus / ".wikitool-release.json").write_text("{}", encoding="utf-8")
    monkeypatch.delenv("WIKI_TRACE", raising=False)
    monkeypatch.delenv("WIKI_TRACE_DIR", raising=False)
    check_trace_destination(corpus)  # must not raise


def test_only_the_two_chosen_transports_are_offered():
    """`sse` is reachable through the SDK and deliberately not offered: it is
    the superseded remote transport, and building on it now only moves the
    migration later."""
    assert TRANSPORTS == ("stdio", "streamable-http")
    with pytest.raises(ValueError, match="unknown transport"):
        serve(transport="sse")


def test_bind_arguments_reach_only_the_transport_that_takes_them(monkeypatch):
    """`run_stdio_async` accepts no host or port; handing it one is a TypeError,
    not a harmless no-op. And the loopback default has to be overridable, or a
    server in a container behind a reverse proxy binds an interface the proxy
    cannot reach."""
    from chemenu.mcp import __main__ as entry

    seen: dict = {}
    monkeypatch.setattr(entry, "serve", lambda **kwargs: seen.update(kwargs))

    entry.main([])
    assert seen == {"transport": "stdio", "root": None}

    seen.clear()
    entry.main(["--transport", "streamable-http", "--host", "0.0.0.0", "--port", "9001"])
    assert seen["transport"] == "streamable-http"
    assert seen["host"] == "0.0.0.0" and seen["port"] == 9001


def test_a_refused_trace_destination_stops_the_process_with_a_message(monkeypatch, capsys):
    from chemenu.mcp import __main__ as entry

    monkeypatch.setenv("WIKI_TRACE", "1")
    monkeypatch.delenv("WIKI_TRACE_DIR", raising=False)
    assert entry.main([]) == 1
    assert "WIKI_TRACE" in capsys.readouterr().err


# --- submit (Gitea #32) --------------------------------------------------------

def _arm_upload(corpus: Path) -> None:
    (corpus / ".wikitool-upload.json").write_text(json.dumps(_UPLOAD_CFG), encoding="utf-8")
    (corpus / "incoming").mkdir(exist_ok=True)


def test_submit_is_absent_without_the_upload_config(corpus):
    server = build_server(corpus, check_trace=False)
    names = {tool.name for tool in asyncio.run(server.list_tools())}
    assert "submit" not in names


def test_submit_is_present_once_armed(corpus):
    _arm_upload(corpus)
    server = build_server(corpus, check_trace=False)
    names = {tool.name for tool in asyncio.run(server.list_tools())}
    assert names == {"search", "types", "describe_type", "lint", "status", "submit"}


def test_a_malformed_upload_config_refuses_to_build(corpus):
    (corpus / ".wikitool-upload.json").write_text("{not json", encoding="utf-8")
    with pytest.raises(ValidationError):
        build_server(corpus, check_trace=False)


def test_submit_writes_only_into_mcp_upload_incoming_stays_untouched(corpus):
    _arm_upload(corpus)
    server = build_server(corpus, check_trace=False)
    before = _tree(corpus)
    content_b64 = base64.b64encode(b"a submitted file\n").decode("ascii")

    ctx = _context(server, {"X-Forwarded-User": "alice"})
    result = asyncio.run(server.call_tool(
        "submit", {"filename": "note.md", "content_base64": content_b64}, context=ctx,
    ))
    if isinstance(result, tuple):
        result = result[1]
    assert not result.is_error, result.content
    manifest = result.structured_content

    assert manifest["submitter"] == "alice"
    assert manifest["submitter_source"] == "X-Forwarded-User"

    after = _tree(corpus)
    changed = {p for p in after if after.get(p) != before.get(p)}
    assert changed
    assert all(p.startswith("mcp-upload/") for p in changed)
    assert list((corpus / "incoming").iterdir()) == []

    # git itself is untouched - the fixture's own committed tree (kb/, types/)
    # is unchanged, and HEAD did not move. `mcp-upload/` and the config file
    # are new, untracked paths (this fixture carries no .gitignore of its
    # own, unlike the real repo), which `git status` reports and is exactly
    # what a real checkout's `.gitignore` would hide - not a write to git.
    status = subprocess.run(
        ["git", "status", "--porcelain"], cwd=corpus, capture_output=True, text=True, check=True
    ).stdout
    changed_tracked = {line[3:] for line in status.splitlines() if not line.startswith("??")}
    assert changed_tracked == set()


def test_submit_without_identity_header_is_a_tool_error(corpus):
    from mcp.server.mcpserver.exceptions import ToolError

    _arm_upload(corpus)
    server = build_server(corpus, check_trace=False)
    content_b64 = base64.b64encode(b"data\n").decode("ascii")

    ctx = _context(server, {})  # headers present, but not the configured one
    with pytest.raises(ToolError):
        asyncio.run(server.call_tool(
            "submit", {"filename": "note.md", "content_base64": content_b64}, context=ctx,
        ))


def test_submit_on_stdio_with_no_headers_at_all_is_a_tool_error(corpus):
    """`ctx.headers` is `None` on stdio, per the SDK's own docstring - the
    tool must treat that the same as an absent identity header, not crash on
    a `None.get(...)`."""
    from mcp.server.mcpserver.exceptions import ToolError

    _arm_upload(corpus)
    server = build_server(corpus, check_trace=False)
    content_b64 = base64.b64encode(b"data\n").decode("ascii")

    ctx = _context(server, headers=None)
    with pytest.raises(ToolError):
        asyncio.run(server.call_tool(
            "submit", {"filename": "note.md", "content_base64": content_b64}, context=ctx,
        ))