Function _post_message

tests/server/test_sse_security.py:236–243

POST a message to an SSE session as `user` and return the response status.

(transport: SseServerTransport, session_id: str, user: AuthenticatedUser | None)

async def _post_message(transport: SseServerTransport, session_id: str, user: AuthenticatedUser | None) -> int:
    """POST a message to an SSE session as `user` and return the response status."""
    body = b'{"jsonrpc": "2.0", "id": 1, "method": "ping", "params": null}'
    scope, receive, send, sent = _sse_scope(
        "POST", "/messages/", user, query_string=f"session_id={session_id}".encode(), body=body
    )
    await transport.handle_post_message(scope, receive, send)
    return _response_status(sent)


_Principal = tuple[str] | tuple[str, str] | tuple[str, str, str]