|
4 | 4 | (server→client and client→server) using the streamable HTTP transport. |
5 | 5 | """ |
6 | 6 |
|
7 | | -import multiprocessing |
8 | | -import socket |
9 | 7 | from collections.abc import AsyncGenerator, Generator |
10 | 8 | from contextlib import asynccontextmanager |
11 | 9 |
|
|
19 | 17 | from mcp.server import Server, ServerRequestContext |
20 | 18 | from mcp.server.streamable_http_manager import StreamableHTTPSessionManager |
21 | 19 | from mcp.types import TextContent, Tool |
22 | | -from tests.test_helpers import wait_for_server |
| 20 | +from tests.test_helpers import run_uvicorn_in_thread |
23 | 21 |
|
24 | 22 | # Test constants with various Unicode characters |
25 | 23 | UNICODE_TEST_STRINGS = { |
|
41 | 39 | } |
42 | 40 |
|
43 | 41 |
|
44 | | -def run_unicode_server(port: int) -> None: # pragma: no cover |
45 | | - """Run the Unicode test server in a separate process.""" |
46 | | - import uvicorn |
47 | | - |
| 42 | +def make_unicode_server_app() -> Starlette: # pragma: no cover |
| 43 | + """Create the Unicode test server.""" |
48 | 44 | # Need to recreate the server setup in this process |
49 | 45 | async def handle_list_tools( |
50 | 46 | ctx: ServerRequestContext, params: types.PaginatedRequestParams | None |
@@ -137,43 +133,14 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: |
137 | 133 | lifespan=lifespan, |
138 | 134 | ) |
139 | 135 |
|
140 | | - # Run the server |
141 | | - config = uvicorn.Config( |
142 | | - app=app, |
143 | | - host="127.0.0.1", |
144 | | - port=port, |
145 | | - log_level="error", |
146 | | - ) |
147 | | - uvicorn_server = uvicorn.Server(config) |
148 | | - uvicorn_server.run() |
149 | | - |
150 | | - |
151 | | -@pytest.fixture |
152 | | -def unicode_server_port() -> int: |
153 | | - """Find an available port for the Unicode test server.""" |
154 | | - with socket.socket() as s: |
155 | | - s.bind(("127.0.0.1", 0)) |
156 | | - return s.getsockname()[1] |
| 136 | + return app |
157 | 137 |
|
158 | 138 |
|
159 | 139 | @pytest.fixture |
160 | | -def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]: |
161 | | - """Start a Unicode test server in a separate process.""" |
162 | | - proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True) |
163 | | - proc.start() |
164 | | - |
165 | | - # Wait for server to be ready |
166 | | - wait_for_server(unicode_server_port) |
167 | | - |
168 | | - try: |
169 | | - yield f"http://127.0.0.1:{unicode_server_port}" |
170 | | - finally: |
171 | | - # Clean up - try graceful termination first |
172 | | - proc.terminate() |
173 | | - proc.join(timeout=2) |
174 | | - if proc.is_alive(): # pragma: no cover |
175 | | - proc.kill() |
176 | | - proc.join(timeout=1) |
| 140 | +def running_unicode_server() -> Generator[str, None, None]: |
| 141 | + """Start a Unicode test server without preselecting a port.""" |
| 142 | + with run_uvicorn_in_thread(make_unicode_server_app()) as url: |
| 143 | + yield url |
177 | 144 |
|
178 | 145 |
|
179 | 146 | @pytest.mark.anyio |
|
0 commit comments