From 9e277d1dd9da68311c6c6e88f42240752c280a24 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Tue, 10 Mar 2026 14:35:15 +0000 Subject: [PATCH 1/3] test: convert WebSocket tests to in-memory transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_ws.py used the subprocess + TCP port pattern that races under pytest-xdist: a worker allocates a port with socket.bind(0), releases it, then spawns a uvicorn subprocess hoping to rebind. Between release and rebind, another worker can claim the port, causing the WS client to connect to an unrelated server (observed: HTTP 403 Forbidden on the WebSocket upgrade). Three of the four tests here verify transport-agnostic MCP semantics (read_resource happy path, MCPError propagation, session recovery after client-side timeout). These now use the in-memory Client transport — no network, no subprocess, no race. The fourth test (test_ws_client_basic_connection) is kept as a smoke test running the real WS stack end-to-end. It uses a new run_uvicorn_in_thread helper that binds port=0 atomically and reads the actual port back from the server's socket — the OS holds the port from bind to shutdown, eliminating the race window entirely. This test alone provides 100% coverage of src/mcp/client/websocket.py. Also removed dead handler code (list_tools/call_tool were never exercised) and the no-longer-needed pragma: no cover annotations on the read_resource handler (it now runs in-process). --- tests/shared/test_ws.py | 229 ++++++++++++---------------------------- tests/test_helpers.py | 66 ++++++++++++ 2 files changed, 136 insertions(+), 159 deletions(-) diff --git a/tests/shared/test_ws.py b/tests/shared/test_ws.py index 9addb661d..fd24a15ae 100644 --- a/tests/shared/test_ws.py +++ b/tests/shared/test_ws.py @@ -1,210 +1,121 @@ -import multiprocessing -import socket -from collections.abc import AsyncGenerator, Generator +"""Tests for the WebSocket transport. + +The smoke test (``test_ws_client_basic_connection``) runs the full WS stack +end-to-end over a real TCP connection and is what provides coverage of +``src/mcp/client/websocket.py``. + +The remaining tests verify transport-agnostic MCP semantics (error +propagation, client-side timeouts) and use the in-memory ``Client`` transport +to avoid the cost and flakiness of real network servers. +""" + +from collections.abc import Generator from urllib.parse import urlparse import anyio import pytest -import uvicorn from starlette.applications import Starlette from starlette.routing import WebSocketRoute from starlette.websockets import WebSocket -from mcp import MCPError +from mcp import Client, MCPError from mcp.client.session import ClientSession from mcp.client.websocket import websocket_client from mcp.server import Server, ServerRequestContext from mcp.server.websocket import websocket_server from mcp.types import ( - CallToolRequestParams, - CallToolResult, EmptyResult, InitializeResult, - ListToolsResult, - PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResult, - TextContent, TextResourceContents, - Tool, ) -from tests.test_helpers import wait_for_server +from tests.test_helpers import run_uvicorn_in_thread SERVER_NAME = "test_server_for_WS" +pytestmark = pytest.mark.anyio -@pytest.fixture -def server_port() -> int: - with socket.socket() as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] +# --- WebSocket transport smoke test (real TCP) ------------------------------- -@pytest.fixture -def server_url(server_port: int) -> str: - return f"ws://127.0.0.1:{server_port}" +def make_server_app() -> Starlette: + srv = Server(SERVER_NAME) -async def handle_read_resource( # pragma: no cover - ctx: ServerRequestContext, params: ReadResourceRequestParams -) -> ReadResourceResult: - parsed = urlparse(str(params.uri)) - if parsed.scheme == "foobar": - return ReadResourceResult( - contents=[TextResourceContents(uri=str(params.uri), text=f"Read {parsed.netloc}", mime_type="text/plain")] - ) - elif parsed.scheme == "slow": - await anyio.sleep(2.0) - return ReadResourceResult( - contents=[ - TextResourceContents( - uri=str(params.uri), text=f"Slow response from {parsed.netloc}", mime_type="text/plain" - ) - ] - ) - raise MCPError(code=404, message="OOPS! no resource with that URI was found") - - -async def handle_list_tools( # pragma: no cover - ctx: ServerRequestContext, params: PaginatedRequestParams | None -) -> ListToolsResult: - return ListToolsResult( - tools=[ - Tool( - name="test_tool", - description="A test tool", - input_schema={"type": "object", "properties": {}}, - ) - ] - ) - - -async def handle_call_tool( # pragma: no cover - ctx: ServerRequestContext, params: CallToolRequestParams -) -> CallToolResult: - return CallToolResult(content=[TextContent(type="text", text=f"Called {params.name}")]) - - -def _create_server() -> Server: # pragma: no cover - return Server( - SERVER_NAME, - on_read_resource=handle_read_resource, - on_list_tools=handle_list_tools, - on_call_tool=handle_call_tool, - ) - - -# Test fixtures -def make_server_app() -> Starlette: # pragma: no cover - """Create test Starlette app with WebSocket transport""" - server = _create_server() - - async def handle_ws(websocket: WebSocket): + async def handle_ws(websocket: WebSocket) -> None: async with websocket_server(websocket.scope, websocket.receive, websocket.send) as streams: - await server.run(streams[0], streams[1], server.create_initialization_options()) - - app = Starlette(routes=[WebSocketRoute("/ws", endpoint=handle_ws)]) - return app - + await srv.run(streams[0], streams[1], srv.create_initialization_options()) -def run_server(server_port: int) -> None: # pragma: no cover - app = make_server_app() - server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error")) - print(f"starting server on {server_port}") - server.run() + return Starlette(routes=[WebSocketRoute("/ws", endpoint=handle_ws)]) -@pytest.fixture() -def server(server_port: int) -> Generator[None, None, None]: - proc = multiprocessing.Process(target=run_server, kwargs={"server_port": server_port}, daemon=True) - print("starting process") - proc.start() - - # Wait for server to be running - print("waiting for server to start") - wait_for_server(server_port) - - yield - - print("killing server") - # Signal the server to stop - proc.kill() - proc.join(timeout=2) - if proc.is_alive(): # pragma: no cover - print("server process failed to terminate") +@pytest.fixture +def ws_server_url() -> Generator[str, None, None]: + with run_uvicorn_in_thread(make_server_app()) as base_url: + yield base_url.replace("http://", "ws://") + "/ws" -@pytest.fixture() -async def initialized_ws_client_session(server: None, server_url: str) -> AsyncGenerator[ClientSession, None]: - """Create and initialize a WebSocket client session""" - async with websocket_client(server_url + "/ws") as streams: +async def test_ws_client_basic_connection(ws_server_url: str) -> None: + async with websocket_client(ws_server_url) as streams: async with ClientSession(*streams) as session: - # Test initialization result = await session.initialize() assert isinstance(result, InitializeResult) assert result.server_info.name == SERVER_NAME - # Test ping ping_result = await session.send_ping() assert isinstance(ping_result, EmptyResult) - yield session +# --- In-memory tests (transport-agnostic MCP semantics) ---------------------- -# Tests -@pytest.mark.anyio -async def test_ws_client_basic_connection(server: None, server_url: str) -> None: - """Test the WebSocket connection establishment""" - async with websocket_client(server_url + "/ws") as streams: - async with ClientSession(*streams) as session: - # Test initialization - result = await session.initialize() - assert isinstance(result, InitializeResult) - assert result.server_info.name == SERVER_NAME - # Test ping - ping_result = await session.send_ping() - assert isinstance(ping_result, EmptyResult) +async def handle_read_resource(ctx: ServerRequestContext, params: ReadResourceRequestParams) -> ReadResourceResult: + parsed = urlparse(str(params.uri)) + if parsed.scheme == "foobar": + return ReadResourceResult( + contents=[TextResourceContents(uri=str(params.uri), text=f"Read {parsed.netloc}", mime_type="text/plain")] + ) + elif parsed.scheme == "slow": + # Block indefinitely so the client-side fail_after() fires; the pending + # server task is cancelled when the Client context manager exits. + await anyio.sleep_forever() + raise MCPError(code=404, message="OOPS! no resource with that URI was found") + + +@pytest.fixture +def server() -> Server: + return Server(SERVER_NAME, on_read_resource=handle_read_resource) -@pytest.mark.anyio -async def test_ws_client_happy_request_and_response( - initialized_ws_client_session: ClientSession, -) -> None: - """Test a successful request and response via WebSocket""" - result = await initialized_ws_client_session.read_resource("foobar://example") - assert isinstance(result, ReadResourceResult) - assert isinstance(result.contents, list) - assert len(result.contents) > 0 - assert isinstance(result.contents[0], TextResourceContents) - assert result.contents[0].text == "Read example" - - -@pytest.mark.anyio -async def test_ws_client_exception_handling( - initialized_ws_client_session: ClientSession, -) -> None: - """Test exception handling in WebSocket communication""" - with pytest.raises(MCPError) as exc_info: - await initialized_ws_client_session.read_resource("unknown://example") - assert exc_info.value.error.code == 404 - - -@pytest.mark.anyio -async def test_ws_client_timeout( - initialized_ws_client_session: ClientSession, -) -> None: - """Test timeout handling in WebSocket communication""" - # Set a very short timeout to trigger a timeout exception - with pytest.raises(TimeoutError): - with anyio.fail_after(0.1): # 100ms timeout - await initialized_ws_client_session.read_resource("slow://example") - - # Now test that we can still use the session after a timeout - with anyio.fail_after(5): # Longer timeout to allow completion - result = await initialized_ws_client_session.read_resource("foobar://example") +async def test_ws_client_happy_request_and_response(server: Server) -> None: + async with Client(server) as client: + result = await client.read_resource("foobar://example") assert isinstance(result, ReadResourceResult) assert isinstance(result.contents, list) assert len(result.contents) > 0 assert isinstance(result.contents[0], TextResourceContents) assert result.contents[0].text == "Read example" + + +async def test_ws_client_exception_handling(server: Server) -> None: + async with Client(server) as client: + with pytest.raises(MCPError) as exc_info: + await client.read_resource("unknown://example") + assert exc_info.value.error.code == 404 + + +async def test_ws_client_timeout(server: Server) -> None: + async with Client(server) as client: + with pytest.raises(TimeoutError): + with anyio.fail_after(0.1): + await client.read_resource("slow://example") + + # Session remains usable after a client-side timeout abandons a request. + with anyio.fail_after(5): + result = await client.read_resource("foobar://example") + assert isinstance(result, ReadResourceResult) + assert isinstance(result.contents, list) + assert len(result.contents) > 0 + assert isinstance(result.contents[0], TextResourceContents) + assert result.contents[0].text == "Read example" diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 5c04c269f..bcc7e3edf 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -1,7 +1,73 @@ """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 + +# How long to wait for the uvicorn server thread to reach `started`. +# Generous to absorb CI scheduling delays — actual startup is typically <100ms. +_SERVER_START_TIMEOUT_S = 20.0 +_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 with an ephemeral port. + + This eliminates the TOCTOU race that occurs when a test picks a free port + with ``socket.bind((host, 0))``, releases it, then starts a server hoping + to rebind the same port — between release and rebind, another pytest-xdist + worker may claim it, causing connection errors or cross-test contamination. + + With ``port=0``, the OS atomically assigns a free port at bind time; the + server holds it from that moment until shutdown. We read the actual port + back from uvicorn's bound socket after startup completes. + + Args: + app: ASGI application to serve. + **config_kwargs: Additional keyword arguments for :class:`uvicorn.Config` + (e.g. ``log_level``, ``limit_concurrency``). ``host`` defaults to + ``127.0.0.1`` and ``port`` is forced to 0. + + Yields: + The base URL of the running server, e.g. ``http://127.0.0.1:54321``. + + Raises: + TimeoutError: If the server does not start within 20 seconds. + RuntimeError: If the server thread dies during startup. + """ + config_kwargs.setdefault("host", "127.0.0.1") + config_kwargs.setdefault("log_level", "error") + config = uvicorn.Config(app=app, port=0, **config_kwargs) + server = uvicorn.Server(config=config) + + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + + # uvicorn sets `server.started = True` at the end of `Server.startup()`, + # after sockets are bound and the lifespan startup phase has completed. + start = time.monotonic() + while not server.started: + if time.monotonic() - start > _SERVER_START_TIMEOUT_S: # pragma: no cover + raise TimeoutError(f"uvicorn server failed to start within {_SERVER_START_TIMEOUT_S}s") + if not thread.is_alive(): # pragma: no cover + raise RuntimeError("uvicorn server thread exited during startup") + time.sleep(0.001) + + # server.servers[0] is the asyncio.Server; its bound socket has the real port + port = server.servers[0].sockets[0].getsockname()[1] + host = config.host + + 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: From cdb087535e4bf854cd58861746e7bc47f90616fb Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Fri, 13 Mar 2026 10:57:55 +0000 Subject: [PATCH 2/3] test: reduce test_ws.py to just the WS smoke test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The three tests converted to in-memory transport in the previous commit weren't testing WebSocket behavior anymore — they were sitting in test_ws.py with misleading names. - test_ws_client_happy_request_and_response: deleted. Duplicates tests/client/test_client.py::test_read_resource. - test_ws_client_timeout: deleted. tests/issues/test_88_random_error.py covers the same session-recovery-after-timeout scenario more thoroughly (uses read_timeout_seconds and an anyio.Event to release the slow handler cleanly). - test_ws_client_exception_handling: moved to test_client.py as test_read_resource_error_propagates. This was the only unique behavior — nothing else asserts that a handler-raised MCPError reaches the client with its error code intact. test_ws.py now contains only what it says on the tin. --- tests/client/test_client.py | 17 +++++++- tests/shared/test_ws.py | 86 ++++--------------------------------- 2 files changed, 24 insertions(+), 79 deletions(-) diff --git a/tests/client/test_client.py b/tests/client/test_client.py index 45300063a..3bdd30570 100644 --- a/tests/client/test_client.py +++ b/tests/client/test_client.py @@ -8,7 +8,7 @@ import pytest from inline_snapshot import snapshot -from mcp import types +from mcp import MCPError, types from mcp.client._memory import InMemoryTransport from mcp.client.client import Client from mcp.server import Server, ServerRequestContext @@ -175,6 +175,21 @@ async def test_read_resource(app: MCPServer): ) +async def test_read_resource_error_propagates(): + """MCPError raised by a server handler propagates to the client with its code intact.""" + + async def handle_read_resource( + ctx: ServerRequestContext, params: types.ReadResourceRequestParams + ) -> ReadResourceResult: + raise MCPError(code=404, message="no resource with that URI was found") + + server = Server("test", on_read_resource=handle_read_resource) + async with Client(server) as client: + with pytest.raises(MCPError) as exc_info: + await client.read_resource("unknown://example") + assert exc_info.value.error.code == 404 + + async def test_get_prompt(app: MCPServer): """Test getting a prompt.""" async with Client(app) as client: diff --git a/tests/shared/test_ws.py b/tests/shared/test_ws.py index fd24a15ae..ce2da3b24 100644 --- a/tests/shared/test_ws.py +++ b/tests/shared/test_ws.py @@ -1,44 +1,27 @@ -"""Tests for the WebSocket transport. +"""Smoke test for the WebSocket transport. -The smoke test (``test_ws_client_basic_connection``) runs the full WS stack -end-to-end over a real TCP connection and is what provides coverage of -``src/mcp/client/websocket.py``. - -The remaining tests verify transport-agnostic MCP semantics (error -propagation, client-side timeouts) and use the in-memory ``Client`` transport -to avoid the cost and flakiness of real network servers. +Runs the full WS stack end-to-end over a real TCP connection to provide +coverage of ``src/mcp/client/websocket.py``. MCP semantics (error +propagation, timeouts, etc.) are transport-agnostic and are covered in +``tests/client/test_client.py`` and ``tests/issues/test_88_random_error.py``. """ from collections.abc import Generator -from urllib.parse import urlparse -import anyio import pytest from starlette.applications import Starlette from starlette.routing import WebSocketRoute from starlette.websockets import WebSocket -from mcp import Client, MCPError from mcp.client.session import ClientSession from mcp.client.websocket import websocket_client -from mcp.server import Server, ServerRequestContext +from mcp.server import Server from mcp.server.websocket import websocket_server -from mcp.types import ( - EmptyResult, - InitializeResult, - ReadResourceRequestParams, - ReadResourceResult, - TextResourceContents, -) +from mcp.types import EmptyResult, InitializeResult from tests.test_helpers import run_uvicorn_in_thread SERVER_NAME = "test_server_for_WS" -pytestmark = pytest.mark.anyio - - -# --- WebSocket transport smoke test (real TCP) ------------------------------- - def make_server_app() -> Starlette: srv = Server(SERVER_NAME) @@ -56,6 +39,7 @@ def ws_server_url() -> Generator[str, None, None]: yield base_url.replace("http://", "ws://") + "/ws" +@pytest.mark.anyio async def test_ws_client_basic_connection(ws_server_url: str) -> None: async with websocket_client(ws_server_url) as streams: async with ClientSession(*streams) as session: @@ -65,57 +49,3 @@ async def test_ws_client_basic_connection(ws_server_url: str) -> None: ping_result = await session.send_ping() assert isinstance(ping_result, EmptyResult) - - -# --- In-memory tests (transport-agnostic MCP semantics) ---------------------- - - -async def handle_read_resource(ctx: ServerRequestContext, params: ReadResourceRequestParams) -> ReadResourceResult: - parsed = urlparse(str(params.uri)) - if parsed.scheme == "foobar": - return ReadResourceResult( - contents=[TextResourceContents(uri=str(params.uri), text=f"Read {parsed.netloc}", mime_type="text/plain")] - ) - elif parsed.scheme == "slow": - # Block indefinitely so the client-side fail_after() fires; the pending - # server task is cancelled when the Client context manager exits. - await anyio.sleep_forever() - raise MCPError(code=404, message="OOPS! no resource with that URI was found") - - -@pytest.fixture -def server() -> Server: - return Server(SERVER_NAME, on_read_resource=handle_read_resource) - - -async def test_ws_client_happy_request_and_response(server: Server) -> None: - async with Client(server) as client: - result = await client.read_resource("foobar://example") - assert isinstance(result, ReadResourceResult) - assert isinstance(result.contents, list) - assert len(result.contents) > 0 - assert isinstance(result.contents[0], TextResourceContents) - assert result.contents[0].text == "Read example" - - -async def test_ws_client_exception_handling(server: Server) -> None: - async with Client(server) as client: - with pytest.raises(MCPError) as exc_info: - await client.read_resource("unknown://example") - assert exc_info.value.error.code == 404 - - -async def test_ws_client_timeout(server: Server) -> None: - async with Client(server) as client: - with pytest.raises(TimeoutError): - with anyio.fail_after(0.1): - await client.read_resource("slow://example") - - # Session remains usable after a client-side timeout abandons a request. - with anyio.fail_after(5): - result = await client.read_resource("foobar://example") - assert isinstance(result, ReadResourceResult) - assert isinstance(result.contents, list) - assert len(result.contents) > 0 - assert isinstance(result.contents[0], TextResourceContents) - assert result.contents[0].text == "Read example" From 97ba3fab54a60a1afe2be7d549d76a06adc5f0d1 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Fri, 13 Mar 2026 11:05:27 +0000 Subject: [PATCH 3/3] test: pre-bind socket in run_uvicorn_in_thread, drop polling loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous version polled server.started with time.sleep(0.001) until uvicorn finished binding. We were waiting for uvicorn to tell us the port — but we can just bind the socket ourselves and hand it to uvicorn via server.run(sockets=[sock]). Once sock.listen() returns, the kernel queues incoming connections (up to the backlog). If a client connects before uvicorn's event loop reaches accept(), the connection sits in the accept queue and is picked up as soon as uvicorn is ready. The kernel is the synchronizer — no cross-thread flag needed. The port is now known before the thread starts. No polling, no sleep, no wait at all. --- tests/test_helpers.py | 56 ++++++++++++++++--------------------------- 1 file changed, 20 insertions(+), 36 deletions(-) diff --git a/tests/test_helpers.py b/tests/test_helpers.py index bcc7e3edf..fc300bc88 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -9,60 +9,44 @@ import uvicorn -# How long to wait for the uvicorn server thread to reach `started`. -# Generous to absorb CI scheduling delays — actual startup is typically <100ms. -_SERVER_START_TIMEOUT_S = 20.0 _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 with an ephemeral port. + """Run a uvicorn server in a background thread on an ephemeral port. - This eliminates the TOCTOU race that occurs when a test picks a free port - with ``socket.bind((host, 0))``, releases it, then starts a server hoping - to rebind the same port — between release and rebind, another pytest-xdist - worker may claim it, causing connection errors or cross-test contamination. + 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. - With ``port=0``, the OS atomically assigns a free port at bind time; the - server holds it from that moment until shutdown. We read the actual port - back from uvicorn's bound socket after startup completes. + 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``, ``limit_concurrency``). ``host`` defaults to - ``127.0.0.1`` and ``port`` is forced to 0. + (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``. - - Raises: - TimeoutError: If the server does not start within 20 seconds. - RuntimeError: If the server thread dies during startup. """ - config_kwargs.setdefault("host", "127.0.0.1") + 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") - config = uvicorn.Config(app=app, port=0, **config_kwargs) - server = uvicorn.Server(config=config) + server = uvicorn.Server(config=uvicorn.Config(app=app, **config_kwargs)) - thread = threading.Thread(target=server.run, daemon=True) + thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True) thread.start() - - # uvicorn sets `server.started = True` at the end of `Server.startup()`, - # after sockets are bound and the lifespan startup phase has completed. - start = time.monotonic() - while not server.started: - if time.monotonic() - start > _SERVER_START_TIMEOUT_S: # pragma: no cover - raise TimeoutError(f"uvicorn server failed to start within {_SERVER_START_TIMEOUT_S}s") - if not thread.is_alive(): # pragma: no cover - raise RuntimeError("uvicorn server thread exited during startup") - time.sleep(0.001) - - # server.servers[0] is the asyncio.Server; its bound socket has the real port - port = server.servers[0].sockets[0].getsockname()[1] - host = config.host - try: yield f"http://{host}:{port}" finally: