0779f7f8a4
* Poll for readiness in `test_multiprocess_health_check` and `run_server` Apply the same polling pattern to `test_multiprocess_health_check` — wait for all processes to be alive instead of a fixed 1s sleep. In `run_server`, wait for `server.started` instead of a fixed 0.1s sleep to avoid connection races on slow setups (e.g. free-threaded Python 3.14). * Avoid uncovered branch in health check polling loop * Add pragma: no cover to health check polling loop body
57 lines
1.5 KiB
Python
57 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import signal
|
|
import sys
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager, contextmanager
|
|
from pathlib import Path
|
|
from socket import socket
|
|
|
|
from uvicorn import Config, Server
|
|
|
|
|
|
@asynccontextmanager
|
|
async def run_server(config: Config, sockets: list[socket] | None = None) -> AsyncIterator[Server]:
|
|
server = Server(config=config)
|
|
task = asyncio.create_task(server.serve(sockets=sockets))
|
|
while not server.started:
|
|
await asyncio.sleep(0.05)
|
|
try:
|
|
yield server
|
|
finally:
|
|
await server.shutdown()
|
|
task.cancel()
|
|
|
|
|
|
@contextmanager
|
|
def assert_signal(sig: signal.Signals):
|
|
"""Check that a signal was received and handled in a block"""
|
|
seen: set[int] = set()
|
|
prev_handler = signal.signal(sig, lambda num, frame: seen.add(num))
|
|
try:
|
|
yield
|
|
assert sig in seen, f"process signal {signal.Signals(sig)!r} was not received or handled"
|
|
finally:
|
|
signal.signal(sig, prev_handler)
|
|
|
|
|
|
@contextmanager
|
|
def as_cwd(path: Path):
|
|
"""Changes working directory and returns to previous on exit."""
|
|
prev_cwd = Path.cwd()
|
|
os.chdir(path)
|
|
try:
|
|
yield
|
|
finally:
|
|
os.chdir(prev_cwd)
|
|
|
|
|
|
def get_asyncio_default_loop_per_os() -> type[asyncio.AbstractEventLoop]:
|
|
"""Get the default asyncio loop per OS."""
|
|
if sys.platform == "win32":
|
|
return asyncio.ProactorEventLoop # type: ignore # pragma: nocover
|
|
else:
|
|
return asyncio.SelectorEventLoop # pragma: nocover
|