"""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"