portfolio-optimiser/tests/test_ingest_mcp_loadbearing.py

"""S2.2 load-bearing seams — MCP as an ``http``-family transport (ingest spec §11's detach-RED regime).

Each test is built so the seam under test is the ONLY thing preventing the observable artifact (the
Fase-2 green-but-dead trap). Every test injects a canned ``McpCallTool`` — no subprocess, no socket.

- TRANSPORT DISCRIMINATOR (§4): ``parse_mcp_url`` refuses a URL the MCP transport does not own, and
  accepts one it does. Asserted on BOTH branches with the SAME parser: dropping the prefix check →
  the refuse assertion goes RED (an ``https://`` manifest is silently served over MCP); making the
  refusal unconditional → the accept assertion goes RED. A refuse-only test would false-green on the
  second detach — the trap called out in ``test_ingest_http_loadbearing``.

- CASE PRESERVATION (§4, ``server_ref`` = env-var name): parsing is string-based, not
  ``urlsplit``-based. RED the moment someone "cleans this up" to ``urlsplit().hostname``, which
  lowercases the host and breaks a case-sensitive environment lookup. Both branches again: an
  uppercase ref survives AND a lowercase ref still works, so an ``upper()``-everywhere mutation
  cannot false-green it.

- NETWORK GATE IS INHERITED, NOT RE-IMPLEMENTED (§8): the payoff of the http-family decision. The
  gate fires before any MCP tool call, on BOTH branches (flag off → refuse + tool never called;
  flag on → succeeds + tool called). The CONTROL that gives this test its meaning: the same
  extraction driven around ``materialize`` reaches the tool with no gate at all — proving the
  refusal comes from staying inside the family, not from anything this module wrote.

- NON-TEXT CONTENT IS AN ERROR (§5 verbatim body): a tool returning an image/resource block raises
  rather than silently contributing nothing. RED if the refusal degrades to skipping, which would
  produce a bundle that looks complete and is not. Control: an all-text result still concatenates.

- MAF-FREE (D7): this module may import ``mcp`` (open protocol) but never ``agent_framework`` (one
  framework's runtime), so the sibling Claude-SDK implementation consumes the identical seam. RED on
  the first MAF import.
"""

from __future__ import annotations

import ast
import json
from pathlib import Path

import pytest

from portfolio_optimiser.ingest import NetworkGateError, materialize
from portfolio_optimiser.ingest_mcp import (
    IngestError,
    _body_from_content,
    mcp_get,
    parse_mcp_url,
)

_INGESTED_AT = "2026-08-02T00:00:00Z"


def _manifest(tmp_path: Path, base_url: str) -> Path:
    payload = {
        "manifest_version": 1,
        "source": {"type": "http", "id": "docs", "base_url": base_url},
        "bundle_summary": "MCP-sourced cost documentation.",
        "extractions": [
            {
                "id": "cost-docs",
                "title": "Cost documentation",
                "query": "retrieve_cost_docs",
                "okf_type": "concept",
                "max_rows": 50,
            }
        ],
    }
    path = tmp_path / "manifest.json"
    path.write_text(json.dumps(payload), encoding="utf-8")
    return path


def _recording_call_tool(body: str, calls: list[tuple[str, str, str | None]]):
    def call_tool(server_ref: str, tool: str, credential: str | None) -> str:
        calls.append((server_ref, tool, credential))
        return body

    return call_tool


# --- the transport discriminator gates BOTH ways -------------------------------------------------


def test_transport_discriminator_refuses_foreign_and_accepts_own() -> None:
    """RED if the prefix check is dropped (foreign URL served) OR made unconditional (own URL
    refused). One parser, both branches, so neither mutation survives."""
    with pytest.raises(IngestError):
        parse_mcp_url("https://example.test/api/retrieve")

    assert parse_mcp_url("mcp+stdio://docs-server/retrieve_cost_docs") == (
        "docs-server",
        "retrieve_cost_docs",
    )


def test_transport_discriminator_gates_through_the_seam_before_calling_the_tool() -> None:
    """The gate must sit in FRONT of the tool call, not after it: a refusal that still invoked the
    MCP server would already have paid the call it was meant to prevent."""
    calls: list[tuple[str, str, str | None]] = []
    get = mcp_get(_recording_call_tool("a,b\n1,2", calls))

    with pytest.raises(IngestError):
        get("https://example.test/api/retrieve", None)
    assert calls == [], "refusal must precede the MCP tool call"

    assert get("mcp+stdio://docs-server/retrieve_cost_docs", None) == "a,b\n1,2"
    assert calls == [("docs-server", "retrieve_cost_docs", None)]


def test_server_ref_case_survives_parsing_both_ways() -> None:
    """RED if parsing moves to ``urlsplit().hostname``, which lowercases the host and silently
    breaks a case-sensitive env lookup. The lowercase branch stops an ``upper()`` mutation from
    false-greening the uppercase one."""
    assert parse_mcp_url("mcp+stdio://PORTFOLIO_DOCS_MCP/t") == ("PORTFOLIO_DOCS_MCP", "t")
    assert parse_mcp_url("mcp+stdio://docs-server/t") == ("docs-server", "t")


# --- the §8 gate is inherited from the http family, not written here -----------------------------


def test_network_gate_covers_mcp_on_both_branches(tmp_path: Path) -> None:
    """Same manifest, same recording tool, both flag states. Removing the gate → the refuse
    assertion goes RED (tool called, no error); making the refusal unconditional → the allow
    assertion goes RED."""
    manifest_path = _manifest(tmp_path, "mcp+stdio://docs-server")
    calls: list[tuple[str, str, str | None]] = []
    get = mcp_get(_recording_call_tool("region,saving\nnorth,1200", calls))

    with pytest.raises(NetworkGateError):
        materialize(manifest_path, tmp_path / "off", ingested_at=_INGESTED_AT, http_get=get)
    assert calls == [], "gate must fire BEFORE any MCP tool call"

    written = materialize(
        manifest_path,
        tmp_path / "on",
        ingested_at=_INGESTED_AT,
        allow_network=True,
        http_get=get,
    )
    assert len(calls) == 1, "with the opt-in set, the tool must actually be reached"
    assert written and written[0].exists()


def test_control_the_gate_comes_from_the_family_not_from_this_module(tmp_path: Path) -> None:
    """CONTROL — this is what makes the test above mean something.

    Driving the SAME extraction around ``materialize`` (i.e. what a bespoke fourth-family connector
    would have done) reaches the MCP tool with no gate whatsoever. So the refusal proven above is
    inherited by staying inside ``type: "http"`` — it is not something this module implements, and a
    fourth family would have had to write it, and could have forgotten it."""
    calls: list[tuple[str, str, str | None]] = []
    get = mcp_get(_recording_call_tool("region,saving\nnorth,1200", calls))

    get("mcp+stdio://docs-server/retrieve_cost_docs", None)

    assert calls == [("docs-server", "retrieve_cost_docs", None)], (
        "bypassing materialize reaches the tool ungated — the gate lives in the family path"
    )


# --- verbatim body: non-text content is an error, never a silent drop ----------------------------


class _ImageBlock:
    """An MCP content block with no ``text`` attribute (image / embedded resource)."""

    mime_type = "image/png"


class _TextBlock:
    def __init__(self, text: str) -> None:
        self.text = text


def test_non_text_content_raises_rather_than_dropping_silently() -> None:
    """RED if the refusal degrades to skipping non-text blocks: the body is rendered verbatim into
    the concept file, so dropping part of an extraction yields a bundle that looks complete and is
    not. Control: an all-text result still concatenates in order."""
    with pytest.raises(IngestError):
        _body_from_content([_TextBlock("region,saving\n"), _ImageBlock()], tool="t", server_ref="s")

    assert (
        _body_from_content([_TextBlock("a,b\n"), _TextBlock("1,2")], tool="t", server_ref="s")
        == "a,b\n1,2"
    )


# --- D7 portability ------------------------------------------------------------------------------


def test_ingest_mcp_is_maf_free() -> None:
    """MCP is an open protocol and may be imported; ``agent_framework`` is one framework's runtime
    and may not. Keeps this transport consumable unchanged by the sibling Claude-SDK stack. RED on
    the first MAF import."""
    module_path = (
        Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" / "ingest_mcp.py"
    )
    tree = ast.parse(module_path.read_text(encoding="utf-8"))

    roots: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            roots.extend(alias.name.split(".")[0] for alias in node.names)
        elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
            roots.append(node.module.split(".")[0])

    assert "agent_framework" not in roots, f"MAF import in ingest_mcp.py: {roots}"
    assert "mcp" in roots, (
        "the MCP transport is expected to import the open-protocol client; if this moved, "
        "re-point the guard rather than deleting it"
    )


def test_the_seam_is_opt_in_at_the_call_site() -> None:
    """``portfolio_optimiser.ingest`` must NOT import this module: it is AST-guarded mcp-free, and
    the MCP transport is opt-in via ``materialize(..., http_get=mcp_get(...))``. That is also what
    stops a plain ``http`` manifest from quietly acquiring an MCP transport."""
    ingest_path = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" / "ingest.py"
    source = ingest_path.read_text(encoding="utf-8")
    assert "ingest_mcp" not in source, "ingest.py must stay independent of the MCP transport"