Function sse_security_client

tests/server/test_sse_security.py:43–68

An httpx client whose requests are served in process by an SSE app with the given settings.

(security_settings: TransportSecuritySettings | None = None)

def sse_security_client(security_settings: TransportSecuritySettings | None = None) -> httpx.AsyncClient:
    """An httpx client whose requests are served in process by an SSE app with the given settings."""
    server = Server(SERVER_NAME)
    sse_transport = SseServerTransport("/messages/", security_settings)

    async def handle_sse(request: Request) -> Response:
        try:
            async with sse_transport.connect_sse(request.scope, request.receive, request._send) as (read, write):
                await server.run(read, write, server.create_initialization_options())
        except ValueError as e:
            # Validation error was already handled inside connect_sse, which sent the rejection
            # response itself; its non-empty body checkpoints, so the test reads the rejection
            # status before the trailing Response() below sends a second response start.
            logger.debug(f"SSE connection failed validation: {e}")
        return Response()

    app = Starlette(
        routes=[
            Route("/sse", endpoint=handle_sse),
            Mount("/messages/", app=sse_transport.handle_post_message),
        ]
    )
    # The SSE GET runs until it observes a disconnect, so the bridge must let the application
    # drain on close rather than cancelling it.
    transport = StreamingASGITransport(app, cancel_on_close=False)
    return httpx.AsyncClient(transport=transport, base_url=BASE_URL)


@pytest.mark.anyio