Test: HTTP lifespan fires once per process across sessions (#4470)

tests/server/http/test_lifespan_once_per_process.py

@@ -0,0 +1,67 @@
+"""Regression test for driving the FastMCP lifespan through the SDK session manager.
+
+Before FastMCP handed its lifespan to the SDK lowlevel Server (PR #4446), the
+SDK v1 lifespan was effectively session-scoped and FastMCP worked around it by
+driving its own ``_lifespan_manager`` beside the session manager. The SDK v2
+``StreamableHTTPSessionManager`` now enters ``app.lifespan(app)`` exactly once
+for the manager's lifetime, so the user lifespan must fire once per process and
+persist across multiple HTTP client sessions -- not once per session.
+"""
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Any
+
+from fastmcp import Client, FastMCP
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.utilities.tests import run_server_async
+
+
+async def test_http_user_lifespan_fires_once_across_sessions():
+    """The user lifespan must be entered exactly once for the server process,
+    even when several independent HTTP client sessions connect and disconnect.
+    """
+    enter_count = 0
+    exit_count = 0
+
+    @asynccontextmanager
+    async def counting_lifespan(mcp: FastMCP) -> AsyncIterator[dict[str, Any]]:
+        nonlocal enter_count, exit_count
+        enter_count += 1
+        try:
+            yield {"initialized": True}
+        finally:
+            exit_count += 1
+
+    server = FastMCP("LifespanOnceServer", lifespan=counting_lifespan)
+
+    @server.tool
+    def ping() -> str:
+        return "pong"
+
+    async with run_server_async(server, transport="http") as mcp_url:
+        # `run_server_async` yields a URL that already includes the `/mcp` path.
+        # Three separate, sequential client sessions against the same process.
+        for _ in range(3):
+            async with Client(StreamableHttpTransport(mcp_url)) as client:
+                result = await client.call_tool("ping", {})
+                assert result.data == "pong"
+            # The lifespan must not have exited when a session closed -- it is
+            # owned by the session manager for the whole process lifetime.
+            assert enter_count == 1
+            assert exit_count == 0
+
+        # Overlapping sessions must also observe a single, still-open lifespan.
+        async with (
+            Client(StreamableHttpTransport(mcp_url)) as c1,
+            Client(StreamableHttpTransport(mcp_url)) as c2,
+        ):
+            assert (await c1.call_tool("ping", {})).data == "pong"
+            assert (await c2.call_tool("ping", {})).data == "pong"
+            assert enter_count == 1
+            assert exit_count == 0
+
+    # After the server process task is torn down, the lifespan has exited once.
+    # (A spurious re-entry during teardown would re-exit, so this also guards
+    # that the lifespan was entered exactly once.)
+    assert exit_count == 1