第 21 章 · WebSocket 与实时应用测试

本章目标：学会监听和断言页面产生的 WebSocket 流量，理解 Playwright 对 WS 的 mock 能力与边界，能对聊天室类实时功能写出稳定的自动化测试。

21.1 实时应用的测试难点

聊天、协同文档、行情推送、在线客服——这类功能的数据不再由"请求-响应"驱动，而是通过 WebSocket 长连接持续推送。传统测试手段在这里纷纷失灵：

expect_response 只能等 HTTP 响应，等不到服务端主动推送；
页面可能在任意时刻更新，断言时机难以拿捏；
长连接的生命周期跨越多个页面事件，关闭时机不当会污染下一个测试。

Playwright 的答案是两个互补的 API：监听（观察真实流量并断言）与 WebSocketRoute（拦截/伪造流量）。

21.2 监听 WebSocket：page.on("websocket")

页面每次建立 WS 连接，都会触发 page.on("websocket") 事件，回调拿到一个 WebSocket 对象：

python

def test_ws_connection_established(page):
    events = []
    # 每当页面新建一条 WebSocket 连接就记录下来
    page.on("websocket", lambda ws: events.append(ws.url))

    page.goto("https://chat.example.com/room/1")
    page.wait_for_timeout(1000)  # 等连接建立（演示用，正式代码用断言等待）

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

拿到 WebSocket 对象后，可以继续监听它的帧事件：

ws.on("framesent", handler)：页面发送了帧，回调参数为 str | bytes；
ws.on("framereceived", handler)：页面收到了帧，回调参数为 str | bytes；
ws.on("socketerror", handler)：连接出错；
ws.on("close", handler)：连接关闭；ws.is_closed() 可随时查询状态。

21.3 断言消息内容：收集 + 等待模式

WS 消息是异步到达的，惯用模式是“收集到列表，再用带截止时间的轮询断言”（注意：Node.js 的 expect(fn).toPass() 在 Python 版中并不存在，我们用一个几行的小助手实现同样的语义）：

python

import json
import time


def wait_until(predicate, timeout_ms=5000, interval_ms=100):
    """轮询直到 predicate 为真，超时抛 AssertionError（模拟 Node 的 toPass 语义）"""
    deadline = time.monotonic() + timeout_ms / 1000
    while time.monotonic() < deadline:
        if predicate():
            return
        time.sleep(interval_ms / 1000)
    raise AssertionError(f"条件在 {timeout_ms}ms 内未满足")


def test_receive_push_message(page):
    received = []

    def on_ws(ws):
        if ws.url.endswith("/ws/room/1"):
            # 把每条收到的帧解析后存入列表
            ws.on("framereceived", lambda data: received.append(json.loads(data)))

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

    # 服务端每 2 秒推送一次在线人数；轮询等待而不是固定 sleep
    wait_until(lambda: any(
        msg.get("type") == "presence" for msg in received
    ), timeout_ms=5000)

发送方向的断言同理——验证前端确实把用户输入发给了服务端：

python

def test_client_sends_chat_message(page):
    sent = []

    def on_ws(ws):
        ws.on("framesent", lambda data: sent.append(data))

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

    page.get_by_label("消息").fill("大家好")
    page.get_by_role("button", name="发送").click()

    assert "大家好" in sent  # framesent 的载荷就是原始字符串/字节

21.4 Mock WebSocket：WebSocketRoute

当后端还没开发完、或你想模拟异常场景（断线、错误帧）时，可以拦截 WS 连接并伪造服务端行为。在 Python 同步 API 中通过 page.route_websocket() 获得 WebSocketRoute 对象：

python

def test_ui_renders_mock_push(page):
    def handle(ws):
        # 拦截发往 /ws/room/1 的连接，本地伪造服务端
        ws.on_message(lambda message: ws.send('{"type": "echo", "data": "%s"}' % message))

    page.route_websocket("**/ws/room/1", handle)
    page.goto("https://chat.example.com/room/1")

    page.get_by_label("消息").fill("hello")
    page.get_by_role("button", name="发送").click()

    # 页面应把 mock 服务端回显的内容渲染出来
    expect(page.get_by_text("hello")).to_be_visible()

WebSocketRoute 的常用能力：

ws.send(payload)：模拟服务端向页面推送消息；
ws.on_message(handler)：观察/改写页面发出的消息（handler 返回值可替换原消息）；
ws.close()：主动模拟断线——测试前端的重连与错误提示逻辑。

边界提醒

Mock 之后流量不再到达真实后端，因此它验证的是前端对 WS 协议的处理，而不是端到端链路。协议契约本身建议用集成测试覆盖，两者不要互相替代。

21.5 综合案例：聊天室收发全链路

把监听与 UI 断言组合，覆盖"发送→广播→渲染"的完整闭环：

python

```python
import json
import time
from playwright.sync_api import expect


def wait_until(predicate, timeout_ms=5000, interval_ms=100):
    deadline = time.monotonic() + timeout_ms / 1000
    while time.monotonic() < deadline:
        if predicate():
            return
        time.sleep(interval_ms / 1000)
    raise AssertionError(f"条件在 {timeout_ms}ms 内未满足")


def test_chat_roundtrip(page):
    received = []

    def on_ws(ws):
        ws.on("framereceived", lambda d: received.append(json.loads(d)))
        ws.on("socketerror", lambda e: print("WS 错误:", e))

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

    # 用户发送消息
    page.get_by_label("消息").fill("测试消息 001")
    page.get_by_role("button", name="发送").click()

    # UI 上出现自己的消息（服务端广播回来后渲染）
    expect(
        page.locator(".chat-list").get_by_text("测试消息 001")
    ).to_be_visible()

    # 协议层确认服务端确实回推了该消息
    wait_until(lambda: any(
        m.get("type") == "chat" and m.get("text") == "测试消息 001"
        for m in received
    ), timeout_ms=5000)

21.6 长连接的清理与隔离

WS 测试最容易留下的坑是连接泄漏：上一个测试的连接还开着，下一个测试收到"幽灵消息"。三条军规：

每个测试用独立的 BrowserContext（pytest-playwright 的 page fixture 默认如此），连接随 context 关闭；
测试结束前主动断言/等待连接关闭，避免 teardown 竞态：

python

def test_connection_closes_cleanly(page):
    sockets = []
    page.on("websocket", lambda ws: sockets.append(ws))

    page.goto("https://chat.example.com/room/1")
    page.close()  # 触发前端断开逻辑

    assert all(ws.is_closed() for ws in sockets)

不要在 framereceived 回调里做重活——回调阻塞可能拖慢页面事件循环，收集数据、断言放在主流程里做。

21.7 本章小结

page.on("websocket") 捕获连接建立，WebSocket 对象提供 framesent/framereceived/socketerror/close 四类事件，帧载荷为 str | bytes；
WS 消息断言用“收集 + 带截止时间的轮询”模式，不要 sleep；
page.route_websocket() + WebSocketRoute 可伪造服务端推送、改写消息、模拟断线，但只覆盖前端逻辑；
长连接测试三要素：独立 context、显式关闭断言、回调里只收集不处理。

🧪 随堂测验

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

1. 在 Playwright Python 中，页面建立 WebSocket 连接时触发的事件是？

2. ws.on("framereceived", handler) 的 handler 收到的参数类型是？

3. 想测试"前端收到断线后是否显示重连提示"，最合适的做法是？

4. 关于 WS 测试的实践，错误的是？

🛠️ 动手实践

给你项目中的一个 WS 接口写"监听版"测试：断言页面加载后 5 秒内收到至少一条指定类型的推送。
用 route_websocket 伪造一条告警推送，断言页面出现对应的通知 UI；再用 ws.close() 模拟断线，验证重连提示。
为聊天发送功能补一条 framesent 断言，确认消息体里的字段名与后端协议文档一致。

最后一章我们把全部技能收拢成可落地的工程体系：第 22 章 · 生产实践：E2E 策略与报告。

第 21 章 · WebSocket 与实时应用测试 ​

21.1 实时应用的测试难点 ​

21.2 监听 WebSocket：page.on("websocket") ​

21.3 断言消息内容：收集 + 等待模式 ​

21.4 Mock WebSocket：WebSocketRoute ​

21.5 综合案例：聊天室收发全链路 ​

21.6 长连接的清理与隔离 ​

21.7 本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 21 章 · WebSocket 与实时应用测试

21.1 实时应用的测试难点

21.2 监听 WebSocket：page.on("websocket")

21.3 断言消息内容：收集 + 等待模式

21.4 Mock WebSocket：WebSocketRoute

21.5 综合案例：聊天室收发全链路

21.6 长连接的清理与隔离

21.7 本章小结

🛠️ 动手实践