test(mcp): require complete reports payloads and process cleanup

tests/integration/_support/conformance.py

@@ -10,6 +10,7 @@ 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:
@@ -33,9 +34,15 @@ def read_checks(directory: Path, scenario: str) -> tuple[ConformanceCheck, ...]:
     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"
-    assert tuple(check.status for check in checks if check.id == scenario) == ("SUCCESS",), (
-        f"conformance {scenario}: required scenario did not pass: {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")
@@ -82,7 +89,7 @@ def run_scenario(root: Path, url: str, scenario: str, directory: Path) -> tuple[
 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 signal_group, stop_root_process
+    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}"
@@ -109,10 +116,17 @@ def reference_server(root: Path, directory: Path, port: int) -> Iterator["McpPee
             yield McpPeer(url + "/mcp", queue.Queue())
         finally:
             stopped: Final = stop_root_process(process)
-            if not stopped:
+            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)
-                process.wait(timeout=3)
-            assert stopped, "Official reference required forced cleanup"
+                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

tests/integration/mcp/test_mcp_official_conformance.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import os
 import uuid
@@ -10,6 +11,9 @@ 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"))
@@ -24,6 +28,24 @@ def test_official_scenario_through_gateway(gateway: Gateway, tmp_path: Path, unu
         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", "test_simple_text", {})
@@ -92,3 +114,31 @@ def test_stalled_reference_is_killed_and_cannot_report_clean_teardown(tmp_path:
         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)

tests/unit/integration_support/test_conformance.py

@@ -53,9 +53,16 @@ def test_multiple_reports_cannot_supply_a_stale_pass(tmp_path: Path) -> None:
 
 
 def test_passing_scenario_keeps_nonbinding_diagnostics(tmp_path: Path) -> None:
-    (tmp_path / "checks.json").write_text('[{"id":"tools-list","status":"SUCCESS"},{"id":"advisory","status":"INFO"}]')
+    (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 tuple((check.id, check.status) for check in checks) == (("tools-list", "SUCCESS"), ("advisory", "INFO"))
+    assert checks[-1].id == "advisory" and checks[-1].status == "INFO"
 
 
 def test_unrecognized_result_status_cannot_pass(tmp_path: Path) -> None:
@@ -123,3 +130,25 @@ def test_only_fixture_name_is_translated_to_advertised_name(method: str) -> None
 )
 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)