Playwright进阶技巧：如何拦截和修改WebSocket通信（含代码示例）

Playwright进阶技巧：如何拦截和修改WebSocket通信（含代码示例）

在当今的Web应用开发中，实时通信已成为不可或缺的功能。从在线聊天系统到实时数据仪表盘，WebSocket协议因其全双工通信能力而广受欢迎。但对于测试工程师来说，如何有效测试这些实时功能却是一个挑战。本文将深入探讨如何使用Playwright这一现代化测试工具来拦截和修改WebSocket通信，为您的测试工作流带来全新可能。

1. WebSocket测试基础与挑战

WebSocket协议与传统的HTTP请求有着本质区别。它建立持久连接后，客户端和服务器可以随时互相发送消息，而不需要像HTTP那样每次都要建立新连接。这种特性虽然带来了实时性的优势，但也给测试带来了独特挑战：

• 连接稳定性：WebSocket连接可能因网络波动而中断
• 消息时序：消息到达的顺序可能影响应用状态
• 数据格式：二进制和文本消息需要不同处理方式
• 状态管理：连接状态（打开/关闭/错误）需要特别关注

Playwright作为新一代浏览器自动化工具，提供了全面的WebSocket拦截和修改API，让我们能够精确控制测试环境中的实时通信。

提示：在开始WebSocket测试前，确保您已熟悉基本的Playwright安装和使用方法。

2. 建立WebSocket连接拦截

Playwright允许我们在页面加载前就设置WebSocket路由，这对于测试那些一加载就建立WebSocket连接的应用特别有用。以下是一个基本示例：

import asyncio from playwright.async_api import async_playwright async def test_websocket_interception(): async with async_playwright() as p: browser = await p.chromium.launch() page = await browser.new_page() # 设置WebSocket路由拦截 async def handle_ws(ws): print(f"WebSocket opened at: {ws.url}") ws.on("framesent", lambda payload: print(f"Sent: {payload}")) ws.on("framereceived", lambda payload: print(f"Received: {payload}")) ws.on("close", lambda: print("WebSocket closed")) await page.route_web_socket("wss://example.com/ws", handle_ws) await page.goto("https://example.com") await asyncio.sleep(5) # 留出时间观察WebSocket活动 await browser.close()

这个示例展示了如何：

1. 监听WebSocket连接建立
2. 捕获发送和接收的消息帧
3. 跟踪连接关闭事件

3. 修改WebSocket消息内容

单纯的监听往往不能满足测试需求，我们经常需要修改传输中的消息。Playwright提供了灵活的消息修改机制：

async def test_modify_websocket_messages(): async with async_playwright() as p: browser = await p.chromium.launch() page = await browser.new_page() async def handle_ws(ws): # 修改接收到的消息 def on_message(payload): if payload == "original": return "modified" return payload ws.on("framereceived", lambda payload: on_message(payload)) # 修改发送的消息 def on_send(payload): if payload == "client_original": return "client_modified" return payload ws.on("framesent", lambda payload: on_send(payload)) await page.route_web_socket("wss://example.com/ws", handle_ws) await page.goto("https://example.com") await asyncio.sleep(3) await browser.close()

在这个进阶示例中，我们实现了：

• 条件性修改接收到的服务器消息
• 动态改写客户端发送的消息
• 保持非目标消息原样通过

4. 模拟完整WebSocket交互

有时我们需要完全模拟WebSocket服务器行为，而不依赖真实后端。Playwright同样支持这种场景：

async def test_full_websocket_simulation(): async with async_playwright() as p: browser = await p.chromium.launch() page = await browser.new_page() async def handle_ws(ws): # 模拟握手响应 await ws.send("Welcome to mock server!") # 处理客户端消息 def on_message(payload): if payload == "get_time": return str(datetime.now()) elif payload == "get_status": return "system_ok" return "unrecognized_command" ws.on("framereceived", lambda payload: ws.send(on_message(payload))) await page.route_web_socket("wss://example.com/ws", handle_ws) await page.goto("https://example.com") await asyncio.sleep(3) await browser.close()

这个模拟器实现了：

1. 连接建立时的欢迎消息
2. 根据客户端请求返回动态响应
3. 对未知命令的统一回复

5. 高级技巧：结合HAR文件记录WebSocket通信

虽然HAR文件主要记录HTTP流量，但我们可以利用Playwright的API结合HAR来增强WebSocket测试：

async def test_websocket_with_har(): async with async_playwright() as p: browser = await p.chromium.launch() context = await browser.new_context() page = await context.new_page() # 开始记录网络活动 await context.route_from_har("websocket_trace.har", update=True) # 设置WebSocket拦截 async def handle_ws(ws): ws.on("framesent", lambda p: print(f"Sent: {p}")) ws.on("framereceived", lambda p: print(f"Received: {p}")) await page.route_web_socket("wss://example.com/ws", handle_ws) await page.goto("https://example.com") await asyncio.sleep(5) # 保存记录 await context.close() await browser.close()

这种方法的价值在于：

• 同时捕获HTTP和WebSocket活动
• 创建可重复使用的测试场景
• 分析完整的网络交互时序

6. 实战：测试实时聊天应用

让我们看一个完整的实时聊天应用测试示例：

async def test_chat_app(): async with async_playwright() as p: browser = await p.chromium.launch() page = await browser.new_page() # 模拟聊天服务器 async def chat_server(ws): users = {} def on_message(payload): if payload.startswith("REGISTER:"): user_id = payload.split(":")[1] users[user_id] = ws return f"USER_{user_id}_REGISTERED" elif payload.startswith("MESSAGE:"): _, to_user, content = payload.split(":", 2) if to_user in users: users[to_user].send(f"MSG:{content}") return "MESSAGE_DELIVERED" return "UNKNOWN_COMMAND" ws.on("framereceived", lambda p: ws.send(on_message(p))) await page.route_web_socket("wss://chat.example.com/ws", chat_server) # 测试聊天功能 await page.goto("https://chat.example.com") await page.fill("#user-id", "test_user") await page.click("#connect-button") # 验证连接状态 await page.wait_for_selector("#status:has-text('Connected')") # 发送测试消息 await page.fill("#message-input", "Hello, World!") await page.click("#send-button") # 验证消息回显 await page.wait_for_selector("#message-list:has-text('Hello, World!')") await browser.close()

这个测试案例展示了：

1. 完整的用户注册流程模拟
2. 消息路由功能验证
3. 前端交互与WebSocket响应的集成测试

7. 调试技巧与最佳实践

在实施WebSocket测试时，以下技巧可以大幅提升效率：

常见问题排查表

性能优化建议

• 复用WebSocket路由处理函数
• 对大量消息使用批处理
• 在非必要情况下避免消息内容解析
• 使用try-catch处理网络异常

代码组织技巧

class WebSocketHelper: def __init__(self, page): self.page = page self.messages = [] async def setup_interception(self, url): async def handler(ws): ws.on("framereceived", self._record_message) await self.page.route_web_socket(url, handler) def _record_message(self, payload): self.messages.append({ "time": datetime.now(), "content": payload }) def get_messages(self): return self.messages

这种封装方式提供了：

• 集中化的消息存储
• 易于复用的拦截逻辑
• 清晰的状态管理

在实际项目中，我发现将WebSocket测试逻辑封装成辅助类可以显著提高代码可维护性。特别是在处理复杂协议或大量消息时，这种结构化的方法能让测试代码保持清晰。