Function run_server

tests/utils.py:16–25

(config: Config, sockets: list[socket] | None = None)

@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