Start a mock SSE server in a background thread and yield its base URL.
(free_port: int)
| 98 | |
| 99 | @pytest.fixture() |
| 100 | def mock_server(free_port: int) -> Generator[str, None, None]: |
| 101 | """Start a mock SSE server in a background thread and yield its base URL.""" |
| 102 | port = free_port |
| 103 | app = aiohttp.web.Application() |
| 104 | app.router.add_post("/v1/chat/completions", mock_chat_handler) |
| 105 | |
| 106 | loop = asyncio.new_event_loop() |
| 107 | runner = aiohttp.web.AppRunner(app) |
| 108 | loop.run_until_complete(runner.setup()) |
| 109 | |
| 110 | started = threading.Event() |
| 111 | thread = threading.Thread( |
| 112 | target=_run_server, args=(loop, runner, port, started), daemon=True |
| 113 | ) |
| 114 | thread.start() |
| 115 | assert started.wait(timeout=5), "Mock server failed to start" |
| 116 | |
| 117 | base_url = f"http://127.0.0.1:{port}" |
| 118 | yield base_url |
| 119 | |
| 120 | loop.call_soon_threadsafe(loop.stop) |
| 121 | thread.join(timeout=5) |
| 122 | loop.run_until_complete(runner.cleanup()) |
| 123 | loop.close() |