Function test_poll_for_disconnect_repeated

tests/middleware/test_base.py:1170–1215

(send_body: bool)

@pytest.mark.anyio
@pytest.mark.parametrize("send_body", [True, False])
async def test_poll_for_disconnect_repeated(send_body: bool) -> None:
    async def app_poll_disconnect(scope: Scope, receive: Receive, send: Send) -> None:
        for _ in range(2):
            msg = await receive()
            while msg["type"] == "http.request":
                msg = await receive()
            assert msg["type"] == "http.disconnect"
        await Response(b"good!")(scope, receive, send)

    class MyMiddleware(BaseHTTPMiddleware):
        async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
            return await call_next(request)

    app = MyMiddleware(app_poll_disconnect)

    scope = {
        "type": "http",
        "version": "3",
        "method": "GET",
        "path": "/",
    }

    async def receive() -> AsyncIterator[Message]:
        # the key here is that we only ever send 1 htt.disconnect message
        if send_body:
            yield {"type": "http.request", "body": b"hello", "more_body": True}
            yield {"type": "http.request", "body": b"", "more_body": False}
        yield {"type": "http.disconnect"}
        raise AssertionError("Should not be called, would hang")  # pragma: no cover

    sent: list[Message] = []

    async def send(message: Message) -> None:
        sent.append(message)

    await app(scope, receive().__anext__, send)

    assert sent == [
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [(b"content-length", b"5")],
        },
        {"type": "http.response.body", "body": b"good!", "more_body": True},
        {"type": "http.response.body", "body": b"", "more_body": False},
    ]


@pytest.mark.anyio