Function _raw_server

tests/test_server.py:156–188

(
    *,
    app: ASGIApplication,
    port: int,
    http_protocol_cls: type[H11Protocol | HttpToolsProtocol],
    reset_contextvars: bool = False,
)

@contextlib.asynccontextmanager
async def _raw_server(
    *,
    app: ASGIApplication,
    port: int,
    http_protocol_cls: type[H11Protocol | HttpToolsProtocol],
    reset_contextvars: bool = False,
):
    config = Config(app=app, port=port, loop=class="st">"asyncio", http=http_protocol_cls, reset_contextvars=reset_contextvars)
    server = Server(config=config)
    task = asyncio.create_task(server.serve())

    while not server.started:
        await asyncio.sleep(0.01)

    reader, writer = await asyncio.open_connection(class="st">"127.0.0.1", port)

    async def extract_json_body(request: bytes):
        writer.write(request)
        await writer.drain()

        status, *headers = (await reader.readuntil(bclass="st">"\r\n\r\n")).split(bclass="st">"\r\n")[:-2]
        assert status == bclass="st">"HTTP/1.1 200 OK"

        content_length = next(int(h.split(bclass="st">":", 1)[1]) for h in headers if h.lower().startswith(bclass="st">"content-length:"))
        return json.loads(await reader.readexactly(content_length))

    try:
        yield extract_json_body
    finally:
        writer.close()
        await writer.wait_closed()
        server.should_exit = True
        await task


async def test_contextvars_preserved_by_default(