diff --git a/.circleci/config.yml b/.circleci/config.yml index 7276da9877b..843660383fb 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -3241,6 +3241,17 @@ jobs: - run: name: Build the candidate dashboard command: cd ui/litellm-dashboard && NEXT_TELEMETRY_DISABLED=1 npm run build + - when: + condition: + equal: [mcp, << parameters.suite >>] + steps: + - install_node + - run: + name: Install pinned official MCP conformance runner and reference + command: | + export MCP_CONFORMANCE_ROOT="/tmp/litellm-mcp-conformance" + bash .circleci/scripts/install_mcp_conformance.sh + echo 'export MCP_CONFORMANCE_ROOT="/tmp/litellm-mcp-conformance"' >> "$BASH_ENV" - start_postgres: image: postgres:16@sha256:e17e86066e5ef83e0952a9347f5c792b7ece00972e2aa787a6986f471b3dd3d5 server_args: "-c shared_preload_libraries=pg_stat_statements -c pg_stat_statements.track=all -c pg_stat_statements.max=20000" @@ -3249,6 +3260,13 @@ jobs: name: Run owned integration contracts command: bash .circleci/scripts/run_integration.sh << parameters.suite >> << parameters.mode >> no_output_timeout: 15m + - when: + condition: + equal: [mcp, << parameters.suite >>] + steps: + - codecov/upload: + file: test-results/integration-mcp/conformance-coverage.xml + flags: mcp-conformance - run: name: Stop owned database and Redis when: always diff --git a/.circleci/scripts/install_mcp_conformance.sh b/.circleci/scripts/install_mcp_conformance.sh new file mode 100644 index 00000000000..8bc5eb22750 --- /dev/null +++ b/.circleci/scripts/install_mcp_conformance.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +set -euo pipefail + +root="${MCP_CONFORMANCE_ROOT:?Set MCP_CONFORMANCE_ROOT to a fresh directory}" +test ! -e "$root" +archive="$(mktemp)" +trap 'rm -f "$archive"' EXIT +curl --fail --location --silent --show-error \ + 'https://codeload.github.com/modelcontextprotocol/conformance/tar.gz/7169291ec0b68eb370fddcd9947313ab0d5e4156' \ + --output "$archive" +PYTHONPATH=tests .venv/bin/python - "$archive" <<'PY' +import sys +from pathlib import Path +from integration._support.conformance import verify_archive + +verify_archive(Path(sys.argv[1]), "51c1e27027f36be5b5f067746eb3a0f240fc3fbb7adcbd64bc4a79d86db65cd8") +PY +mkdir -p "$root" +tar -xzf "$archive" --strip-components=1 -C "$root" +npm ci --ignore-scripts --prefix "$root" +# The current reference misclassifies legacy initialize _meta as stateless traffic. +# Use the last official legacy reference; leave its source and lockfile untouched. +curl --fail --location --silent --show-error \ + 'https://codeload.github.com/modelcontextprotocol/conformance/tar.gz/8f3994c75ff1aed1e39f91cff9358e2bc2c81dcd' \ + --output "$archive" +PYTHONPATH=tests .venv/bin/python - "$archive" <<'PYREF' +import sys +from pathlib import Path +from integration._support.conformance import verify_archive + +verify_archive(Path(sys.argv[1]), "181119bd222f29db208b2952537430fcc865f91b64f2ac3d7525a67f490ebaa1") +PYREF +mkdir "$root/legacy-reference" +tar -xzf "$archive" --strip-components=1 -C "$root/legacy-reference" +npm ci --ignore-scripts --prefix "$root/legacy-reference/examples/servers/typescript" diff --git a/.circleci/scripts/run_integration.sh b/.circleci/scripts/run_integration.sh index 47ad2274e2f..d4418c8a73c 100644 --- a/.circleci/scripts/run_integration.sh +++ b/.circleci/scripts/run_integration.sh @@ -221,6 +221,12 @@ if [ "${CIRCLE_NODE_TOTAL:-1}" -gt 1 ]; then printf '%s\n' "${node_files[@]}" > "$results/node-files.txt" fi +test_command=(.venv/bin/python tests/integration/run.py "$suite" --results "$results" "${node_files[@]}") +if [ "$suite" = mcp ]; then + test_command=(.venv/bin/python -m coverage run --rcfile=tests/integration/conformance_coverage.toml \ + tests/integration/run.py "$suite" --results "$results" "${node_files[@]}") +fi + env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \ INTEGRATION_RUN_ID="$integration_identity" \ DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \ @@ -234,7 +240,17 @@ env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \ INTEGRATION_PROXY_DATABASE_URL="$INTEGRATION_PROXY_DATABASE_URL" \ INTEGRATION_PROXY_READ_REPLICA_URL="$INTEGRATION_PROXY_READ_REPLICA_URL" \ INTEGRATION_ROUTING="$INTEGRATION_ROUTING" \ - .venv/bin/python tests/integration/run.py "$suite" --results "$results" "${node_files[@]}" + MCP_CONFORMANCE_ROOT="${MCP_CONFORMANCE_ROOT:-}" \ + COVERAGE_FILE="$PWD/$results/.coverage-conformance" "${test_command[@]}" + +if [ "$suite" = mcp ]; then + export COVERAGE_FILE="$PWD/$results/.coverage-conformance" + .venv/bin/python -m coverage run --rcfile=tests/integration/conformance_coverage.toml \ + -m pytest tests/unit/integration_support/test_conformance.py -q + .venv/bin/python -m coverage combine --rcfile=tests/integration/conformance_coverage.toml + .venv/bin/python -m coverage xml --rcfile=tests/integration/conformance_coverage.toml \ + -o "$results/conformance-coverage.xml" +fi if [ "${INTEGRATION_COVERAGE:-0}" = 1 ]; then for covered_pid in "$proxy_pid" "$peer_pid"; do diff --git a/tests/integration/_support/conformance.py b/tests/integration/_support/conformance.py new file mode 100644 index 00000000000..f864bfce666 --- /dev/null +++ b/tests/integration/_support/conformance.py @@ -0,0 +1,174 @@ +import hashlib +import json +import os +import queue +import signal +import subprocess +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from typing import TYPE_CHECKING, Final, Literal + +import httpx +import psutil +from pydantic import BaseModel, Field, JsonValue, TypeAdapter + +if TYPE_CHECKING: + from integration._support.mcp import McpPeer + + +class ConformanceCheck(BaseModel): + id: str + status: Literal["SUCCESS", "FAILURE", "WARNING", "INFO"] + errorMessage: str | None = None + details: dict[str, JsonValue] = Field(default_factory=dict) + + +def verify_archive(archive: Path, expected_sha256: str) -> None: + actual: Final = hashlib.sha256(archive.read_bytes()).hexdigest() + assert actual == expected_sha256, f"SHA-256 mismatch for {archive.name}: {actual}" + + +def read_checks(directory: Path, scenario: str) -> tuple[ConformanceCheck, ...]: + reports: Final = tuple(directory.rglob("checks.json")) + assert len(reports) == 1, f"conformance {scenario}: expected one fresh report, found {len(reports)}" + checks: Final = TypeAdapter(tuple[ConformanceCheck, ...]).validate_json(reports[0].read_bytes()) + assert len({check.id for check in checks}) == len(checks), f"conformance {scenario}: duplicate checks" + required: Final = { + "server-initialize": ("server-initialize", "server-session-id-visible-ascii", "wire-schema-valid"), + "tools-list": ("tools-list", "tools-name-format", "wire-schema-valid"), + "tools-call-image": ("tools-call-image", "wire-schema-valid"), + }.get(scenario, (scenario,)) + for identity in required: + assert tuple(check.status for check in checks if check.id == identity) == ("SUCCESS",), ( + f"conformance {scenario}: required check {identity} did not pass: {checks}" + ) + assert all(check.status != "FAILURE" for check in checks), f"conformance {scenario}: failed checks: {checks}" + if scenario == "tools-call-simple-text": + result: Final = next(check for check in checks if check.id == scenario).details.get("result") + # The official scenario accepts any nonempty text, including tool errors. + assert isinstance(result, dict) and result.get("isError", False) is False, f"reference payload: {result}" + assert result.get("content") == [{"type": "text", "text": "This is a simple text response for testing."}], ( + f"reference payload: {result}" + ) + return checks + + +def run_scenario(root: Path, url: str, scenario: str, directory: Path) -> tuple[ConformanceCheck, ...]: + directory.mkdir(parents=True, exist_ok=False) + with (directory / "runner.log").open("w") as log: + result: Final = subprocess.run( + [ + "node", + "--import", + "tsx", + "src/index.ts", + "server", + "--url", + url, + "--scenario", + scenario, + "--spec-version", + "2025-11-25", + "--output-dir", + str(directory.resolve()), + "--timeout", + "30000", + ], + cwd=root, + stdout=log, + stderr=subprocess.STDOUT, + timeout=45, + check=False, + ) + assert result.returncode == 0, (directory / "runner.log").read_text() + return read_checks(directory, scenario) + + +@contextmanager +def reference_server(root: Path, directory: Path, port: int) -> Iterator["McpPeer"]: + from integration._support.client import eventually + from integration._support.mcp import McpPeer + from integration._support.process import group_members, signal_group, stop_root_process + + directory.mkdir(parents=True, exist_ok=True) + url: Final = f"http://127.0.0.1:{port}" + with (directory / "reference.log").open("w") as log: + process: Final = subprocess.Popen( + ["node", "--import", "tsx", "everything-server.ts"], + cwd=root / "legacy-reference/examples/servers/typescript", + env={"PATH": os.environ["PATH"], "PORT": str(port)}, + stdout=log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + try: + with httpx.Client(timeout=1, trust_env=False) as client: + + def ready() -> bool: + assert process.poll() is None, (directory / "reference.log").read_text() + try: + return client.get(url + "/mcp").status_code == 400 + except httpx.TransportError: + return False + + eventually(ready, bool, seconds=15) + yield McpPeer(url + "/mcp", queue.Queue()) + finally: + stopped: Final = stop_root_process(process) + residual: Final = group_members(process.pid) + if residual: + signal_group(process.pid, signal.SIGTERM) + psutil.wait_procs(residual, timeout=5) + remaining: Final = group_members(process.pid) + if remaining: + signal_group(process.pid, signal.SIGKILL) + psutil.wait_procs(remaining, timeout=3) + process.wait(timeout=3) + assert not group_members(process.pid), "Official reference child survived cleanup" + assert stopped and not remaining, "Official reference required forced cleanup" + + +@contextmanager +def authenticated_endpoint(target: str, key: str, alias: str) -> Iterator[str]: + from integration._support.asgi import asgi_server + from starlette.applications import Starlette + from starlette.requests import Request + from starlette.responses import StreamingResponse + from starlette.routing import Mount + from starlette.types import Receive, Scope, Send + + async def forward(scope: Scope, receive: Receive, send: Send) -> None: + request: Final = Request(scope, receive) + original: Final = await request.body() + body: Final = prefixed_request(original, alias) + headers: Final = tuple( + (name, value) + for name, value in request.headers.raw + if name.lower() != b"authorization" and (name.lower() != b"content-length" or body == original) + ) + ((b"authorization", f"Bearer {key}".encode()),) + async with httpx.AsyncClient(timeout=35, trust_env=False) as client: + async with client.stream(request.method, target, headers=headers, content=body) as response: + streamed: Final = StreamingResponse(response.aiter_raw(), status_code=response.status_code) + streamed.raw_headers = [ + (name, value) + for name, value in response.headers.raw + if name.lower() not in (b"transfer-encoding", b"connection") + ] + await streamed(scope, receive, send) + + with asgi_server(Starlette(routes=[Mount("/mcp", app=forward)])) as url: + yield url + "/mcp/" + + +def prefixed_request(body: bytes, alias: str) -> bytes: + try: + request: Final = json.loads(body) + except (ValueError, UnicodeDecodeError): + return body + if not isinstance(request, dict) or request.get("method") not in ("tools/call", "prompts/get"): + return body + params: Final = request.get("params") + if not isinstance(params, dict) or not isinstance(params.get("name"), str): + return body + return json.dumps({**request, "params": {**params, "name": f"{alias}-{params['name']}"}}).encode() diff --git a/tests/integration/_support/process.py b/tests/integration/_support/process.py index fcbaf7c8d8c..9d5d7a17db7 100644 --- a/tests/integration/_support/process.py +++ b/tests/integration/_support/process.py @@ -103,6 +103,8 @@ def owned_proxy_process( "LITELLM_MASTER_KEY": gateway.key, "LITELLM_SALT_KEY": os.environ.get("LITELLM_SALT_KEY", "sk-integration-salt"), "STORE_MODEL_IN_DB": "True", + "COVERAGE_PROCESS_CONFIG": "", + "COVERAGE_PROCESS_START": "", **overrides, } output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(directory))) diff --git a/tests/integration/conformance_coverage.toml b/tests/integration/conformance_coverage.toml new file mode 100644 index 00000000000..ca1c2724341 --- /dev/null +++ b/tests/integration/conformance_coverage.toml @@ -0,0 +1,9 @@ +[tool.coverage.run] +branch = true +parallel = true +relative_files = true +patch = ["subprocess"] +include = ["tests/integration/_support/conformance.py", "tests/integration/_support/process.py"] + +[tool.coverage.report] +show_missing = true diff --git a/tests/integration/mcp/test_mcp_official_conformance.py b/tests/integration/mcp/test_mcp_official_conformance.py new file mode 100644 index 00000000000..5e585f65136 --- /dev/null +++ b/tests/integration/mcp/test_mcp_official_conformance.py @@ -0,0 +1,144 @@ +import asyncio +import json +import os +import uuid +from pathlib import Path +from typing import Final + +import httpx +import pytest +from integration._support.client import Gateway +from integration._support.conformance import authenticated_endpoint, reference_server, run_scenario +from integration._support.mcp import official_client_outcomes, register_mcp +from integration._support.wire import Reply, wire_server +from mcp import ClientSession +from mcp.client.streamable_http import streamable_http_client +from mcp.types import ImageContent + + +@pytest.mark.parametrize("name", ("server-initialize", "tools-list", "tools-call-image")) +def test_official_scenario_through_gateway(gateway: Gateway, tmp_path: Path, unused_tcp_port: int, name: str) -> None: + root: Final = Path(os.environ["MCP_CONFORMANCE_ROOT"]) + output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(tmp_path))) / f"conformance-{uuid.uuid4().hex}" + with reference_server(root, output, unused_tcp_port) as reference, gateway.scenario() as scenario: + alias: Final = "official" + uuid.uuid4().hex[:8] + identity: Final = register_mcp(scenario, reference, alias, mcp_info={"protocol_version": "2025-11-25"}) + key: Final = scenario.key(object_permission={"mcp_servers": [identity]}) + direct: Final = run_scenario(root, reference.url, name, output / "direct") + endpoint: Final = str(gateway.client.base_url).rstrip("/") + f"/{alias}/mcp" + with authenticated_endpoint(endpoint, key, alias) as authenticated: + proxied: Final = run_scenario(root, authenticated, name, output / "gateway") + if name == "tools-call-image": + + async def check_image(url: str, tool: str, token: str | None = None) -> None: + async with httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"} if token else {}) as http: + async with streamable_http_client(url, http_client=http) as streams: + async with ClientSession(streams[0], streams[1]) as session: + await session.initialize() + result: Final = await session.call_tool(tool, {}) + assert result.is_error is False and len(result.content) == 1, result + content: Final = result.content[0] + assert isinstance(content, ImageContent) and content.mime_type == "image/png", content + assert content.data == ( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlE" + "QVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg==" + ), content + + asyncio.run(check_image(reference.url, "test_image_content")) + asyncio.run(check_image(endpoint, f"{alias}-test_image_content", key)) + assert tuple(check.status for check in direct if check.id == name) == ("SUCCESS",) + assert tuple(check.status for check in proxied if check.id == name) == ("SUCCESS",) + listed, called = official_client_outcomes(gateway, key, f"/{alias}/mcp", f"{alias}-test_simple_text", {}) + assert f"{alias}-test_simple_text" in listed.tools, listed + assert called.ok and called.text == "This is a simple text response for testing.", called + + +@pytest.mark.parametrize("status", (200, 403)) +def test_conformance_bridge_preserves_headers_payload_and_error_status(status: int) -> None: + body: Final = {"jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "test_image_content"}} + response_body: Final = b'data: {"jsonrpc":"2.0","id":7,"error":{"code":-32000,"message":"denied"}}\n\n' + with wire_server( + lambda request: Reply( + status=status, chunks=(response_body[:12], response_body[12:]), content_type="text/event-stream" + ) + ) as upstream: + with authenticated_endpoint(upstream.url + "/mcp", "test-key", "official") as endpoint: + response: Final = httpx.post( + endpoint, + json=body, + headers={ + "Host": "127.0.0.1", + "Origin": "https://untrusted.example", + "Authorization": "Bearer replaced", + "MCP-Protocol-Version": "2025-03-26", + "Mcp-Session-Id": "same-session", + }, + ) + assert response.status_code == status and response.content == response_body + requests: Final = upstream.drain() + assert len(requests) == 1 + received: Final = requests[0] + assert json.loads(received.body) == {**body, "params": {"name": "official-test_image_content"}} + assert { + name: received.headers[name] + for name in ("host", "origin", "authorization", "mcp-protocol-version", "mcp-session-id") + } == { + "host": "127.0.0.1", + "origin": "https://untrusted.example", + "authorization": "Bearer test-key", + "mcp-protocol-version": "2025-03-26", + "mcp-session-id": "same-session", + } + + +def test_official_runner_rejects_unknown_scenario(tmp_path: Path) -> None: + output: Final = tmp_path / "unknown-scenario" + with pytest.raises(AssertionError): + run_scenario(Path(os.environ["MCP_CONFORMANCE_ROOT"]), "http://127.0.0.1:1/mcp", "missing-scenario", output) + assert "missing-scenario" in (output / "runner.log").read_text() + + +def test_stalled_reference_is_killed_and_cannot_report_clean_teardown(tmp_path: Path, unused_tcp_port: int) -> None: + import signal + from contextlib import ExitStack + + import psutil + + children: Final = frozenset(child.pid for child in psutil.Process().children()) + with ExitStack() as cleanup: + cleanup.enter_context(reference_server(Path(os.environ["MCP_CONFORMANCE_ROOT"]), tmp_path, unused_tcp_port)) + started: Final = tuple(child for child in psutil.Process().children() if child.pid not in children) + assert len(started) == 1, started + victim: Final = started[0] + victim.send_signal(signal.SIGSTOP) + with pytest.raises(AssertionError, match="forced cleanup"): + cleanup.close() + assert not psutil.pid_exists(victim.pid), "Stopped official reference survived forced cleanup" + + +def test_reference_children_are_stopped_after_the_root_exits(tmp_path: Path, unused_tcp_port: int) -> None: + import psutil + + source: Final = tmp_path / "legacy-reference/examples/servers/typescript" + source.mkdir(parents=True) + (source / "node_modules").symlink_to( + Path(os.environ["MCP_CONFORMANCE_ROOT"]) / "legacy-reference/examples/servers/typescript/node_modules", + target_is_directory=True, + ) + (source / "everything-server.ts").write_text( + "import http from 'node:http';\n" + "import { spawn } from 'node:child_process';\n" + "import { writeFileSync } from 'node:fs';\n" + "const child = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], {stdio: 'ignore'});\n" + "writeFileSync('child.pid', String(child.pid));\n" + "http.createServer((req, res) => { res.writeHead(400); res.end(); }).listen(Number(process.env.PORT));\n" + "process.on('SIGTERM', () => process.exit(0));\n" + ) + with reference_server(tmp_path, tmp_path / "logs", unused_tcp_port): + child: Final = psutil.Process(int((source / "child.pid").read_text())) + try: + assert not child.is_running(), "Reference child survived a clean root exit" + finally: + if child.is_running(): + child.kill() + psutil.wait_procs((child,), timeout=3) diff --git a/tests/unit/integration_support/test_conformance.py b/tests/unit/integration_support/test_conformance.py new file mode 100644 index 00000000000..8f0acda74ca --- /dev/null +++ b/tests/unit/integration_support/test_conformance.py @@ -0,0 +1,183 @@ +import hashlib +import json +import os +from pathlib import Path +from typing import Final +from unittest.mock import Mock, patch + +import pytest +from tests.integration._support.conformance import read_checks, verify_archive, prefixed_request +from pydantic import ValidationError + + +@pytest.mark.parametrize("explicit", (False, True)) +def test_owned_proxy_isolates_automatic_coverage_unless_requested( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, explicit: bool +) -> None: + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2])) + from integration._support.client import Gateway + from integration._support.process import owned_proxy_process + + monkeypatch.setenv("COVERAGE_PROCESS_CONFIG", "parent-config") + monkeypatch.setenv("COVERAGE_PROCESS_START", "parent.toml") + overrides: Final = {"COVERAGE_PROCESS_CONFIG": "child-config"} if explicit else {} + gateway: Final = Gateway(Mock(), "owner-key", "http://upstream") + with ( + patch("integration._support.process.subprocess.Popen") as spawn, + patch("integration._support.process.httpx.Client") as client, + patch("integration._support.process.group_members", return_value=()), + ): + spawn.return_value.poll.return_value = None + client.return_value.__enter__.return_value.get.return_value.status_code = 200 + with owned_proxy_process(gateway, tmp_path, overrides): + child: Final = spawn.call_args.kwargs["env"] + assert child["COVERAGE_PROCESS_CONFIG"] == ("child-config" if explicit else "") + assert child["COVERAGE_PROCESS_START"] == "" + assert child["LITELLM_MASTER_KEY"] == gateway.key + assert os.environ["COVERAGE_PROCESS_CONFIG"] == "parent-config" + + +def test_changed_archive_is_rejected(tmp_path: Path) -> None: + archive: Final = tmp_path / "reference.tar.gz" + archive.write_bytes(b"changed reference") + with pytest.raises(AssertionError, match="SHA-256"): + verify_archive(archive, hashlib.sha256(b"approved reference").hexdigest()) + + +def test_matching_archive_is_accepted(tmp_path: Path) -> None: + archive: Final = tmp_path / "reference.tar.gz" + archive.write_bytes(b"approved reference") + assert verify_archive(archive, "5a870d3ee9520e3a912c735d007d7b40f6d3cbf8b61d909ebb8361f423d4ba1e") is None + + +@pytest.mark.parametrize( + "checks", + ( + [], + [{"id": "tools-list", "status": "FAILURE"}], + [{"id": "tools-list", "status": "WARNING"}], + [{"id": "wire-schema", "status": "SUCCESS"}], + [{"id": "tools-list", "status": "SUCCESS"}, {"id": "wire-schema", "status": "FAILURE"}], + [{"id": "tools-list", "status": "SUCCESS"}, {"id": "tools-list", "status": "SUCCESS"}], + ), +) +def test_incomplete_or_failed_scenario_cannot_pass(tmp_path: Path, checks: list[dict[str, str]]) -> None: + (tmp_path / "checks.json").write_text(json.dumps(checks)) + with pytest.raises(AssertionError, match="conformance"): + read_checks(tmp_path, "tools-list") + + +def test_skipped_scenario_without_a_report_cannot_pass(tmp_path: Path) -> None: + with pytest.raises(AssertionError, match="conformance"): + read_checks(tmp_path, "tools-list") + + +def test_multiple_reports_cannot_supply_a_stale_pass(tmp_path: Path) -> None: + for name in ("previous", "current"): + directory: Final = tmp_path / name + directory.mkdir() + (directory / "checks.json").write_text('[{"id":"tools-list","status":"SUCCESS"}]') + with pytest.raises(AssertionError, match="conformance"): + read_checks(tmp_path, "tools-list") + + +def test_passing_scenario_keeps_nonbinding_diagnostics(tmp_path: Path) -> None: + (tmp_path / "checks.json").write_text( + json.dumps( + [ + {"id": identity, "status": "INFO" if identity == "advisory" else "SUCCESS"} + for identity in ("tools-list", "tools-name-format", "wire-schema-valid", "advisory") + ] + ) + ) + checks: Final = read_checks(tmp_path, "tools-list") + assert checks[-1].id == "advisory" and checks[-1].status == "INFO" + + +def test_unrecognized_result_status_cannot_pass(tmp_path: Path) -> None: + (tmp_path / "checks.json").write_text('[{"id":"tools-list","status":"SKIPPED"}]') + with pytest.raises(ValidationError, match="status"): + read_checks(tmp_path, "tools-list") + + +@pytest.mark.parametrize( + "result", + ( + {"content": [{"type": "text", "text": "Tool not found"}], "isError": True}, + {"content": [{"type": "text", "text": "wrong upstream payload"}]}, + ), +) +def test_official_text_success_cannot_hide_an_error_or_wrong_payload(tmp_path: Path, result: dict[str, object]) -> None: + (tmp_path / "checks.json").write_text( + json.dumps([{"id": "tools-call-simple-text", "status": "SUCCESS", "details": {"result": result}}]) + ) + with pytest.raises(AssertionError, match="reference payload"): + read_checks(tmp_path, "tools-call-simple-text") + + +def test_official_text_success_requires_the_reference_payload(tmp_path: Path) -> None: + (tmp_path / "checks.json").write_text( + json.dumps( + [ + { + "id": "tools-call-simple-text", + "status": "SUCCESS", + "details": { + "result": { + "content": [{"type": "text", "text": "This is a simple text response for testing."}], + "isError": False, + } + }, + } + ] + ) + ) + assert read_checks(tmp_path, "tools-call-simple-text")[0].status == "SUCCESS" + + +@pytest.mark.parametrize("method", ("tools/call", "prompts/get")) +def test_only_fixture_name_is_translated_to_advertised_name(method: str) -> None: + request: Final = { + "jsonrpc": "2.0", + "id": 4, + "method": method, + "params": {"name": "test_tool", "_meta": {"progressToken": "keep"}}, + } + changed: Final = json.loads(prefixed_request(json.dumps(request).encode(), "official")) + assert changed == {**request, "params": {"name": "official-test_tool", "_meta": {"progressToken": "keep"}}} + assert "arguments" not in changed["params"] + + +@pytest.mark.parametrize( + "body", + ( + b"malformed JSON", + b"[]", + b'{"method":"initialize","params":{"protocolVersion":"2025-03-26"}}', + b'{"method":"tools/call","params":{"name":17}}', + ), +) +def test_name_integration_preserves_other_requests(body: bytes) -> None: + assert prefixed_request(body, "official") == body + + +@pytest.mark.parametrize( + "scenario,identities", + ( + ("server-initialize", ("server-initialize", "server-session-id-visible-ascii", "wire-schema-valid")), + ("tools-list", ("tools-list", "tools-name-format", "wire-schema-valid")), + ("tools-call-image", ("tools-call-image", "wire-schema-valid")), + ), +) +def test_missing_secondary_checks_cannot_report_complete_conformance( + tmp_path: Path, scenario: str, identities: tuple[str, ...] +) -> None: + report: Final = tmp_path / "checks.json" + for missing in identities: + report.write_text( + json.dumps([{"id": identity, "status": "SUCCESS"} for identity in identities if identity != missing]) + ) + with pytest.raises(AssertionError, match="conformance"): + read_checks(tmp_path, scenario) + report.write_text(json.dumps([{"id": identity, "status": "SUCCESS"} for identity in identities])) + assert len(read_checks(tmp_path, scenario)) == len(identities)