mcp endpoint test scaffold

tests/web/conftest.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import os
+from collections.abc import AsyncIterator
+from collections.abc import Iterator
+
+import pytest
+import pytest_asyncio
+from fastapi.testclient import TestClient
+from httpx import AsyncClient
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+from personal_mcp.web.app import create_app
+
+
+@pytest.fixture
+def client() -> Iterator[TestClient]:
+    """Provides a TestClient bound to a fresh application instance."""
+    with TestClient(create_app()) as test_client:
+        yield test_client
+
+
+@pytest.fixture
+def mcp_endpoint_url() -> str:
+    """Provides the MCP endpoint URL for SDK-based client sessions."""
+    return os.getenv("PERSONAL_MCP_TEST_HTTP_URL", "")
+
+
+@pytest_asyncio.fixture
+async def mcp_http_client() -> AsyncIterator[AsyncClient]:
+    """Provides a reusable async HTTP client for MCP SDK transports."""
+    async with AsyncClient(timeout=10.0) as http_client:
+        yield http_client
+
+
+@pytest_asyncio.fixture
+async def mcp_session_uninitialized(
+    mcp_endpoint_url: str,
+    mcp_http_client: AsyncClient,
+) -> AsyncIterator[ClientSession]:
+    """Provides a connected MCP SDK session before initialize is called."""
+    if not mcp_endpoint_url:
+        pytest.skip("Set PERSONAL_MCP_TEST_HTTP_URL to run SDK-backed MCP endpoint tests.")
+
+    async with (
+        streamable_http_client(
+            mcp_endpoint_url,
+            http_client=mcp_http_client,
+        ) as (read_stream, write_stream, _),
+        ClientSession(read_stream, write_stream) as session,
+    ):
+        yield session
+
+
+@pytest_asyncio.fixture
+async def mcp_session(
+    mcp_session_uninitialized: ClientSession,
+) -> AsyncIterator[ClientSession]:
+    """Provides an initialized MCP SDK session ready for endpoint calls."""
+    await mcp_session_uninitialized.initialize()
+    yield mcp_session_uninitialized

tests/web/test_mcp_endpoints.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+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."""
+
+    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."""
+
+    class TestMcpRoute:
+        """Covers MCP transport endpoint smoke behavior."""
+
+        @pytest.mark.asyncio
+        async def test_rejects_get_stream_without_support(self, mcp_session: Any) -> None:
+            """Ensures GET /mcp returns method not allowed for current transport mode."""
+
+        @pytest.mark.asyncio
+        async def test_accepts_initialize_jsonrpc_request(self, mcp_session_uninitialized: Any) -> None:
+            """Ensures POST /mcp accepts an initialize JSON-RPC request."""