SurfSense WebSocket 与实时功能测试实战Playwright 下 WebSocket / SSE 的监听、Mock 与重连测试【免费下载链接】SurfSenseOpen-source NotebookLM alternative. Research the open web with live data(Reddit, YT, IG, TikTok, Indeed, Google Search, Maps etc) through one platform, API or MCP server. Join our Discord: https://discord.gg/ejRNvftDp9项目地址: https://gitcode.com/GitHub_Trending/su/SurfSense导读实时交互聊天消息、通知推送、协作编辑、流式响应是现代 Web 应用的标配而 WebSocket 与 Server-Sent EventsSSE恰恰是这类功能最脆弱、最难稳定的测试点。本文基于 SurfSense 仓库内 .cursor/skills/playwright-testing/browser-apis/websockets.md 这一份 Playwright 测试技能文档系统讲解如何用 Playwright 监听 WebSocket 帧、注入模拟消息、构造可复用的 WebSocket Mock Fixture、测试实时通知与协作光标并顺带覆盖 SSE 流与断线重连场景。读完本文你将掌握一套可直接复制到 SurfSense 前端 E2E 测试surfsense_web/tests中的实时功能测试方法论并理解其背后的真实技术栈与配置依据。一、SurfSense 的实时技术栈为什么实时测试如此关键在进入具体 API 之前先交代清楚这套测试方法在 SurfSense 项目中的落点。从仓库文档 surfsense_web/content/docs/how-to/zero-sync.mdx 可以看到SurfSense 使用 Rocicorp Zero 实现实时数据同步zero-cache服务位于 PostgreSQL 与浏览器之间通过逻辑复制把数据库变更同步到本地 SQLite 副本浏览器则通过WebSocket连接 zero-cache从而让通知notifications、文档documents、聊天消息new_chat_messages、评论chat_comments以及会话协作状态chat_session_state实现“数据库一变更、前端立即更新”。与此同时聊天类接口采用 SSE 流式返回后端 surfsense_backend/app/services/streaming/envelope/sse.py 定义了format_sse()、format_done()与get_response_headers()返回的响应头正是Content-Type: text/event-stream、Cache-Control: no-cache、Connection: keep-alive与本文 SSE 测试小节中 Mock 的响应头完全一致该信封被 new_chat_routes.py、public_chat_routes.py 等路由复用。这意味着 SurfSense 的实时功能横跨两类协议Zero 的 WebSocket 同步通道与AI 回答的 SSE 流。二者的测试策略恰好都包含在本文技能文档中且 E2E 运行环境已就绪——surfsense_web/playwright.config.ts 中通过NEXT_PUBLIC_ZERO_CACHE_URL默认http://localhost:4848指向 zero-cache并设置了x-playwright-test: true请求头与setup登录项目测试可直接在surfsense_web下以pnpm exec playwright test运行。二、WebSocket 基础等待连接、监听消息、捕获发送帧Playwright 对 WebSocket 提供了第一等的原生支持page.waitForEvent(websocket)与page.on(websocket)。核心原则是在触发导航或交互之前先注册监听器否则会错过 WebSocket 建立瞬间的早期帧。2.1 等待 WebSocket 连接建立以下用例监听并等待聊天页的 WebSocket 连接随后校验其 URL 与首个发送帧test(chat connects via websocket, async ({ page }) { // Listen for WebSocket connection const wsPromise page.waitForEvent(websocket); await page.goto(/chat); const ws await wsPromise; expect(ws.url()).toContain(/ws/chat); // Wait for connection to be established await ws.waitForEvent(framesent); });关键点有两个ws.url()用于断言客户端连接到了正确的端点。在 SurfSense 中前端通过NEXT_PUBLIC_ZERO_CACHE_URL决定 WebSocket 目标因此断言时按环境灵活处理例如http://localhost:4848或经 Caddy 代理的${SURFSENSE_PUBLIC_URL}/zero见 zero-sync.mdx 的“Custom Domain / Reverse Proxy”一节。ws.waitForEvent(framesent)确保连接不仅建立、还完成了握手后的首次通信避免后续断言过早执行。2.2 持续监听接收帧framereceived若要收集服务端推送的全部消息使用page.on(websocket)在导航前挂上监听test(receives real-time updates, async ({ page }) { const messages: string[] []; // Set up listener before navigation page.on(websocket, (ws) { ws.on(framereceived, (frame) { messages.push(frame.payload as string); }); }); await page.goto(/dashboard); // Wait for some messages await expect.poll(() messages.length).toBeGreaterThan(0); // Verify message format const data JSON.parse(messages[0]); expect(data).toHaveProperty(type); });frame.payload通常为 JSON 字符串所以用JSON.parse还原结构。这里使用expect.poll详见同技能库的 .cursor/skills/playwright-testing/core/assertions-waiting.md轮询等待消息到达而不是sleep硬等保证了测试的稳定性与速度。2.3 捕获发送帧framesent反向验证“客户端发给服务端”的消息格式test(sends correct message format, async ({ page }) { const sentMessages: string[] []; page.on(websocket, (ws) { ws.on(framesent, (frame) { sentMessages.push(frame.payload as string); }); }); await page.goto(/chat); await page.getByLabel(Message).fill(Hello!); await page.getByRole(button, { name: Send }).click(); // Verify sent message await expect.poll(() sentMessages.length).toBeGreaterThan(0); const sent JSON.parse(sentMessages[sentMessages.length - 1]); expect(sent).toEqual({ type: message, content: Hello!, }); });这种“发送帧快照断言”非常适合守护消息协议一旦前端改了负载结构比如新增字段、改了字段名测试立即报警防止前后端契约悄悄漂移。三、Mocking WebSocket 消息从注入到完整 Fixture真实 WebSocket 服务在 CI 中往往不稳定、时序敏感因此技能文档给出了三档 Mock 方案从最轻量的运行时注入到可复用的 Fixture。3.1 通过 page.evaluate 注入消息如果应用把 WebSocket 实例暴露在window上测试用全局变量如window.chatSocket可以直接派发MessageEvent模拟服务端下行消息test(displays incoming chat message, async ({ page }) { await page.goto(/chat); // Wait for WebSocket to be ready await page.waitForFunction( () (window as any).chatSocket?.readyState 1, ); // Simulate incoming message await page.evaluate(() { const event new MessageEvent(message, { data: JSON.stringify({ type: message, from: Alice, content: Hello there!, }), }); (window as any).chatSocket.dispatchEvent(event); }); await expect(page.getByText(Alice: Hello there!)).toBeVisible(); });必须先用page.waitForFunction等待readyState 1WebSocket.OPEN这正是文档末尾“Anti-Patterns”表中第一条要规避的坑不等连接就绪就发消息事件会在监听器挂载前丢失。在 SurfSense 中若要在 E2E 里这样做需要先在应用代码里把 Zero 的同步连接或 socket 实例挂到window仅测试环境暴露可用x-playwright-test请求头或环境变量做开关避免污染生产构建。3.2 在应用层整体替换 WebSocketcontext.route()无法直接 fulfill WebSocket 升级请求Playwright 的路由拦截不覆盖 Upgrade 握手因此更彻底的做法是用page.addInitScript在页面脚本执行前替换全局WebSocket构造器test(mock websocket entirely, async ({ page, context }) { // Intercept the WebSocket upgrade await context.route(**/ws/**, async (route) { // For WebSocket routes, we cant fulfill directly // Instead, use page.evaluate to mock the client-side }); // Alternative: Mock at application level await page.addInitScript(() { const OriginalWebSocket window.WebSocket; (window as any).WebSocket function (url: string) { const ws { readyState: 1, send: (data: string) { console.log(WS Send:, data); }, close: () {}, addEventListener: () {}, removeEventListener: () {}, }; setTimeout(() ws.onopen?.(), 100); return ws; }; }); await page.goto(/chat); });addInitScript的注入时机早于任何页面脚本因此应用代码拿到的就是一个“永远已连接”的假 socket不依赖任何网络与后端服务测试完全离线可跑。注意假 socket 必须实现应用实际调用的接口send/close/addEventListener/onopen等否则会在运行时抛TypeError。3.3 WebSocket Mock Fixture监听与注入合二为一把“捕获发送帧”与“注入接收消息”封装成自定义 Fixture是技能文档推荐的生产级做法可跨用例复用// fixtures/websocket.fixture.ts import { test as base, Page } from playwright/test; type WsMessage { type: string; [key: string]: any }; type WebSocketFixtures { mockWebSocket: { injectMessage: (message: WsMessage) Promisevoid; getSentMessages: () PromiseWsMessage[]; }; }; export const test base.extendWebSocketFixtures({ mockWebSocket: async ({ page }, use) { const sentMessages: WsMessage[] []; // Capture sent messages await page.addInitScript(() { (window as any).__wsSent []; const OriginalWebSocket window.WebSocket; window.WebSocket function (url: string) { const ws new OriginalWebSocket(url); const originalSend ws.send.bind(ws); ws.send (data: string) { (window as any).__wsSent.push(JSON.parse(data)); originalSend(data); }; (window as any).__ws ws; return ws; } as any; }); await use({ injectMessage: async (message) { await page.evaluate((msg) { const event new MessageEvent(message, { data: JSON.stringify(msg), }); (window as any).__ws?.dispatchEvent(event); }, message); }, getSentMessages: async () { return page.evaluate(() (window as any).__wsSent || []); }, }); }, });这一版 Fixture 的高明之处在于不替换真实 WebSocket而是包一层代理window.WebSocket返回new OriginalWebSocket(url)的真实实例仅对send做旁路记录并保存实例到window.__ws。这样既能注入消息、又能读取真实发送帧还保留了与真实服务的连接能力。用法示例// Usage test(chat with mocked websocket, async ({ page, mockWebSocket }) { await page.goto(/chat); // Inject incoming message await mockWebSocket.injectMessage({ type: message, from: Bob, content: Hi!, }); await expect(page.getByText(Bob: Hi!)).toBeVisible(); // Send a reply await page.getByLabel(Message).fill(Hello Bob!); await page.getByRole(button, { name: Send }).click(); // Verify sent message const sent await mockWebSocket.getSentMessages(); expect(sent).toContainEqual( expect.objectContaining({ content: Hello Bob! }), ); });在 SurfSense 中这一模式可直接演化为zeroSync或liveChat专用 Fixture例如把injectMessage换成“注入一条notification消息触发 inbox 更新”“注入一条new_chat_messages变更触发消息列表刷新”等业务语义进一步贴近 zero-sync.mdx 中 Synced Tables 的清单。四、实时功能测试通知、数据刷新与协作编辑4.1 实时通知Live Notificationstest(displays live notification, async ({ page }) { await page.goto(/dashboard); // Simulate notification via WebSocket await page.evaluate(() { const event new MessageEvent(message, { data: JSON.stringify({ type: notification, title: New Order, message: Order #123 received, }), }); (window as any).notificationSocket.dispatchEvent(event); }); await expect(page.getByRole(alert)).toContainText(Order #123 received); });对应到 SurfSensenotifications表经 Zero 同步后驱动收件箱角标与通知弹层。测试时不必真的在后端插入一条通知记录直接在 socket 层注入等同负载即可验证 UI 渲染逻辑。4.2 实时数据更新Live Data Updatestest(updates stock price in real-time, async ({ page }) { await page.goto(/stocks/AAPL); const priceElement page.getByTestId(stock-price); const initialPrice await priceElement.textContent(); // Simulate price update await page.evaluate(() { const event new MessageEvent(message, { data: JSON.stringify({ type: price_update, symbol: AAPL, price: 150.25, }), }); (window as any).stockSocket.dispatchEvent(event); }); await expect(priceElement).not.toHaveText(initialPrice!); await expect(priceElement).toContainText(150.25); });这段代码演示了“先记录初始值、注入更新、断言值发生改变且为新值”的三段式写法。在 SurfSense 中类似场景包括文档处理进度条documents表的status列实时变化与自动化运行进度automation_runs薄列集同步重字段走 REST见 zero-sync.mdx 的 Synced Tables。4.3 协作编辑Collaborative Editingtest(shows collaborator cursor, async ({ page }) { await page.goto(/document/123); // Simulate another users cursor position await page.evaluate(() { const event new MessageEvent(message, { data: JSON.stringify({ type: cursor, userId: user-456, userName: Alice, position: { x: 100, y: 200 }, }), }); (window as any).docSocket.dispatchEvent(event); }); await expect(page.getByTestId(cursor-user-456)).toBeVisible(); await expect(page.getByText(Alice)).toBeVisible(); });注意这里用一个 socket 模拟“另一名用户”的输入。若你的实时协作建立在独立连接上技能库还提供了更贴近真实的多用户方案——见 .cursor/skills/playwright-testing/advanced/multi-user.md多 context 模拟多名真实用户同时操作适用于 SurfSense 中“谁在输入”等chat_session_state协作指示器的端到端验证。五、Server-Sent EventsSSE测试SurfSense 的 AI 聊天回答走 SSE 流式返回后端信封实现位于 surfsense_backend/app/services/streaming/envelope/sse.py每个数据块以data: json\n\n结尾、末尾发送data: [DONE]\n\n。Playwright 的page.route可以直接 Mock 这类流式端点。5.1 Mock 单个 SSE 更新test(receives SSE updates, async ({ page }) { // Mock SSE endpoint await page.route(**/api/events, (route) { route.fulfill({ status: 200, headers: { Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, }, body: data: {type:update,value:42}\n\n, }); }); await page.goto(/live-data); await expect(page.getByTestId(value)).toHaveText(42); });请对比后端 sse.py 的get_response_headers()两者在Content-Type、Cache-Control、Connection上完全对应body中的data: {...}\n\n也正是format_sse()的产出格式字符串直接拼data:前缀对象则先json.dumps。用与真实实现一致的 Mock才能让测试真正代表线上行为。5.2 模拟连续多个 SSE 事件test(handles multiple SSE events, async ({ page }) { await page.route(**/api/events, async (route) { const encoder new TextEncoder(); const events [ data: {count:1}\n\n, data: {count:2}\n\n, data: {count:3}\n\n, ]; route.fulfill({ status: 200, headers: { Content-Type: text/event-stream }, body: events.join(), }); }); await page.goto(/counter); // Should receive all events await expect(page.getByTestId(count)).toHaveText(3); });SSE 的边界是\n\n分隔的事件块浏览器EventSource会逐块解析。连续事件测试可用于验证 SurfSense 聊天界面对分片 token 流的拼接逻辑——真实场景中后端会按 token 分批format_sse()推送前端逐块渲染最后收到format_done()的[DONE]标记收尾。补充page.route的用法route.fulfill/route.continue/ 按请求体条件 Mock / 限流可进一步参考 .cursor/skills/playwright-testing/advanced/network-advanced.md它与本文互为补充WebSocket 无法被 route 直接拦截而 HTTP含 SSE则可以。六、断线与重连测试Reconnection Testing实时功能最容易被忽略的就是连接状态机断开时要有降级提示恢复后要能自动重连。6.1 模拟连接丢失test(handles connection loss gracefully, async ({ page }) { await page.goto(/chat); // Simulate connection close await page.evaluate(() { (window as any).chatSocket.close(); }); // Should show disconnected state await expect(page.getByText(Reconnecting...)).toBeVisible(); });6.2 模拟重连成功test(reconnects after connection loss, async ({ page }) { await page.goto(/chat); // Simulate disconnect await page.evaluate(() { (window as any).chatSocket.close(); }); await expect(page.getByText(Reconnecting...)).toBeVisible(); // Simulate reconnection await page.evaluate(() { const event new Event(open); (window as any).chatSocket { readyState: 1 }; (window as any).chatSocket.dispatchEvent?.(event); }); // Force component to re-check connection await page.evaluate(() { window.dispatchEvent(new Event(online)); }); await expect(page.getByText(Connected)).toBeVisible(); });第二段代码组合了三种手段派发open事件、替换 socket 的readyState、派发online事件浏览器在线状态以驱动前端的状态机从“Reconnecting...” 回到 “Connected”。在实际项目中请根据前端连接管理器的真实接口例如基于readyState轮询、open/close回调或 Zero 同步客户端的连接状态事件调整注入方式——关键是让应用看到“断开”与“恢复”两个可观测状态而不是只改 UI 文本。对于更真实的网络层断连如 context 级离线可使用context.setOffline(true)模拟整体断网相关模式同样收录在 .cursor/skills/playwright-testing/advanced/network-advanced.md 的 Offline Mode 一节。七、反模式清单Anti-Patterns技能文档用一张表总结了实时测试最容易踩的坑这里逐条结合 SurfSense 落地说明反模式问题正确做法不等待 WebSocket readyreadyState 1就发消息消息过早发送、事件在监听器挂载前丢失用page.waitForFunction(() socket.readyState 1)等待就绪直接测试真实 WebSocket 服务依赖真实 zero-cache/后端时序抖动导致 flaky用本文的addInitScript替换或 Fixture 注入模拟消息忽略连接状态只测“有数据时”的路径断开/重连逻辑无人守护显式测试 Connected / Disconnected / Reconnecting 三种状态监听器不做清理用例间事件监听互相泄漏导致内存与断言污染用 Fixture 封装监听生命周期或按用例作用域挂载/卸载配套的网络层反模式Mock 面过大、路由不清理、不分请求方法等参见 network-advanced.md轮询断言的正确姿势见 .cursor/skills/playwright-testing/core/assertions-waiting.md多用户协作的真实场景见 .cursor/skills/playwright-testing/advanced/multi-user.md。八、在 SurfSense 中落地配置与运行若要把上述模式写进 SurfSense 的 E2E 测试需要注意的工程前提均来自 surfsense_web/playwright.config.ts零配置即可连 Zero配置会依据环境变量自动设置NEXT_PUBLIC_ZERO_CACHE_URL本地默认http://localhost:4848启用代理时为${baseURL}/zero因此测试断言 WebSocket 目标时优先读取该变量而不是硬编码。身份与鉴权项目采用setup项目testMatch: /.*\.setup\.ts/先行登录产出playwright/.auth/user.json作为 chromium 项目的storageState测试请求统一携带x-playwright-test: true头。运行方式在surfsense_web下执行pnpm exec playwright testCI 环境会自动pnpm build pnpm start本地则走pnpm exec next dev。若需要跳过自动起服务可设置PLAYWRIGHT_NO_WEB_SERVER。后端侧的证据可对照SSE Mock 的响应头与 surfsense_backend/app/services/streaming/envelope/sse.py 的get_response_headers()保持一致的语义实时表的清单见 zero-sync.mdx 的 Synced Tables 表格。小结WebSocket 与 SSE 的实时测试没有银弹但有清晰的方法论连接阶段用waitForEvent(websocket)readyState 1等待就绪下行用MessageEvent注入或 Fixture 统一注入上行用framesent捕获并断言协议负载断连用close()open事件驱动状态机SSE则复用page.route与后端 sse.py 对齐的帧格式进行 Mock。把这套模式落地到 SurfSense 的 tests 目录中即可为通知同步、聊天消息、协作状态与 AI 流式回答构建出稳定、可离线、可复用的实时功能测试体系。【免费下载链接】SurfSenseOpen-source NotebookLM alternative. Research the open web with live data(Reddit, YT, IG, TikTok, Indeed, Google Search, Maps etc) through one platform, API or MCP server. Join our Discord: https://discord.gg/ejRNvftDp9项目地址: https://gitcode.com/GitHub_Trending/su/SurfSense创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考