Function test_streaming_response_on_client_disconnects

tests/test_responses.py:636–665

()

@pytest.mark.anyio
async def test_streaming_response_on_client_disconnects() -> None:
    chunks = bytearray()
    streamed = False

    async def receive_disconnect() -> Message:
        raise NotImplementedError

    async def send(message: Message) -> None:
        nonlocal streamed
        if message["type"] == "http.response.body":
            if not streamed:
                chunks.extend(message.get("body", b""))
                streamed = True
            else:
                raise OSError

    async def stream_indefinitely() -> AsyncGenerator[bytes, None]:
        while True:
            await anyio.sleep(0)
            yield b"chunk"

    stream = stream_indefinitely()
    response = StreamingResponse(content=stream)

    with anyio.move_on_after(1) as cancel_scope:
        with pytest.raises(ClientDisconnect):
            await response({"type": "http", "asgi": {"spec_version": "2.4"}}, receive_disconnect, send)
    assert not cancel_scope.cancel_called, "Content streaming should stop itself."
    assert chunks == b"chunk"
    await stream.aclose()


@pytest.mark.anyio