FastAPI WebSockets with token auth and broadcast, tested with TestClient: working versions (tested 2026-10-15)

fastapi==0.143.1
httpx==0.28.1
pytest==9.1.1

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, status

app = FastAPI()
TOKENS = {"t-ada": "ada", "t-bob": "bob"}

class Hub:
    def __init__(self):
        self.clients: dict[WebSocket, str] = {}

    async def broadcast(self, data: dict):
        for ws in list(self.clients):
            await ws.send_json(data)

hub = Hub()

@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket, token: str = ""):
    user = TOKENS.get(token)
    if not user:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    await websocket.accept()
    hub.clients[websocket] = user
    try:
        while True:
            msg = await websocket.receive_json()
            await hub.broadcast({"from": user, "text": msg.get("text", "")})
    except WebSocketDisconnect:
        hub.clients.pop(websocket, None)

import pytest
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect
from app.main import app

client = TestClient(app)

def test_echo_broadcast_to_all():
    with client.websocket_connect("/ws?token=t-ada") as a, client.websocket_connect("/ws?token=t-bob") as b:
        a.send_json({"text": "hi"})
        assert a.receive_json() == {"from": "ada", "text": "hi"}
        assert b.receive_json() == {"from": "ada", "text": "hi"}

def test_bad_token_closed_with_1008():
    with pytest.raises(WebSocketDisconnect) as e:
        with client.websocket_connect("/ws?token=nope") as ws:
            ws.receive_json()
    assert e.value.code == 1008

[pytest]
pythonpath = .