Function test_reset_contextvars_asyncio

tests/test_server.py:214–263

With reset_contextvars=True, each ASGI run starts with a fresh context. Non-regression test for https://github.com/encode/uvicorn/issues/2167.

(
    http_protocol_cls: type[H11Protocol | HttpToolsProtocol], unused_tcp_port: int
)

async def test_reset_contextvars_asyncio(
    http_protocol_cls: type[H11Protocol | HttpToolsProtocol], unused_tcp_port: int
):
    """With reset_contextvars=True, each ASGI run starts with a fresh context.

    Non-regression test for https://github.com/encode/uvicorn/issues/2167.
    """
    default_contextvars = {c.name for c in contextvars.copy_context().keys()}
    ctx: contextvars.ContextVar[str] = contextvars.ContextVar("ctx")

    async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
        assert scope["type"] == "http"

        # initial context should be empty
        initial_context = {
            n: v for c, v in contextvars.copy_context().items() if (n := c.name) not in default_contextvars
        }
        # set any contextvar before the body is read
        ctx.set(scope["path"])

        while True:
            message = await receive()
            assert message["type"] == "http.request"
            if not message["more_body"]:
                break

        body = json.dumps(initial_context).encode("utf-8")
        headers = [(b"content-type", b"application/json"), (b"content-length", str(len(body)).encode("utf-8"))]
        await send({"type": "http.response.start", "status": 200, "headers": headers})
        await send({"type": "http.response.body", "body": body})

    # body larger than HIGH_WATER_LIMIT forces a reading pause on the main thread
    # and a resumption inside the ASGI task, which is where the original pollution showed up.
    large_body = b"a" * (HIGH_WATER_LIMIT + 1)
    large_request = b"\r\n".join(
        [
            b"POST /large-body HTTP/1.1",
            b"Host: example.org",
            b"Content-Type: application/octet-stream",
            f"Content-Length: {len(large_body)}".encode(),
            b"",
            large_body,
        ]
    )

    async with _raw_server(
        app=app, http_protocol_cls=http_protocol_cls, port=unused_tcp_port, reset_contextvars=True
    ) as extract_json_body:
        assert await extract_json_body(large_request) == {}
        assert await extract_json_body(SIMPLE_GET_REQUEST) == {}