add unit tests

Author: tonyxwz

tests/server/fastmcp/test_server.py

@@ -1283,21 +1283,40 @@ def prompt_fn(name: str) -> str:
             with pytest.raises(McpError, match="Missing required arguments"):
                 await client.get_prompt("prompt_fn")
 
+    def test_single_tenant_default_false(self):
+        """Test that single_tenant defaults to False."""
+        mcp = FastMCP("test")
+        assert mcp.settings.single_tenant is False
+
+    def test_single_tenant_can_be_set_true(self):
+        """Test that single_tenant can be set to True."""
+        mcp = FastMCP("test", single_tenant=True)
+        assert mcp.settings.single_tenant is True
+
+    def test_single_tenant_passed_to_session_manager(self):
+        """Test that single_tenant is passed to StreamableHTTPSessionManager."""
+        mcp = FastMCP("test", single_tenant=True)
+
+        # Access the session manager to trigger its creation
+        # We need to call streamable_http_app() first to initialize the session manager
+        mcp.streamable_http_app()
+        session_manager = mcp.session_manager
+        assert session_manager.single_tenant is True
+
+    def test_streamable_http_no_redirect(self) -> None:
+        """Test that streamable HTTP routes are correctly configured."""
+        mcp = FastMCP()
+        app = mcp.streamable_http_app()
+
+        # Find routes by type - streamable_http_app creates Route objects, not Mount objects
+        streamable_routes = [
+            r
+            for r in app.routes
+            if isinstance(r, Route) and hasattr(r, "path") and r.path == mcp.settings.streamable_http_path
+        ]
 
-def test_streamable_http_no_redirect() -> None:
-    """Test that streamable HTTP routes are correctly configured."""
-    mcp = FastMCP()
-    app = mcp.streamable_http_app()
-
-    # Find routes by type - streamable_http_app creates Route objects, not Mount objects
-    streamable_routes = [
-        r
-        for r in app.routes
-        if isinstance(r, Route) and hasattr(r, "path") and r.path == mcp.settings.streamable_http_path
-    ]
-
-    # Verify routes exist
-    assert len(streamable_routes) == 1, "Should have one streamable route"
+        # Verify routes exist
+        assert len(streamable_routes) == 1, "Should have one streamable route"
 
-    # Verify path values
-    assert streamable_routes[0].path == "/mcp", "Streamable route path should be /mcp"
+        # Verify path values
+        assert streamable_routes[0].path == "/mcp", "Streamable route path should be /mcp"

tests/server/test_streamable_http_manager.py

@@ -5,14 +5,73 @@
 
 import anyio
 import pytest
-from starlette.types import Message
+from starlette.requests import Request
+from starlette.types import Message, Scope
 
 from mcp.server import streamable_http_manager
 from mcp.server.lowlevel import Server
 from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, StreamableHTTPServerTransport
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 
 
+def test_single_tenant_validation():
+    """Test that single_tenant=True with stateless=True raises ValueError."""
+    app = Server("test")
+
+    with pytest.raises(ValueError, match="A single-tenant server must stateful"):
+        StreamableHTTPSessionManager(app=app, single_tenant=True, stateless=True)
+
+
+def test_single_tenant_default_false():
+    """Test that single_tenant defaults to False."""
+    app = Server("test")
+    manager = StreamableHTTPSessionManager(app=app)
+    assert manager.single_tenant is False
+
+
+def test_single_tenant_can_be_set_true():
+    """Test that single_tenant can be set to True."""
+    app = Server("test")
+    manager = StreamableHTTPSessionManager(app=app, single_tenant=True)
+    assert manager.single_tenant is True
+
+
+@pytest.mark.anyio
+async def test_single_tenant_reuses_existing_session():
+    """Test that single_tenant mode reuses existing session."""
+    app = Server("test")
+    manager = StreamableHTTPSessionManager(app=app, single_tenant=True)
+
+    mock_mcp_run = AsyncMock(return_value=None)
+    # This will be called by StreamableHTTPSessionManager's run_server -> self.app.run
+    app.run = mock_mcp_run
+
+    # Manually add a session to simulate existing session
+    existing_session_id = "existing-session-123"
+    mock_transport = AsyncMock()
+    manager._server_instances[existing_session_id] = mock_transport
+
+    # Create a request with different session ID
+    request_mcp_session_id = "different-session-id"
+    scope: Scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "headers": [
+            (b"content-type", b"application/json"),
+            (MCP_SESSION_ID_HEADER.encode("latin-1"), request_mcp_session_id.encode("latin-1")),
+        ],
+    }
+
+    async with manager.run():
+        await manager.handle_request(scope, AsyncMock(), AsyncMock())
+        headers = Request(scope).headers
+        modified_session_id = headers[MCP_SESSION_ID_HEADER]
+
+        assert modified_session_id == existing_session_id
+        assert len(manager._server_instances) == 1
+
+
 @pytest.mark.anyio
 async def test_run_can_only_be_called_once():
     """Test that run() can only be called once per instance."""