# WebSockets 与实时测试 > **适用场景**：当你的应用使用 WebSocket、Server-Sent Events（SSE）或轮询来实现实时功能时——聊天、实时仪表盘、通知、协作编辑、股票行情、实时体育比分。 > **前置知识**：[core/assertions-and-waiting.md](assertions-and-waiting.md)、[core/fixtures-and-hooks.md](fixtures-and-hooks.md) ## 快速参考 ```typescript // 监听 WebSocket 连接 page.on("websocket", (ws) => { console.log("WebSocket 已打开：", ws.url()) ws.on("framesent", (frame) => console.log("发送：", frame.payload)) ws.on("framereceived", (frame) => console.log("接收：", frame.payload)) ws.on("close", () => console.log("WebSocket 已关闭")) }) // 通过路由模拟 WebSocket（Playwright 1.48+） await page.routeWebSocket("**/ws", (ws) => { ws.onMessage((message) => { ws.send(JSON.stringify({ echo: message })) }) }) ``` ## 模式 ### 观察 WebSocket 流量 **适用场景**：你需要验证应用发送和接收了正确的 WebSocket 消息，且无需修改这些消息。 **避免场景**：你需要拦截或模拟消息。此时应改用 `routeWebSocket`。 **TypeScript** ```typescript import { test, expect } from "@playwright/test" test("聊天消息通过 WebSocket 发送", async ({ page }) => { const messages: { direction: string; payload: string }[] = [] page.on("websocket", (ws) => { ws.on("framesent", (frame) => { messages.push({ direction: "sent", payload: String(frame.payload) }) }) ws.on("framereceived", (frame) => { messages.push({ direction: "received", payload: String(frame.payload) }) }) }) await page.goto("/chat") await page.getByRole("textbox", { name: "Message" }).fill("Hello!") await page.getByRole("button", { name: "Send" }).click() // 等待消息出现在 UI 中（确认往返成功） await expect(page.getByText("Hello!")).toBeVisible() // 验证 WebSocket 流量 const sentMessage = messages.find((m) => m.direction === "sent" && m.payload.includes("Hello!")) expect(sentMessage).toBeDefined() }) ``` **JavaScript** ```javascript const { test, expect } = require("@playwright/test") test("聊天消息通过 WebSocket 发送", async ({ page }) => { const messages = [] page.on("websocket", (ws) => { ws.on("framesent", (frame) => { messages.push({ direction: "sent", payload: String(frame.payload) }) }) ws.on("framereceived", (frame) => { messages.push({ direction: "received", payload: String(frame.payload) }) }) }) await page.goto("/chat") await page.getByRole("textbox", { name: "Message" }).fill("Hello!") await page.getByRole("button", { name: "Send" }).click() await expect(page.getByText("Hello!")).toBeVisible() const sentMessage = messages.find((m) => m.direction === "sent" && m.payload.includes("Hello!")) expect(sentMessage).toBeDefined() }) ``` ### 等待特定 WebSocket 消息 **适用场景**：你的测试需要依赖某条特定的服务器推送消息才能继续执行。 **避免场景**：UI 已经反映了该消息。此时应直接在 UI 上进行断言。 **TypeScript** ```typescript import { test, expect } from "@playwright/test" test("等待服务器通过 WebSocket 发送确认", async ({ page }) => { // 创建一个 Promise，在收到特定消息时 resolve const ackPromise = new Promise((resolve) => { page.on("websocket", (ws) => { ws.on("framereceived", (frame) => { const data = JSON.parse(String(frame.payload)) if (data.type === "message_ack") { resolve() } }) }) }) await page.goto("/chat") await page.getByRole("textbox", { name: "Message" }).fill("Important update") await page.getByRole("button", { name: "Send" }).click() // 等待服务器确认 await ackPromise // 现在验证消息显示了一个"已送达"的勾选标记 await expect(page.getByTestId("message-status").last()).toHaveText("Delivered") }) ``` **JavaScript** ```javascript const { test, expect } = require("@playwright/test") test("等待服务器通过 WebSocket 发送确认", async ({ page }) => { const ackPromise = new Promise((resolve) => { page.on("websocket", (ws) => { ws.on("framereceived", (frame) => { const data = JSON.parse(String(frame.payload)) if (data.type === "message_ack") { resolve() } }) }) }) await page.goto("/chat") await page.getByRole("textbox", { name: "Message" }).fill("Important update") await page.getByRole("button", { name: "Send" }).click() await ackPromise await expect(page.getByTestId("message-status").last()).toHaveText("Delivered") }) ``` ### 使用 `routeWebSocket` 模拟 WebSocket 消息 **适用场景**：你需要控制服务器发送的内容来测试特定的 UI 状态——错误消息、边界情况、大量数据——而无需真实的后端。 **避免场景**：你需要测试实际服务器行为。此时应使用真实后端。 **TypeScript** ```typescript import { test, expect } from "@playwright/test" test("服务器推送警报时显示通知", async ({ page }) => { const wsRoute = await page.routeWebSocket("**/ws/notifications", (ws) => { // 让应用发送初始握手消息 ws.onMessage((message) => { const data = JSON.parse(message) if (data.type === "subscribe") { ws.send(JSON.stringify({ type: "subscribed", channel: data.channel })) } }) // 短暂延迟后推送通知 setTimeout(() => { ws.send( JSON.stringify({ type: "notification", title: "Server Alert", body: "Deployment completed successfully", severity: "info", }) ) }, 500) }) await page.goto("/dashboard") // 验证通知出现在 UI 中 await expect(page.getByRole("alert")).toContainText("Deployment completed successfully") }) test("优雅地处理 WebSocket 服务器错误", async ({ page }) => { await page.routeWebSocket("**/ws", (ws) => { // 立即以错误码关闭 ws.close({ code: 1011, reason: "Internal server error" }) }) await page.goto("/chat") // 应用应显示重连提示，而非崩溃 await expect(page.getByText("Connection lost. Reconnecting...")).toBeVisible() }) ``` **JavaScript** ```javascript const { test, expect } = require("@playwright/test") test("服务器推送警报时显示通知", async ({ page }) => { await page.routeWebSocket("**/ws/notifications", (ws) => { ws.onMessage((message) => { const data = JSON.parse(message) if (data.type === "subscribe") { ws.send(JSON.stringify({ type: "subscribed", channel: data.channel })) } }) setTimeout(() => { ws.send( JSON.stringify({ type: "notification", title: "Server Alert", body: "Deployment completed successfully", severity: "info", }) ) }, 500) }) await page.goto("/dashboard") await expect(page.getByRole("alert")).toContainText("Deployment completed successfully") }) test("优雅地处理 WebSocket 服务器错误", async ({ page }) => { await page.routeWebSocket("**/ws", (ws) => { ws.close({ code: 1011, reason: "Internal server error" }) }) await page.goto("/chat") await expect(page.getByText("Connection lost. Reconnecting...")).toBeVisible() }) ``` ### 带修改的转发（中间人模式） **适用场景**：你需要连接到真实服务器，但拦截、修改或注入消息。 **避免场景**：完全模拟（不带 `connectToServer` 的 `routeWebSocket`）已经足够。 **TypeScript** ```typescript import { test, expect } from "@playwright/test" test("向真实数据流中注入一条伪造的高优先级消息", async ({ page }) => { await page.routeWebSocket("**/ws/feed", (ws) => { const server = ws.connectToServer() // 将所有来自服务器的消息转发给客户端，并注入额外消息 server.onMessage((message) => { ws.send(message) // 转发真实消息 }) // 将所有客户端消息转发给服务器 ws.onMessage((message) => { server.send(message) }) // 1 秒后注入一条合成消息 setTimeout(() => { ws.send( JSON.stringify({ type: "alert", priority: "high", text: "Injected test alert", }) ) }, 1000) }) await page.goto("/live-feed") await expect(page.getByText("Injected test alert")).toBeVisible() }) ``` **JavaScript** ```javascript const { test, expect } = require("@playwright/test") test("向真实数据流中注入一条伪造的高优先级消息", async ({ page }) => { await page.routeWebSocket("**/ws/feed", (ws) => { const server = ws.connectToServer() server.onMessage((message) => { ws.send(message) }) ws.onMessage((message) => { server.send(message) }) setTimeout(() => { ws.send( JSON.stringify({ type: "alert", priority: "high", text: "Injected test alert", }) ) }, 1000) }) await page.goto("/live-feed") await expect(page.getByText("Injected test alert")).toBeVisible() }) ``` ### Server-Sent Events（SSE）测试 **适用场景**：你的应用使用 `EventSource` 实现服务器到客户端的流式传输（实时日志、进度更新、新闻推送）。 **避免场景**：应用使用的是 WebSocket。SSE 基于 HTTP，拦截方式不同。 SSE 响应是标准的 HTTP 响应——使用 `page.route()` 拦截并返回一个流式响应即可。 **TypeScript** ```typescript import { test, expect } from "@playwright/test" test("SSE 实时日志流显示条目", async ({ page }) => { // 拦截 SSE 端点并返回受控事件 await page.route("**/api/logs/stream", async (route) => { const events = [ 'data: {"level":"info","message":"Server started"}\n\n', 'data: {"level":"warn","message":"High memory usage"}\n\n', 'data: {"level":"error","message":"Connection timeout"}\n\n', ] await route.fulfill({ status: 200, headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive", }, body: events.join(""), }) }) await page.goto("/admin/logs") await expect(page.getByText("Server started")).toBeVisible() await expect(page.getByText("High memory usage")).toBeVisible() await expect(page.getByText("Connection timeout")).toBeVisible() }) test("SSE 连接断开后自动重连", async ({ page }) => { let requestCount = 0 await page.route("**/api/events", async (route) => { requestCount++ if (requestCount === 1) { // 第一次请求：发送一条事件后立即关闭 await route.fulfill({ status: 200, headers: { "Content-Type": "text/event-stream" }, body: 'data: {"msg":"first"}\n\n', }) } else { // 重连：发送下一条事件 await route.fulfill({ status: 200, headers: { "Content-Type": "text/event-stream" }, body: 'data: {"msg":"reconnected"}\n\n', }) } }) await page.goto("/live") await expect(page.getByText("first")).toBeVisible() // EventSource 自动重连；验证应用能够处理 await expect(page.getByText("reconnected")).toBeVisible({ timeout: 10000 }) }) ``` **JavaScript** ```javascript const { test, expect } = require("@playwright/test") test("SSE 实时日志流显示条目", async ({ page }) => { await page.route("**/api/logs/stream", async (route) => { const events = [ 'data: {"level":"info","message":"Server started"}\n\n', 'data: {"level":"warn","message":"High memory usage"}\n\n', 'data: {"level":"error","message":"Connection timeout"}\n\n', ] await route.fulfill({ status: 200, headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive", }, body: events.join(""), }) }) await page.goto("/admin/logs") await expect(page.getByText("Server started")).toBeVisible() await expect(page.getByText("High memory usage")).toBeVisible() await expect(page.getByText("Connection timeout")).toBeVisible() }) ``` ### 基于轮询的实时测试 **适用场景**：你的应用使用 HTTP 轮询（setInterval + fetch）而非 WebSocket 或 SSE。 **避免场景**：应用使用的是 WebSocket 或 SSE——请使用上述模式。 **TypeScript** ```typescript import { test, expect } from "@playwright/test" test("轮询在每个时间间隔更新仪表盘数据", async ({ page }) => { let callCount = 0 await page.route("**/api/dashboard/stats", async (route) => { callCount++ const data = callCount === 1 ? { activeUsers: 100, revenue: 5000 } : { activeUsers: 142, revenue: 5250 } await route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify(data), }) }) await page.goto("/dashboard") // 第一次轮询结果 await expect(page.getByTestId("active-users")).toHaveText("100") // 等待第二次轮询更新 UI await expect(page.getByTestId("active-users")).toHaveText("142", { timeout: 15000 }) // 验证至少发出了 2 次请求 expect(callCount).toBeGreaterThanOrEqual(2) }) ``` **JavaScript** ```javascript const { test, expect } = require("@playwright/test") test("轮询在每个时间间隔更新仪表盘数据", async ({ page }) => { let callCount = 0 await page.route("**/api/dashboard/stats", async (route) => { callCount++ const data = callCount === 1 ? { activeUsers: 100, revenue: 5000 } : { activeUsers: 142, revenue: 5250 } await route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify(data), }) }) await page.goto("/dashboard") await expect(page.getByTestId("active-users")).toHaveText("100") await expect(page.getByTestId("active-users")).toHaveText("142", { timeout: 15000 }) expect(callCount).toBeGreaterThanOrEqual(2) }) ``` ### WebSocket 连接生命周期 **适用场景**：你需要验证应用能否正确处理连接、断开连接和重新连接。 **避免场景**：连接生命周期对用户不可见。 **TypeScript** ```typescript import { test, expect } from "@playwright/test" test("应用在 WebSocket 断开后重新连接", async ({ page }) => { let connectionCount = 0 await page.routeWebSocket("**/ws", (ws) => { connectionCount++ if (connectionCount === 1) { // 第一次连接：短暂保持后关闭 setTimeout(() => ws.close({ code: 1006, reason: "Abnormal closure" }), 500) } else { // 第二次连接（重连）：保持打开并响应 ws.onMessage((message) => { ws.send(JSON.stringify({ type: "pong" })) }) } }) await page.goto("/app") // 应用检测到断开连接并显示状态 await expect(page.getByText("Reconnecting...")).toBeVisible() // 应用重新连接，状态恢复正常 await expect(page.getByText("Connected")).toBeVisible({ timeout: 10000 }) expect(connectionCount).toBe(2) }) ``` **JavaScript** ```javascript const { test, expect } = require("@playwright/test") test("应用在 WebSocket 断开后重新连接", async ({ page }) => { let connectionCount = 0 await page.routeWebSocket("**/ws", (ws) => { connectionCount++ if (connectionCount === 1) { setTimeout(() => ws.close({ code: 1006, reason: "Abnormal closure" }), 500) } else { ws.onMessage((message) => { ws.send(JSON.stringify({ type: "pong" })) }) } }) await page.goto("/app") await expect(page.getByText("Reconnecting...")).toBeVisible() await expect(page.getByText("Connected")).toBeVisible({ timeout: 10000 }) expect(connectionCount).toBe(2) }) ``` ## 方案选择指南 | 场景 | 方案 | 理由 | | ----------------------------------------- | -------------------------------------------------------------- | ----------------------------------------------------------- | | 验证应用发送了正确的 WebSocket 消息 | `page.on('websocket')` + `ws.on('framesent')` | 观察而不拦截 | | 验证应用正确处理了服务器推送 | `page.routeWebSocket()` + 模拟服务器 | 完全控制"服务器"发送的内容 | | 使用真实服务器测试，但注入额外消息 | `routeWebSocket` + `connectToServer()` | 中间人模式：转发真实流量并注入额外内容 | | 测试 SSE 端点 | `page.route()` + `text/event-stream` 内容类型 | SSE 是 HTTP——标准路由拦截即可生效 | | 测试 HTTP 轮询 | `page.route()` + 每次调用返回不同的响应 | 递增计数器；每次返回不同的数据 | | 验证重连逻辑 | `routeWebSocket` 关闭第一次连接 | 模拟服务器故障，验证应用重试 | | 测试二进制 WebSocket 数据 | `ws.on('framereceived')`，检查 `frame.payload` 作为 Buffer | 在 Node.js 中，二进制帧以 `Buffer` 形式到达 | ## 反模式 | 不要这样做 | 问题 | 应该这样做 | | -------------------------------------------------------- | ------------------------------------------------------------------- | --------------------------------------------------------------------------- | | 使用 `page.waitForTimeout(3000)` 等待 WebSocket 消息 | 任意延迟；不稳定且缓慢 | `await expect(page.getByText('msg')).toBeVisible()` 或等待一个 Promise | | 在 `page.evaluate` 中直接构造 `WebSocket` | 你将失去 Playwright 的观察和路由能力 | 让应用创建自己的 WebSocket；通过 `routeWebSocket` 进行拦截 | | 在模拟中忽略 WebSocket 关闭码 | 应用对 1000（正常）与 1006（异常）的行为可能不同 | 使用正确的关闭码：`ws.close({ code: 1000 })` | | 针对真实的第三方服务器测试实时功能 | 不稳定、缓慢，且可能产生费用 | 模拟 WebSocket 或 SSE 端点 | | 在每个测试中都对原始 WebSocket 帧内容进行断言 | 测试与网络协议耦合；当负载格式变更时容易失效 | 在 UI 上断言——那是用户能看到的内容 | | 忘记区分二进制帧和文本帧 | `frame.payload` 可能是 `string` 或 `Buffer` | 检查帧类型，或统一使用 `String(frame.payload)` | ## 故障排除 | 症状 | 原因 | 解决方法 | | ------------------------------------------------- | ------------------------------------------------------------------ | ------------------------------------------------------------------------------------------- | | `page.on('websocket')` 从未触发 | WebSocket 在监听器绑定之前已经连接 | 在 `page.goto()` 之前注册监听器 | | `routeWebSocket` 未拦截 | URL 模式与实际 WebSocket URL 不匹配 | 在 DevTools 的 Network 标签页中检查 URL；更新 glob 模式 | | SSE 模拟一次性返回所有事件 | `route.fulfill` 同步发送响应体 | 如需真正的流式传输，使用真实服务器或通过 `page.evaluate` 分块发送响应，并加入暂停 | | WebSocket 消息已到达但 UI 未更新 | 应用异步处理消息；断言执行过早 | 使用 `await expect(...).toBeVisible()`，该方法会自动重试 | | 二进制帧显示为乱码文本 | 对二进制数据使用 `String(frame.payload)` 会产生乱码 | 将 `frame.payload` 作为 `Buffer` 处理并适当解码 | | 重连测试不稳定 | 应用使用指数退避策略；超时时间过短 | 增加断言超时：`toBeVisible({ timeout: 15000 })` | ## 相关文档 - [core/multi-user-and-collaboration.md](multi-user-and-collaboration.md)——依赖 WebSocket 实现实时同步的多用户测试 - [core/assertions-and-waiting.md](assertions-and-waiting.md)——用于异步 UI 更新的自动重试断言 - [core/when-to-mock.md](when-to-mock.md)——决定何时模拟 WebSocket 与使用真实服务器 - [core/debugging.md](debugging.md)——在 Playwright 跟踪中追踪 WebSocket 帧