Run a server in a background thread and yield it.
(server: TestServer)
| 540 | |
| 541 | |
| 542 | def serve_in_thread(server: TestServer) -> Iterator[TestServer]: |
| 543 | """Run a server in a background thread and yield it.""" |
| 544 | thread = threading.Thread(target=server.run, daemon=True) |
| 545 | thread.start() |
| 546 | try: |
| 547 | # Bound the startup wait: when uvicorn fails to bind (e.g. port collision under xdist), the worker thread |
| 548 | # exits without ever setting server.started, so an unbounded loop would hang until pytest-timeout kills |
| 549 | # the suite 30 minutes later. |
| 550 | deadline = time.monotonic() + 30 |
| 551 | while not server.started: |
| 552 | if not thread.is_alive(): |
| 553 | raise RuntimeError('Test server thread exited before becoming ready (likely a bind failure).') |
| 554 | if time.monotonic() > deadline: |
| 555 | raise RuntimeError('Test server did not become ready within 30s.') |
| 556 | time.sleep(1e-3) |
| 557 | yield server |
| 558 | finally: |
| 559 | server.should_exit = True |
| 560 | thread.join(timeout=10) |
| 561 | if thread.is_alive(): |
| 562 | # Uvicorn occasionally ignores should_exit; force_exit aborts the asyncio loop so teardown cannot hang |
| 563 | # the suite indefinitely. |
| 564 | server.force_exit = True |
| 565 | thread.join(timeout=5) |