Skip to content

Testing WebSocket Servers

WebSocket handlers are long-running conversations, so their bugs hide in sequences: the second message, the reply that never comes, the client that leaves halfway, the token that expires. Testing them does not need a deployed server. Measured with pytest on Python 3.14: 200 tests that each opened a WebSocket through Starlette's TestClient, exchanged a message and closed ran in 0.77 s; 200 tests that each started a real websockets server on a free port and connected a real client ran in 0.69 s — about 3–4 ms per test either way. A rejection test asserted the server's close code 4401 in the same suite. This guide sets up both styles, tests the paths that matter, and keeps tests from hanging when the server misbehaves.

Prerequisites

1. Test ASGI apps with Starlette's TestClient

For FastAPI and Starlette, TestClient.websocket_connect runs the app in-process without a server or a socket:

from starlette.testclient import TestClient
from starlette.websockets import WebSocketDisconnect


def test_ping_pong():
    with TestClient(app) as client, client.websocket_connect("/ws?token=t") as ws:
        ws.send_json({"type": "ping"})
        assert ws.receive_json() == {"type": "pong"}


def test_rejects_without_token():
    with TestClient(app) as client:
        with pytest.raises(WebSocketDisconnect) as exc_info:
            with client.websocket_connect("/ws") as ws:
                ws.receive_json()
        assert exc_info.value.code == 4401

Measured: 200 such tests ran in 0.77 s. TestClient is synchronous — it runs the app in a background thread with its own event loop — so tests are plain functions. Using it as a context manager (with TestClient(app) as client) also runs the app's lifespan, so hubs, pools and relays created at startup exist during the test. The rejection test asserts the exact close code, which is the contract clients code against.

Verify: the suite covers accept, every rejection close code, and at least one full message exchange per message type.

200 WebSocket tests, by style 2 horizontal bars comparing Starlette TestClient, in process with the others. 200 WebSocket tests, by style Starlette TestClient, in process 0.77 s real websockets server per test 0.69 s Python 3.14, pytest with pytest-asyncio; each test connects, exchanges one message and closes. Both styles cost a few milliseconds per test; use whichever matches your server.

2. Test real sockets with a server on port 0

For servers built on the websockets library, or to test behaviour that depends on the real protocol — compression, ping timeouts, frame sizes — start the server in the test on a free port:

import pytest_asyncio
from websockets.asyncio.client import connect
from websockets.asyncio.server import serve


@pytest_asyncio.fixture
async def ws_url():
    async with serve(handler, "127.0.0.1", 0) as server:            # 0: the OS picks a port
        port = next(iter(server.server.sockets)).getsockname()[1]
        yield f"ws://127.0.0.1:{port}"


async def test_echo(ws_url):
    async with connect(ws_url) as ws:
        await ws.send("hi")
        assert await asyncio.wait_for(ws.recv(), 1.0) == "hi"

Measured: 200 tests, each with its own server, in 0.69 s. A server per test isolates state between tests at negligible cost. The same fixture shape works for an ASGI app under a real Uvicorn server started in-process, when you need Uvicorn's own WebSocket behaviour rather than TestClient's.

Verify: tests run in parallel (pytest -n auto) without port conflicts.

3. Bound every receive in tests

A handler bug that means "no reply" turns await ws.recv() into a test that hangs until the CI job times out, with no useful failure message. Wrap every receive:

async def recv_json(ws, timeout: float = 1.0) -> dict:
    try:
        return json.loads(await asyncio.wait_for(ws.recv(), timeout))
    except TimeoutError:
        pytest.fail(f"no message within {timeout}s")


async def assert_silent(ws, for_seconds: float = 0.2) -> None:
    """Assert the server sends nothing - e.g. before authentication."""
    with pytest.raises(TimeoutError):
        await asyncio.wait_for(ws.recv(), for_seconds)

assert_silent covers the important negative cases: no data before authentication, no messages from rooms the client did not join. Keep the timeouts short; the in-process server replies in microseconds, so a second is generous. A global safety net such as pytest-timeout catches anything missed, but per-receive timeouts give a failure that names the step.

Verify: deliberately break the handler so it does not reply; the test fails in about a second with "no message within 1.0s".

The shape of a robust WebSocket test A flow of 5 stages. The shape of a robust WebSocket test server on port 0 or TestClient connect with auth variant send + recv(timeout) assert reply assert silence where nothing is due close path assert close code Every receive has a deadline; every ending has an assertion.

4. Test disconnects, multiple clients and ordering

The paths that break in production involve more than one client or an abrupt end. Test them explicitly:

async def test_broadcast_reaches_other_clients(ws_url):
    async with connect(ws_url + "/rooms/a") as alice, connect(ws_url + "/rooms/a") as bob:
        await alice.send("hello")
        assert await asyncio.wait_for(bob.recv(), 1.0) == "hello"


async def test_cleanup_after_abrupt_disconnect(ws_url, hub):
    ws = await connect(ws_url + "/rooms/a")
    ws.transport.abort()                                  # no close frame: like a dropped network
    await wait_until(lambda: hub.member_count("a") == 0, timeout=2.0)


async def test_messages_arrive_in_order(ws_url):
    async with connect(ws_url + "/stream") as ws:
        received = [json.loads(await asyncio.wait_for(ws.recv(), 1.0))["seq"] for _ in range(100)]
    assert received == list(range(100))

transport.abort() simulates a client that vanishes without a close frame, the case that leaks subscriptions when cleanup only runs on clean closes. wait_until polls a condition with a deadline instead of sleeping a fixed time, so the test is fast when the server is fast and fails clearly when it is not. Multi-client tests catch the per-process broadcast bug from scaling WebSockets across processes with Redis pub/sub when run against two server instances.

Verify: each of these tests fails when the corresponding cleanup or fan-out code is removed.

5. Test time-dependent behaviour without waiting

Heartbeats, auth deadlines and token expiry depend on time. Make the durations configurable so tests run them in milliseconds:

@pytest_asyncio.fixture
async def fast_server():
    settings = Settings(auth_deadline=0.1, ping_interval=0.05, ping_timeout=0.05)
    async with serve(make_handler(settings), "127.0.0.1", 0,
                     ping_interval=settings.ping_interval,
                     ping_timeout=settings.ping_timeout) as server:
        yield server


async def test_silent_client_closed_after_auth_deadline(fast_server_url):
    async with connect(fast_server_url) as ws:
        with pytest.raises(ConnectionClosed) as exc_info:
            await asyncio.wait_for(ws.recv(), 1.0)
    assert exc_info.value.rcvd.code == 4401

The production deadline of 2 s, as in authenticating WebSocket connections, becomes 0.1 s in the test, exercising the same code path in a fraction of the time. For heartbeat tests, a client that stops responding to pings can be simulated by pausing its reading (ws.transport.pause_reading()), so the server's ping timeout fires.

Verify: the time-dependent tests together take well under a second, and changing a deadline in code without changing behaviour does not break them.

Which test harness fits this WebSocket code? A decision on What is under test with 4 outcomes. Which test harness fits this WebSocket code? What is under test? FastAPI/Starlette handler logic TestClient.websocket_connect no sockets websockets server, protocol details real server on port 0 ping, compression cross-instance broadcast two servers + broker real fan-out deadlines, heartbeats, expiry configurable short durations ms, not s Fast in-process tests can cover almost every WebSocket behaviour.

Verification

A WebSocket test suite is solid when:

  • Every receive has a timeout, and silence is asserted where nothing should arrive.
  • Rejections assert exact close codes.
  • Abrupt disconnects, multiple clients and ordering each have tests.
  • Time-dependent behaviour runs with short, configurable durations.

Diagnostic Hook: track test durations and flakiness for the WebSocket suite. Tests that occasionally take exactly their receive timeout point at races in the handler — a reply sent before a subscription is registered, or cleanup that sometimes runs late; fix the handler, not the timeout.

Pitfalls & edge cases

  • Unbounded recv() in tests. A missing reply hangs the whole suite.
  • Only testing clean closes. Abrupt disconnects are where subscriptions leak.
  • Fixed sleeps. Poll with a deadline instead.
  • Production timeouts in tests. Make them configurable and short.

Frequently Asked Questions

How do I test FastAPI WebSockets with pytest?

Use starlette.testclient.TestClient(app) as a context manager and client.websocket_connect(path) to send and receive messages in a synchronous test. In testing, 200 such tests ran in 0.77 s.

How do I test a websockets library server in pytest?

Start it in an async fixture with serve(handler, "127.0.0.1", 0), read the assigned port from the server's socket, and connect a real client to it. 200 tests with a server each ran in 0.69 s.

How do I assert a WebSocket close code in a test?

With TestClient, catch WebSocketDisconnect and check its code; with the websockets client, catch ConnectionClosed and check exc.rcvd.code.

How do I simulate a client that drops its connection?

Call ws.transport.abort() on the client, which closes the TCP connection without a WebSocket close frame, and then assert the server cleaned up within a deadline.