Function _sse_scope

tests/server/test_sse_security.py:206–228

Build an ASGI scope/receive/send triple for a request to the SSE transport.

(
    method: str, path: str, user: AuthenticatedUser | None, *, query_string: bytes = b"", body: bytes = b""
)

def _sse_scope(
    method: str, path: str, user: AuthenticatedUser | None, *, query_string: bytes = b"", body: bytes = b""
) -> tuple[Scope, Receive, Send, list[Message]]:
    """Build an ASGI scope/receive/send triple for a request to the SSE transport."""
    scope: Scope = {
        "type": "http",
        "method": method,
        "path": path,
        "root_path": "",
        "query_string": query_string,
        "headers": [(b"content-type", b"application/json")],
    }
    if user is not None:
        scope["user"] = user
    sent: list[Message] = []

    async def receive() -> Message:
        return {"type": "http.request", "body": body, "more_body": False}

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

    return scope, receive, send, sent


def _response_status(sent: list[Message]) -> int: