from __future__ import annotations import pytest from fastapi.testclient import TestClient from mcp import ClientSession pytestmark = pytest.mark.smoke class TestMcpHttpEndpoints: """Covers smoke-level HTTP checks for mounted MCP runtime endpoints.""" class TestHealthz: """Covers health endpoint smoke behavior.""" def test_returns_ok_payload(self, client: TestClient) -> None: """Ensures GET /healthz responds with a healthy status payload.""" response = client.get("/healthz") assert response.status_code == 200 assert response.json() == {"status": "ok"} class TestDocsRoute: """Covers static docs route smoke behavior.""" def test_serves_docs_entrypoint(self, client: TestClient) -> None: """Ensures GET /docs returns the docs site entrypoint response.""" response = client.get("/docs") assert response.status_code == 200 assert "text/html" in response.headers["content-type"] class TestMcpRoute: """Covers MCP transport endpoint smoke behavior.""" @pytest.mark.asyncio async def test_rejects_get_stream_without_support( self, client: TestClient, mcp_session: ClientSession, ) -> None: """Ensures GET /mcp returns method not allowed for current transport mode.""" response = client.get( "/mcp", headers={"Accept": "text/event-stream"}, ) # Keep the SDK-backed session in use for this route smoke lane. await mcp_session.list_tools() assert response.status_code == 405 @pytest.mark.asyncio async def test_accepts_initialize_jsonrpc_request( self, mcp_session_uninitialized: ClientSession, ) -> None: """Ensures POST /mcp accepts an initialize JSON-RPC request.""" initialize_result = await mcp_session_uninitialized.initialize() assert initialize_result.protocolVersion assert initialize_result.serverInfo.name