Function test_asgi_pathsend_events

tests/middleware/test_base.py:1219–1259

(tmpdir: Path)

@pytest.mark.anyio
async def test_asgi_pathsend_events(tmpdir: Path) -> None:
    path = tmpdir / "example.txt"
    with path.open("w") as file:
        file.write("<file content>")

    response_complete = anyio.Event()
    events: list[Message] = []

    async def endpoint_with_pathsend(_: Request) -> FileResponse:
        return FileResponse(path)

    async def passthrough(request: Request, call_next: RequestResponseEndpoint) -> Response:
        return await call_next(request)

    app = Starlette(
        middleware=[Middleware(BaseHTTPMiddleware, dispatch=passthrough)],
        routes=[Route("/", endpoint_with_pathsend)],
    )

    scope = {
        "type": "http",
        "version": "3",
        "method": "GET",
        "path": "/",
        "headers": [],
        "extensions": {"http.response.pathsend": {}},
    }

    async def receive() -> Message:
        raise NotImplementedError("Should not be called!")  # pragma: no cover

    async def send(message: Message) -> None:
        events.append(message)
        if message["type"] == "http.response.pathsend":
            response_complete.set()

    await app(scope, receive, send)

    assert len(events) == 2
    assert events[0]["type"] == "http.response.start"
    assert events[1]["type"] == "http.response.pathsend"


def test_error_context_propagation(test_client_factory: TestClientFactory) -> None: