test: fix WS test port race; narrow to single smoke test covering both transport ends (#2267)

tests/test_helpers.py

@@ -1,7 +1,61 @@
 """Common test utilities for MCP server tests."""
 
 import socket
+import threading
 import time
+from collections.abc import Generator
+from contextlib import contextmanager
+from typing import Any
+
+import uvicorn
+
+_SERVER_SHUTDOWN_TIMEOUT_S = 5.0
+
+
+@contextmanager
+def run_uvicorn_in_thread(app: Any, **config_kwargs: Any) -> Generator[str, None, None]:
+    """Run a uvicorn server in a background thread on an ephemeral port.
+
+    The socket is bound and put into listening state *before* the thread
+    starts, so the port is known immediately with no wait. The kernel's
+    listen queue buffers any connections that arrive before uvicorn's event
+    loop reaches ``accept()``, so callers can connect as soon as this
+    function yields — no polling, no sleeps, no startup race.
+
+    This also avoids the TOCTOU race of the old pick-a-port-then-rebind
+    pattern: the socket passed here is the one uvicorn serves on, with no
+    gap where another pytest-xdist worker could claim it.
+
+    Args:
+        app: ASGI application to serve.
+        **config_kwargs: Additional keyword arguments for :class:`uvicorn.Config`
+            (e.g. ``log_level``). ``host``/``port`` are ignored since the
+            socket is pre-bound.
+
+    Yields:
+        The base URL of the running server, e.g. ``http://127.0.0.1:54321``.
+    """
+    host = "127.0.0.1"
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind((host, 0))
+    sock.listen()
+    port = sock.getsockname()[1]
+
+    config_kwargs.setdefault("log_level", "error")
+    # Uvicorn's interface autodetection calls asyncio.iscoroutinefunction,
+    # which Python 3.14 deprecates. Under filterwarnings=error this crashes
+    # the server thread silently. Starlette is asgi3; skip the autodetect.
+    config_kwargs.setdefault("interface", "asgi3")
+    server = uvicorn.Server(config=uvicorn.Config(app=app, **config_kwargs))
+
+    thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
+    thread.start()
+    try:
+        yield f"http://{host}:{port}"
+    finally:
+        server.should_exit = True
+        thread.join(timeout=_SERVER_SHUTDOWN_TIMEOUT_S)
 
 
 def wait_for_server(port: int, timeout: float = 20.0) -> None: