ersatztv/scripts/tests/test_mcp_smoke.py

"""`scripts/mcp_smoke.py` answers its own question correctly (ersatztv#796).

WHY THIS FILE EXISTS AND WHY IT IS NOT A ROW FOR `mcp_smoke.py`. That script is the checker that
carried three of #793's five false greens: it accepted any reply carrying the right id, its
message/line caps re-parsed an over-long line's suffix as a fresh message, and it registered a
pending id before sending so a server could pre-answer a predictable one. Each was reproduced with a
control at the time and then fixed, and NONE of those controls survived as a test — so nothing
re-ran them and a later edit could silently undo what they proved. `testing.verification-code-needs-
its-own-proof` is the rule that says a checker owes an executed proof; this is that proof.

`mcp_smoke.py` itself gets no `docs/guard-inventory.md` row: the inventory's population is derived
from workflow and hook call sites and this script is reached only transitively, through
`scripts/check-local-lsp.sh` (the transitive-calls scope limit, item 6), so a row for it is rejected as a phantom. THIS
file joins the population as an ordinary `scripts/tests/test_*.py` and carries the row, and the
manifest entry names it as `guard` with `scripts/mcp_smoke.py` as `target` — the same shape
`test_mutation_harness.py` uses for `mutation_harness_lib.py`.

HERMETIC BY CONSTRUCTION, which is what makes the proof possible at all. `mcp_smoke.py` takes its
config path and server name as POSITIONAL ARGUMENTS, so nothing here touches the gitignored
`.mcp.json` or starts a language server: every case below drives it against a synthetic config in a
tmpdir pointing at the stub in `_STUB`. "It needs a real server" exempts the INTEGRATION —
`check-local-lsp.sh` driving the actual csharp-lsp — never this protocol logic.

WHAT IS COVERED, stated as a list rather than as "every branch", because the honest figure is a
subset: the success path, and the refusals for a pre-answered `initialize` (9), a pre-answered
`tools/list` (10), an `initialize` reply with no `result` (12), a wrong `serverInfo.name` (13) and a
missing expected tool (14). `mcp_smoke.py` has many other `fail()` sites — among them the usage,
config, command-resolution, `--project` and spawn stages, an `initialize` that returns an error,
zero well-formed tools, and further malformed shapes — and none of those is driven here.

NOT TESTED, DELIBERATELY: the unbounded raw line buffer. `mcp_smoke.py` states that as an ACCEPTED
LIMIT, because the caps that would bound it were themselves defect 4. A test pushing toward
reintroducing them would drive the file back into the failure it already paid for.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
SMOKE = REPO_ROOT / "scripts" / "mcp_smoke.py"

# The stub speaks just enough JSON-RPC to exercise each case. The two `preanswer` arms are the
# security cases: they emit replies for the LOW, GUESSABLE ids a predictable client would use,
# BEFORE the corresponding request arrives, and then go silent — so they can only be accepted by a
# client whose ids can be guessed. Each drops a MARKER file once it has emitted, because otherwise
# a stub that silently emitted NOTHING would satisfy the same assertion: "no answer" and "the
# pre-answer was refused" are the same observation from outside, and only the marker separates them.
_STUB = r"""
import json, sys, time

mode, marker = sys.argv[1], sys.argv[2]

def emit(obj):
    sys.stdout.write(json.dumps(obj) + "\n")
    sys.stdout.flush()

def mark():
    with open(marker, "w") as fh:
        fh.write("emitted\n")

def init_result(name="stub-mcp"):
    return {"serverInfo": {"name": name, "version": "0.1"}, "capabilities": {}}

def tools_result(names):
    return {"tools": [{"name": n} for n in names]}

if mode == "preanswer_init":
    emit({"jsonrpc": "2.0", "id": 1, "result": init_result()})
    mark()
    time.sleep(30)
    sys.exit(0)

for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    try:
        msg = json.loads(line)
    except ValueError:
        continue
    mid, method = msg.get("id"), msg.get("method")
    if method == "initialize":
        if mode == "impostor":
            emit({"jsonrpc": "2.0", "id": mid, "result": init_result("impostor-mcp")})
        elif mode == "no_result":
            emit({"jsonrpc": "2.0", "id": mid})
        else:
            emit({"jsonrpc": "2.0", "id": mid, "result": init_result()})
        if mode == "preanswer_tools":
            # answer initialize honestly, then pre-answer the tools/list id we hope to guess
            emit({"jsonrpc": "2.0", "id": 2, "result": tools_result(["alpha", "beta"])})
            mark()
    elif method == "tools/list":
        if mode == "preanswer_tools":
            continue  # already "answered" it above, at a guessed id
        names = ["alpha"] if mode == "missing_tool" else ["alpha", "beta"]
        emit({"jsonrpc": "2.0", "id": mid, "result": tools_result(names)})
"""


def _run(tmp_path: Path, mode: str, *extra: str, timeout: int = 5):
    """Drive `mcp_smoke.py` against the stub in `mode`. Returns (rc, output, marker_path)."""
    stub = tmp_path / "stub_server.py"
    stub.write_text(_STUB)
    marker = tmp_path / "emitted.marker"
    cfg = tmp_path / "synthetic.mcp.json"
    cfg.write_text(
        json.dumps({"mcpServers": {"stub": {"command": sys.executable, "args": [str(stub), mode, str(marker)]}}})
    )
    proc = subprocess.run(
        [sys.executable, str(SMOKE), str(cfg), "stub", str(timeout), *extra],
        capture_output=True,
        text=True,
        timeout=timeout + 40,
    )
    return proc.returncode, proc.stdout + proc.stderr, marker


# ------------------------------------------------------------------------------------------------
# ANTI-VACUITY AS A DEPENDENCY, NOT A CONVENTION. Every case below except the positive control itself
# asserts a REFUSAL, and a harness that could not make the checker SUCCEED would report all of those
# green while proving nothing. A
# sibling test placed first does not establish that: `-k`, a node-id selection, or the mutation
# harness running one test by id all skip it. A fixture the rejection tests DEPEND on cannot be
# skipped that way.
# ------------------------------------------------------------------------------------------------


@pytest.fixture(scope="module")
def harness_is_live(tmp_path_factory) -> None:
    rc, out, _ = _run(
        tmp_path_factory.mktemp("positive_control"),
        "good",
        "--expect-server",
        "stub-mcp",
        "--expect-tool",
        "alpha",
    )
    assert rc == 0 and "OK:" in out and "stub-mcp" in out, (
        f"the positive control did not pass, so every refusal asserted in this file proves nothing "
        f"— the stub or the config shape is broken, not the checker (rc={rc}):\n{out}"
    )


def test_the_harness_can_make_the_checker_PASS(harness_is_live) -> None:
    """The positive control, surfaced as a test so it is visible in the report as well as depended on."""


def test_an_IMPOSTOR_server_is_rejected_on_identity(harness_is_live, tmp_path) -> None:
    """#793 defect 2: 'it answered' is not 'the configured server answered'."""
    rc, out, _ = _run(tmp_path, "impostor", "--expect-server", "stub-mcp")
    assert rc == 13, f"an impostor server was accepted (rc={rc}): {out}"
    assert "wrong server" in out, out


def test_a_reply_carrying_only_an_ID_is_rejected_as_malformed(harness_is_live, tmp_path) -> None:
    """#793 defect 2, the other half: a bare `{"id": N}` is a reply, not an answer."""
    rc, out, _ = _run(tmp_path, "no_result")
    assert rc == 12, f"a body with no 'result' was accepted (rc={rc}): {out}"
    assert "no 'result' object" in out, out


def test_a_MISSING_expected_tool_is_reported(harness_is_live, tmp_path) -> None:
    rc, out, _ = _run(tmp_path, "missing_tool", "--expect-tool", "alpha", "--expect-tool", "beta")
    assert rc == 14, f"a server missing an expected tool was accepted (rc={rc}): {out}"
    assert "missing expected tool" in out, out


def test_MUTATION_a_PRE_ANSWERED_id_is_refused_because_the_request_ids_are_UNGUESSABLE(
    harness_is_live, tmp_path
) -> None:
    """#793 defect 5, and the declared clause mutation for this guard.

    The stub answers id 1 before reading anything, then goes silent. With random request ids the
    pre-answer cannot match, so `initialize` times out and the run is refused at THAT stage. Replace
    `id_init = secrets.randbelow(...)` with `id_init = 1` and the pre-answer is accepted: the run
    gets past identity and dies one stage later, at `tools/list`. Asserting the STAGE rather than
    merely 'it failed' is what makes the mutation visible — the mutant still fails, just not here.

    The marker assertion is not decoration. Without it this test is satisfied by a stub that emits
    NOTHING AT ALL, since a silent server also produces rc 9 — an absence standing in for the
    refusal the test claims to observe.
    """
    rc, out, marker = _run(tmp_path, "preanswer_init", "--expect-server", "stub-mcp")
    assert marker.exists(), (
        "the stub never emitted its pre-answer, so this case did not pose the attack it narrates: "
        "rc 9 below would only mean the server said nothing."
    )
    assert rc == 9 and "no 'initialize' response" in out, (
        f"a pre-answered id was ACCEPTED: the checker got past `initialize` against a server that "
        f"replied before it was asked (rc={rc}). At THIS stage an unguessable id is the only thing "
        f"closing that race — the pending-registration cannot help, because the id is already in "
        f"flight when the pre-answer arrives.\n{out}"
    )


def test_a_PRE_ANSWERED_tools_list_is_refused_by_TWO_mechanisms_TOGETHER(harness_is_live, tmp_path) -> None:
    """The `tools/list` twin of the case above — and the measurement that explains why `id_tools` is
    NOT this guard's declared clause.

    `mcp_smoke.py` draws two unguessable ids, and only `id_init` is singly load-bearing. Measured on
    an isolated copy: replacing `id_tools` alone leaves every case here green, because the reader
    retains only the reply matching the id currently in flight (`if rid != pending ... continue`), so
    a `tools/list` frame emitted while `initialize` is still pending is dropped whatever its id.
    Disarming that retention clause alone is likewise green, because the real `id_tools` is still
    unguessable. Disarm BOTH and this case goes red at rc 0 — the checker accepts a tool list at an
    id it never requested.

    So the tools stage is held by two mechanisms that mask each other (#685's shape, benign here but
    worth naming): neither is detectable alone, which is exactly why the declared clause is the
    `initialize` id, where the pre-answer IS singly exploitable. This case is behavioural coverage
    for the pair, not a second mutation proof.
    """
    rc, out, marker = _run(tmp_path, "preanswer_tools", "--expect-server", "stub-mcp", "--expect-tool", "alpha")
    assert marker.exists(), "the stub never emitted its pre-answered tools/list, so this case posed nothing"
    assert rc == 10 and "no 'tools/list' response" in out, (
        f"a pre-answered tools/list was ACCEPTED: the checker took a tool list at an id it never "
        f"requested (rc={rc}). That needs BOTH the pending-registration and an unguessable "
        f"`id_tools`; this went green with one of them gone.\n{out}"
    )