wehub/skillhub-147-playwright-skill
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
skillhub-147-playwright-skill/core/websockets-and-realtime.md
T
wehub-skill-sync 5d144ced31 chore: import zh skill playwright-skill
2026-07-13 21:36:47 +08:00

22 KiB
Raw Permalink Blame History

WebSockets 与实时测试

适用场景：当你的应用使用 WebSocket、Server-Sent Events（SSE）或轮询来实现实时功能时——聊天、实时仪表盘、通知、协作编辑、股票行情、实时体育比分。 前置知识：core/assertions-and-waiting.md、core/fixtures-and-hooks.md

快速参考

// 监听 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

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

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

import { test, expect } from "@playwright/test"

test("等待服务器通过 WebSocket 发送确认", async ({ page }) => {
  // 创建一个 Promise，在收到特定消息时 resolve
  const ackPromise = new Promise<void>((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

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

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

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

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

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

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

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

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

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

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

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 })

相关文档

Reference in New Issue View Git Blame Copy Permalink