Class DiscardingMiddleware

tests/middleware/test_base.py:476–494

    test_client_factory: TestClientFactory,
) -> None:
    class DiscardingMiddleware(BaseHTTPMiddleware):
        async def dispatch(
            self,
            request: Request,
            call_next: Any,
        ) -> PlainTextResponse:
            # As a matter of ordering, this test targets the case where the downstream
            # app response is discarded while it is sending a response body.
            # We need to wait for the downstream app to begin sending a response body
            # before sending the middleware response that will overwrite the downstream
            # response.
            downstream_app_response = await call_next(request)
            body_generator = downstream_app_response.body_iterator
            try:
                await body_generator.__anext__()
            finally:
                await body_generator.aclose()

            return PlainTextResponse("Custom")

    async def downstream_app(
        scope: Scope,