Chapter 21 · WebSocket and Real-Time Application Testing

Chapter goals: Learn to listen to and assert WebSocket traffic generated by pages, understand Playwright's WS mock capabilities and boundaries, and write stable automated tests for chat-room-style real-time features.

21.1 The Testing Challenge of Real-Time Apps

Chat, collaborative docs, market-data feeds, live support — data in these features is no longer driven by "request-response" but pushed continuously over WebSocket long connections. Traditional test approaches all fail here:

• expect_response can only wait for HTTP responses, not server-initiated pushes;
• The page may update at any moment — assertion timing is hard to nail;
• Long connection lifecycles span multiple page events; closing at the wrong time pollutes the next test.

Playwright's answer is two complementary APIs: listening (observe real traffic and assert) and WebSocketRoute (intercept/fabricate traffic).

21.2 Listening to WebSocket: page.on("websocket")

Every time the page establishes a WS connection, the page.on("websocket") event fires, passing a WebSocket object to the callback:

def test_ws_connection_established(page):
    events = []
    # Record every new WebSocket connection the page opens
    page.on("websocket", lambda ws: events.append(ws.url))

    page.goto("https://chat.example.com/room/1")
    page.wait_for_timeout(1000)  # Wait for connection (for demo; use assertions in real code)

    assert any(url.endswith("/ws/room/1") for url in events)

Once you have the WebSocket object, you can listen to its frame events:

• ws.on("framesent", handler): page sent a frame; callback receives str | bytes;
• ws.on("framereceived", handler): page received a frame; callback receives str | bytes;
• ws.on("socketerror", handler): connection error;
• ws.on("close", handler): connection closed; ws.is_closed() queries status anytime.

21.3 Asserting Message Content: Collect-and-Wait Pattern

WS messages arrive asynchronously; the standard pattern is "collect into a list, then assert with a polled timeout" (note: Node.js's expect(fn).toPass() does not exist in Python — we implement the same semantics with a few lines of helper):

import json
import time


def wait_until(predicate, timeout_ms=5000, interval_ms=100):
    """Poll until predicate is True; raise AssertionError on timeout (simulates Node's toPass)"""
    deadline = time.monotonic() + timeout_ms / 1000
    while time.monotonic() < deadline:
        if predicate():
            return
        time.sleep(interval_ms / 1000)
    raise AssertionError(f"Condition not met within {timeout_ms}ms")


def test_receive_push_message(page):
    received = []

    def on_ws(ws):
        if ws.url.endswith("/ws/room/1"):
            # Parse and store each received frame
            ws.on("framereceived", lambda data: received.append(json.loads(data)))

    page.on("websocket", on_ws)
    page.goto("https://chat.example.com/room/1")

    # Server pushes online count every 2s; poll-wait instead of fixed sleep
    wait_until(lambda: any(
        msg.get("type") == "presence" for msg in received
    ), timeout_ms=5000)

The send direction uses the same pattern — verify the frontend actually sent the user's input to the server:

def test_client_sends_chat_message(page):
    sent = []

    def on_ws(ws):
        if ws.url.endswith("/ws/room/1"):
            ws.on("framesent", lambda data: sent.append(json.loads(data)))

    page.on("websocket", on_ws)
    page.goto("https://chat.example.com/room/1")
    page.get_by_label("Message").fill("Hello team")
    page.get_by_role("button", name="Send").click()

    wait_until(lambda: any(
        msg.get("type") == "chat" and msg.get("content") == "Hello team"
        for msg in sent
    ), timeout_ms=3000)

21.4 Mocking WebSocket Traffic: WebSocketRoute

For tests that need deterministic control over server pushes (e.g., simulate "new message arrived" on demand), use page.route_websocket():

def test_mocked_ws_push(page):
    def handle(ws):
        # Simulate a server push after 500ms
        ws.send(json.dumps({"type": "chat", "content": "Hi from mock"}))

    page.route_websocket("**/ws/room/*", handle)
    page.goto("https://chat.example.com/room/1")
    expect(page.get_by_text("Hi from mock")).to_be_visible(timeout=3000)

route_websocket matches the connection URL and replaces the server's behavior entirely — useful for testing UI reactions to specific messages without depending on a real backend.

21.5 Chapter Summary

• page.on("websocket") + ws.on("framereceived"/"framesent") lets you observe real WS traffic;
• Collect-then-poll-wait is the standard assertion pattern for async WS messages;
• page.route_websocket() mocks WS traffic for deterministic testing;
• WS connections outlive page navigations — manage lifecycle carefully to avoid cross-test pollution.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. How do you capture WebSocket messages in Playwright?

2. Why use a collect-then-poll-wait pattern instead of a fixed sleep for WS assertions?

3. What does page.route_websocket() do?

4. What lifecycle concern is unique to WebSocket testing?

🛠️ Hands-On Exercises

1. Build a local WebSocket echo server; write a test that connects, sends a message, and asserts the echoed response via framereceived.
2. Use route_websocket to mock a streaming chat; send three synthetic messages and assert each appears on the page in order.
3. Intentionally leak a WS connection between two tests; observe the cross-test pollution and fix it with proper cleanup.

Proceed to Chapter 22: Production Practice — E2E Strategy and Reporting.