portfolio-optimiser/tests/test_ingest_golden_mcp.py

"""Golden regression for the MCP stdio transport — run against a REAL MCP server subprocess.

Mirrors ``tests/test_ingest_golden_http.py`` / ``_sql.py``, and closes the one gap those two could
not: every other MCP test injects a canned ``McpCallTool``, so ``stdio_call_tool`` — the real stdio
path — was written but never executed. Docs said so explicitly. This module executes it.

**Why a real subprocess and not an in-process server.** Driving ``datasource.build_mcp_server``
in-process would exercise ``mcp_get`` (already measured, six detach mutations) and not one line of
``stdio_call_tool``: the env-var resolution, ``StdioServerParameters``, the ``stdio_client``
transport, session initialisation, the ``isError`` branch and the timeout wrapper all live on the
subprocess path. Only spawning a server makes "run against a real MCP server" a true claim. The cost
discipline this repo enforces is about MODEL calls and Azure — a local Python subprocess costs
neither, and the canned-tool contract tests stay subprocess-free.

**Why the golden server takes no arguments.** The transport carries its two coordinates in the URL
(``mcp+stdio://<server_ref>/<tool>``) and calls the tool with an empty argument dict.
``datasource.build_mcp_server`` therefore CANNOT serve this path — its ``retrieve_cost_docs(query)``
has a required parameter, so the call fails with an error result. That is a real property of the two
seams, not an accident of the fixture: ``build_mcp_server`` serves the agents' retrieval path, this
serves ingest.
"""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

from portfolio_optimiser.ingest import materialize
from portfolio_optimiser.ingest_mcp import IngestError, mcp_get, stdio_call_tool

GOLDEN = Path(__file__).resolve().parents[1] / "examples" / "ingest-golden-mcp"
_SERVER_REF = "PORTFOLIO_GOLDEN_MCP"


def _bundle_bytes(directory: Path) -> dict[str, bytes]:
    return {p.name: p.read_bytes() for p in sorted(directory.iterdir()) if p.is_file()}


def _transport(server_script: Path, monkeypatch: pytest.MonkeyPatch, **kwargs: object):
    """Point ``server_ref`` at this interpreter and pass the server script as its argument.

    The manifest names an ENV VAR, never an executable path (spec §4), which is exactly why the
    golden stays checkout-location independent: the path travels in ``args``, not in the bundle.
    """
    monkeypatch.setenv(_SERVER_REF, sys.executable)
    return mcp_get(stdio_call_tool(args=(str(server_script),), **kwargs))  # type: ignore[arg-type]


def test_golden_mcp_extraction_is_bit_deterministic(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The §11 golden case for ``mcp+stdio``, byte-for-byte, against a live server subprocess.

    RED when any byte of the expected bundle diverges — and, unlike every other MCP test, RED if the
    real stdio transport stops working at all.
    """
    ingested_at = (GOLDEN / "ingested-at.txt").read_text(encoding="utf-8").strip()
    get = _transport(GOLDEN / "server.py", monkeypatch)
    out = tmp_path / "bundle"

    materialize(
        GOLDEN / "manifest.json",
        out,
        ingested_at=ingested_at,
        allow_network=True,
        http_get=get,
    )

    expected = _bundle_bytes(GOLDEN / "expected-bundle")
    actual = _bundle_bytes(out)
    # File-SET equality first — catches extra AND missing files, not just diverging bytes.
    assert actual.keys() == expected.keys()
    for name, content in expected.items():
        assert actual[name] == content, f"golden byte divergence in {name}"

    # §10: a second run over the same output leaves every byte unchanged.
    materialize(
        GOLDEN / "manifest.json",
        out,
        ingested_at=ingested_at,
        allow_network=True,
        http_get=get,
    )
    assert _bundle_bytes(out) == expected


def _write_server(tmp_path: Path, body: str) -> Path:
    """A throwaway MCP stdio server for the failure paths — never the committed golden fixture."""
    script = tmp_path / "failing_server.py"
    script.write_text(
        "from mcp.server.fastmcp import FastMCP\n"
        "server = FastMCP('probe')\n" + body + "\nif __name__ == '__main__':\n    server.run()\n",
        encoding="utf-8",
    )
    return script


def test_tool_error_reaches_the_caller_as_ingest_error(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A failing tool must surface as ``IngestError(code='mcp_tool_error')`` AT THE CALL SITE.

    This is the property only a real server can prove. ``stdio_call_tool`` raises inside two nested
    anyio task groups (``stdio_client`` and ``ClientSession``), and anyio re-packages anything
    leaving a task group as a ``BaseExceptionGroup``. So the raise the module carefully wrote never
    reached callers as an ``IngestError`` — and the entire ingest path is built on catching that
    type, by ``code``. RED before the unwrap; the canned-tool tests cannot see this at all, because
    they never enter a task group.
    """
    script = _write_server(
        tmp_path,
        "@server.tool()\ndef cost_docs() -> str:\n    raise RuntimeError('server-side failure')\n",
    )
    get = _transport(script, monkeypatch)

    with pytest.raises(IngestError) as excinfo:
        get(f"mcp+stdio://{_SERVER_REF}/cost_docs", None)
    assert excinfo.value.code == "mcp_tool_error"


def test_tool_timeout_reaches_the_caller_as_ingest_error(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A tool that outlives ``timeout_seconds`` must surface as a typed ``IngestError``.

    MEASURED (2026-08-03), not assumed: wrapping ``run()`` — itself two nested anyio task groups
    (``stdio_client``, ``ClientSession``) — in ``asyncio.wait_for`` from OUTSIDE anyio's own
    structure does not raise ``TimeoutError`` at all here. anyio's cancellation and asyncio's do
    not compose across that boundary; the actual failure was an ``anyio.BrokenResourceError``
    inside a ``BaseExceptionGroup`` (a background reader losing its write end mid-teardown). Only
    a REAL hanging server proves this: a canned-tool test never enters a task group and cannot
    observe it. RED until the deadline moves to ``anyio.fail_after``, nested INSIDE both task
    groups, where anyio tears its own structure down cleanly and raises a plain ``TimeoutError``.
    """
    script = _write_server(
        tmp_path,
        "import time\n@server.tool()\ndef cost_docs() -> str:\n    time.sleep(5)\n    return 'late'\n",
    )
    get = _transport(script, monkeypatch, timeout_seconds=0.2)

    with pytest.raises(IngestError) as excinfo:
        get(f"mcp+stdio://{_SERVER_REF}/cost_docs", None)
    assert excinfo.value.code == "mcp_timeout"


def test_unrelated_timeout_error_inside_the_scope_is_not_mislabeled(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A ``TimeoutError`` NOT caused by our own deadline firing must reach the caller untouched.

    Mirrors ``_unwrap_ingest_error``'s ownership rule: own it, wrap it; otherwise, untouched.
    Raised from ``StdioServerParameters`` construction — inside our ``anyio.fail_after`` scope,
    but BEFORE either nested task group (``stdio_client``, ``ClientSession``) is entered, so it
    reaches our ``except`` clause as a bare ``TimeoutError``, not grouped (that grouping is a
    separate, already-covered property — see the ``mcp_tool_error`` test above). ``timeout_seconds``
    is generous (30s) and nothing here waits on it, so the scope never actually cancels: this proves
    the label tracks the SCOPE's ``cancelled_caught``, not the exception TYPE.

    No live trigger for this exists today, checked rather than assumed: a server-side
    ``TimeoutError`` becomes an ordinary ``isError`` result (same as any other tool exception,
    verified against a real server), and MCP's own internal per-request read-timeout
    (``ClientSession.send_request``) converts its ``anyio.fail_after`` timeout to ``McpError``
    before it ever reaches us. This test pins the discriminator against that defect class
    directly, since the class has no reachable path to exercise it end-to-end.
    """

    def boom(*args: object, **kwargs: object) -> None:
        raise TimeoutError("unrelated — not our deadline")

    monkeypatch.setattr("mcp.StdioServerParameters", boom)
    get = _transport(GOLDEN / "server.py", monkeypatch, timeout_seconds=30.0)

    with pytest.raises(TimeoutError, match="unrelated"):
        get(f"mcp+stdio://{_SERVER_REF}/cost_docs", None)


def test_missing_server_ref_fails_before_any_process_is_spawned(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """CONTROL — the fail-fast branch that runs BEFORE the async machinery.

    It proves the unwrap above is not papering over a blanket re-raise: this error never enters a
    task group, so it must arrive as a plain ``IngestError`` both before and after the fix. It also
    pins the §4 promise that a missing env var is refused without spawning anything.
    """
    monkeypatch.delenv(_SERVER_REF, raising=False)
    get = mcp_get(stdio_call_tool())

    with pytest.raises(IngestError) as excinfo:
        get(f"mcp+stdio://{_SERVER_REF}/cost_docs", None)
    assert excinfo.value.code == "mcp_server_ref_unset"