Function test_read_request_disconnected_client

tests/middleware/test_base.py:894–934

If we receive a disconnect message when the downstream ASGI app calls receive() the Request instance passed into the dispatch function should get marked as disconnected. The downstream ASGI app should not get a ClientDisconnect raised, instead if should just receive the disconnect me

()

@pytest.mark.anyio
async def test_read_request_disconnected_client() -> None:
    """If we receive a disconnect message when the downstream ASGI
    app calls receive() the Request instance passed into the dispatch function
    should get marked as disconnected.
    The downstream ASGI app should not get a ClientDisconnect raised,
    instead if should just receive the disconnect message.
    """

    async def endpoint(scope: Scope, receive: Receive, send: Send) -> None:
        msg = await receive()
        assert msg["type"] == "http.disconnect"
        await Response()(scope, receive, send)

    class ConsumingMiddleware(BaseHTTPMiddleware):
        async def dispatch(
            self,
            request: Request,
            call_next: RequestResponseEndpoint,
        ) -> Response:
            response = await call_next(request)
            disconnected = await request.is_disconnected()
            assert disconnected is True
            return response

    scope = {"type": "http", "method": "POST", "path": "/"}

    async def receive() -> AsyncGenerator[Message, None]:
        yield {"type": "http.disconnect"}
        raise AssertionError("Should not be called, would hang")  # pragma: no cover

    async def send(msg: Message) -> None:
        if msg["type"] == "http.response.start":
            assert msg["status"] == 200

    app: ASGIApp = ConsumingMiddleware(endpoint)

    rcv = receive()

    await app(scope, rcv.__anext__, send)

    await rcv.aclose()


@pytest.mark.anyio