Playwright WebSocketRoute 深度指南：WebSocket 路由、Mock、拦截与源码实现

Playwright WebSocketRoute 深度指南WebSocket 路由、Mock、拦截与源码实现【免费下载链接】playwrightPlaywright is a framework for Web Testing and Automation. It allows testing Chromium, Firefox and WebKit with a single API.项目地址: https://gitcode.com/GitHub_Trending/pl/playwright本文围绕 Playwright 的WebSocketRoute类 API自 v1.48 引入展开系统讲解如何通过page.routeWebSocket()/browserContext.routeWebSocket()对页面中的 WebSocket 连接进行完整 Mock、消息拦截与双向转发控制并结合 WebSocketRoute 客户端实现、WebSocketRouteDispatcher 分发器实现 与 route-web-socket 测试套件 剖析其底层事件通道、二进制消息编解码与连接生命周期管理。读完本文你可以独立搭建可复现的 WebSocket 服务端 Mock并在真实服务器上实现消息改写、阻断与协议协商控制。一、WebSocketRoute 是什么路由设置后的“服务端替身”在 Playwright 中每当通过Page.routeWebSocket或BrowserContext.routeWebSocket设置了一条 WebSocket 路由对应的WebSocketRoute对象就会允许你像真正的服务端一样处理该 WebSocket 连接。原始 API 文档见 class-websocketroute.md。两条入口方法的签名在源码中定义如下JS/TypeScriptpage.tsawait page.routeWebSocket(url, handler); // url: URLMatchstring 支持 glob / RegExp / function // handler: (ws: WebSocketRoute) void | PromisevoidJavapage.routeWebSocket(url, handler)Pythonpage.route_web_socket(url, handler).NETpage.RouteWebSocketAsync(url, handler)。从源码结构看routeWebSocket的核心行为是将新的WebSocketRouteHandler以unshift插入路由列表头部后注册的规则优先匹配随后调用_updateWebSocketInterceptionPatterns把 URL 匹配模式下发到浏览器端见 page.ts 与 browserContext.ts。BrowserContext级别的路由则作用于该上下文内所有页面。二、Mock不连接真实服务器模拟整条 WebSocket 通信默认情况下被路由的 WebSocket不会连接到真实服务器。因此你可以完整地 Mock 掉 WebSocket 通信。下面是一个响应request消息并回复response的完整示例继承自原文档覆盖四种语言await page.routeWebSocket(wss://example.com/ws, ws { ws.onMessage(message { if (message request) ws.send(response); }); });page.routeWebSocket(wss://example.com/ws, ws - { ws.onMessage(frame - { if (request.equals(frame.text())) ws.send(response); }); });def message_handler(ws: WebSocketRoute, message: Union[str, bytes]): if message request: ws.send(response) await page.route_web_socket(wss://example.com/ws, lambda ws: ws.on_message( lambda message: message_handler(ws, message) ))def message_handler(ws: WebSocketRoute, message: Union[str, bytes]): if message request: ws.send(response) page.route_web_socket(wss://example.com/ws, lambda ws: ws.on_message( lambda message: message_handler(ws, message) ))await page.RouteWebSocketAsync(wss://example.com/ws, ws { ws.OnMessage(frame { if (frame.Text request) ws.Send(response); }); });为什么不调用connectToServer就是 Mock在 network.ts 中可以看到路由处理函数返回后客户端会执行_afterHandle()只要没有connectToServerthis._connected为 false就调用ensureOpened通道命令让注入到页面里的 mock 脚本把 WebSocket 直接标记为“已打开”状态从而保证页面里的ws.send()不会因没有真实连接而抛错async _afterHandle() { if (this._connected) return; // Ensure that websocket is open and can send messages without an actual server connection. // If this happens after the page has been closed, ignore the error. await this._channel.ensureOpened({}, kNoTimeout).catch(() {}); }这也是官方文档所述“Playwright assumes that WebSocket will be mocked, and opens the WebSocket inside the page automatically”的实现依据。处理 JSON 消息Mock 场景中最常见的是结构化消息原文档给出了 JSON 处理示例await page.routeWebSocket(wss://example.com/ws, ws { ws.onMessage(message { const json JSON.parse(message); if (json.request question) ws.send(JSON.stringify({ response: answer })); }); });page.routeWebSocket(wss://example.com/ws, ws - { ws.onMessage(frame - { JsonObject json new JsonParser().parse(frame.text()).getAsJsonObject(); if (question.equals(json.get(request).getAsString())) { MapString, String result new HashMap(); result.put(response, answer); ws.send(gson.toJson(result)); } }); });def message_handler(ws: WebSocketRoute, message: Union[str, bytes]): json_message json.loads(message) if json_message[request] question: ws.send(json.dumps({ response: answer })) await page.route_web_socket(wss://example.com/ws, lambda ws: ws.on_message( lambda message: message_handler(ws, message) ))await page.RouteWebSocketAsync(wss://example.com/ws, ws { ws.OnMessage(frame { using var jsonDoc JsonDocument.Parse(frame.Text); JsonElement root jsonDoc.RootElement; if (root.TryGetProperty(request, out JsonElement requestElement) requestElement.GetString() question) { var response new Dictionarystring, string { [response] answer }; string jsonResponse JsonSerializer.Serialize(response); ws.Send(jsonResponse); } }); });语言差异提示JS 与 Python 的onMessagehandler 直接接收原始消息string | bytes/Buffer而 Java 与 C# 的 handler 接收的是WebSocketFrame对象需要调用frame.text()/frame.Text取文本载荷、frame.binary()/frame.Binary取二进制载荷其定义见 class-websocketframe.md该类自 v1.9 起存在仅用于 C# 与 Java。三、InterceptingconnectToServer 之后的双向转发与拦截如果你希望连接真实服务器但想在中间截获并修改或阻断消息可以调用WebSocketRoute.connectToServer。它返回一个“服务端侧”的WebSocketRoute实例你可以向它发消息或处理来自服务器的消息。修改页面发给服务器的消息以下示例改写页面上行消息而服务器下行消息保持默认转发await page.routeWebSocket(/ws, ws { const server ws.connectToServer(); ws.onMessage(message { if (message request) server.send(request2); else server.send(message); }); });page.routeWebSocket(/ws, ws - { WebSocketRoute server ws.connectToServer(); ws.onMessage(frame - { if (request.equals(frame.text())) server.send(request2); else server.send(frame.text()); }); });def message_handler(server: WebSocketRoute, message: Union[str, bytes]): if message request: server.send(request2) else: server.send(message) def handler(ws: WebSocketRoute): server ws.connect_to_server() ws.on_message(lambda message: message_handler(server, message)) await page.route_web_socket(/ws, handler)def message_handler(server: WebSocketRoute, message: Union[str, bytes]): if message request: server.send(request2) else: server.send(message) def handler(ws: WebSocketRoute): server ws.connect_to_server() ws.on_message(lambda message: message_handler(server, message)) page.route_web_socket(/ws, handler)await page.RouteWebSocketAsync(/ws, ws { var server ws.ConnectToServer(); ws.OnMessage(frame { if (frame.Text request) server.Send(request2); else server.Send(frame.Text); }); });转发语义onMessage 会“接管”某个方向连接服务器之后所有消息默认在页面与服务器之间自动双向转发。但存在两条明确的规则原文档核心要点也与源码实现一一对应在原始路由上调用onMessage后页面到服务器的消息不再自动转发必须由 handler 自行处理例如调用server.send显式转发在服务端路由上调用onMessage后服务器到页面的消息不再自动转发必须由 handler 接管。这两条规则可以直接在 network.ts 的通道事件处理中得到印证this._channel.on(messageFromPage, ({ message, isBase64 }) { if (this._onPageMessage) this._onPageMessage(isBase64 ? Buffer.from(message, base64) : message); else if (this._connected) this._channel.sendToServer({ message, isBase64 }, kNoTimeout).catch(() {}); }); this._channel.on(messageFromServer, ({ message, isBase64 }) { if (this._onServerMessage) this._onServerMessage(isBase64 ? Buffer.from(message, base64) : message); else this._channel.sendToPage({ message, isBase64 }, kNoTimeout).catch(() {}); });即有 handler 走 handler没有 handler 且已连接则自动转发到对端。关闭事件closePage/closeServer也遵循同样的“handler 优先否则自动向对端传播”模式。双向阻断消息下面这个示例在两个方向都调用onMessage因此完全不保留自动转发用于阻断特定消息await page.routeWebSocket(/ws, ws { const server ws.connectToServer(); ws.onMessage(message { if (message ! blocked-from-the-page) server.send(message); }); server.onMessage(message { if (message ! blocked-from-the-server) ws.send(message); }); });page.routeWebSocket(/ws, ws - { WebSocketRoute server ws.connectToServer(); ws.onMessage(frame - { if (!blocked-from-the-page.equals(frame.text())) server.send(frame.text()); }); server.onMessage(frame - { if (!blocked-from-the-server.equals(frame.text())) ws.send(frame.text()); }); });def ws_message_handler(server: WebSocketRoute, message: Union[str, bytes]): if message ! blocked-from-the-page: server.send(message) def server_message_handler(ws: WebSocketRoute, message: Union[str, bytes]): if message ! blocked-from-the-server: ws.send(message) def handler(ws: WebSocketRoute): server ws.connect_to_server() ws.on_message(lambda message: ws_message_handler(server, message)) server.on_message(lambda message: server_message_handler(ws, message)) await page.route_web_socket(/ws, handler)await page.RouteWebSocketAsync(/ws, ws { var server ws.ConnectToServer(); ws.OnMessage(frame { if (frame.Text ! blocked-from-the-page) server.Send(frame.Text); }); server.OnMessage(frame { if (frame.Text ! blocked-from-the-server) ws.Send(frame.Text); }); });注意 Python 侧的函数参数命名要刻意区分两个WebSocketRoute原始路由与服务端路由否则闭包内会引用错对象。四、API 方法逐项详解以下各方法均自 v1.48 引入protocols除外见第六节与 class-websocketroute.md 保持一一对应。close(options)关闭 WebSocket 连接的一侧WebSocketRoute.close关闭 WebSocket 连接的一侧接受两个可选参数参数类型说明codeint可选的 WebSocket 关闭码reasonstring可选的关闭原因文本从 源码 看在原始路由页面侧上调用close()实际会向closePage通道发送{ code, reason, wasClean: true }而在服务端路由上调用则走closeServer通道。connectToServer()建立到真实服务器的连接返回服务端侧的WebSocketRoute实例可向它发消息或处理来自服务器的消息。默认被路由的 WebSocket 不连接服务器以便 Mock 整条通信调用此方法后即连接真实服务器。连接后的转发行为服务器消息自动转发到页面除非在服务端路由上调用了onMessage页面WebSocket.send()的消息自动转发到服务器除非在原始路由上调用了onMessage。重复调用会抛错源码 中connectToServer()先检查this._connected若已连接则抛出Already connected to the server对应测试用例should throw when connecting twiceroute-web-socket.spec.ts。服务端路由上的connectToServer不可用客户端会直接抛出connectToServer must be called on the page-side WebSocketRoute见 network.ts这是对 API 使用方向的源码级保护。onMessage(handler)接管某一方向的消息在原始路由上调用处理页面发出的消息。你可以用send直接应答页面、用connectToServer返回的服务端连接转发、或做其他处理。一旦调用消息不再自动转发到服务器或页面需要手动调用send。重复调用会覆盖旧 handler源码中 handler 以单一字段_onPageMessage/_onServerMessage存储直接赋值覆盖见 network.ts。handler 签名按语言区分JS / Pythonhandler(message: string | Buffer) any直接接收原始消息C# / Javahandler(frame: WebSocketFrame)通过text()/binary()区分文本帧与二进制帧。send(message)向 WebSocket 发送消息message类型为string | Buffer。在原始路由上调用会把消息发给页面在connectToServer返回的服务端路由上调用则发给服务器。二进制编解码在 源码 中处理字符串消息以isBase64: false直接透传Buffer消息先toString(base64)再发送接收侧再按isBase64标志还原为Buffer。这意味着send天然支持文本帧与二进制帧两种载荷。onClose(handler)处理 WebSocket 关闭onClose用于处理WebSocket.close事件。默认行为是连接任一侧页面或服务器关闭时另一侧也会被自动关闭一旦设置了onClosehandler默认的关闭转发被禁用由 handler 自行决定是否向对端传播关闭。handler 接收可选的关闭码与关闭原因JS / Python(code?: int, reason?: string) anyJava(Integer code, String reason)可能为nullC#(int? code, string? reason)。url()页面中创建的 WebSocket 的 URL返回字符串即路由命中的 WebSocket 实际 URL。实现上直接读取初始化数据this._initializer.url见 network.ts。五、路由匹配机制与底层注入原理客户端WebSocketRouteHandler 与模式匹配WebSocketRouteHandler 封装了routeWebSocket的 URL 与 handlerglob 模式提前校验构造函数中若url是字符串会立即调用resolveGlobToRegexPattern验证使非法模式在page.routeWebSocket()调用点就抛错而不是延迟到连接时才失败源码注释明确说明了这一设计意图baseURL 支持相对 URL如/ws会结合 context 的baseURL解析测试用例should work with relative WebSocket URL与should work with baseURL覆盖了这一行为模式序列化下发prepareInterceptionPatterns把所有 handler 的 URL 模式序列化为浏览器端拦截模式若任一 handler 的模式为空匹配全部则整体下发**/*。服务端WebSocketRouteDispatcher 与页面注入webSocketRouteDispatcher.ts 揭示了完整的拦截链路安装拦截install静态方法在BrowserContext上通过exposeBinding(__pwWebSocketBinding)注入页面通信入口并通过addInitScript注入由injected/webSocketMock生成的 mock 脚本即页面内对WebSocket构造函数的劫持。页面中的 WebSocket 创建事件会携带id、url、protocols上报。路由归属判断收到onCreate后服务端先用matchesPattern依次尝试页面级与上下文级的拦截模式_webSocketInterceptionPatterns命中则创建WebSocketRouteDispatcher并通知客户端派发 handler未命中则下发passthrough让页面直连真实服务器。消息转发页面消息onMessageFromPage、服务器消息onMessageFromServer、两侧关闭事件onClosePage/onCloseServer都通过 binding 上报到 dispatcher再由客户端WebSocketRoute按“handler 优先、否则自动转发”的规则处理。执行上下文失效保护dispatcher 监听页面/框架的InternalFrameNavigatedToNewDocument、FrameDetached、Page.Close、Page.Crash事件一旦发生认为 mock WebSocket 已无通信能力会派发closePage/closeServerwasClean: true对应测试should emit close upon frame navigation、should emit close upon frame detach与should not throw after page closure。单客户端约束install中若发现同一 context 已有其他连接在路由 WebSocket会抛出Another client is already routing WebSockets——即一个浏览器上下文的 WebSocket 路由只能由一个 Playwright 连接管理。六、protocols()自 v1.60 起的子协议协商WebSocketRoute.protocols自v1.60引入返回页面通过WebSocket构造函数 第二个参数请求的子协议列表对应Sec-WebSocket-Protocol请求头未指定任何协议时返回空数组。服务端路由上同样可以读取源码中protocols()在原始路由与服务端侧对象上均实现了相同的[...this._initializer.protocols]逻辑见 network.ts 与 L545-L547。典型用法是根据子协议选择处理策略不支持则用标准关闭码1002Protocol Error关闭await page.routeWebSocket(wss://example.com/ws, ws { if (ws.protocols().includes(chat.v2)) ws.onMessage(message ws.send(JSON.stringify({ version: 2, echo: message }))); else ws.close({ code: 1002, reason: Unsupported protocol }); });page.routeWebSocket(wss://example.com/ws, ws - { if (ws.protocols().contains(chat.v2)) { ws.onMessage(frame - ws.send(v2: frame.text())); } else { ws.close(1002, Unsupported protocol); } });async def handler(ws: WebSocketRoute): if chat.v2 in ws.protocols: ws.on_message(lambda message: ws.send(fv2:{message})) else: await ws.close(code1002, reasonUnsupported protocol) await page.route_web_socket(wss://example.com/ws, handler)def handler(ws: WebSocketRoute): if chat.v2 in ws.protocols: ws.on_message(lambda message: ws.send(fv2:{message})) else: ws.close(code1002, reasonUnsupported protocol) page.route_web_socket(wss://example.com/ws, handler)await page.RouteWebSocketAsync(wss://example.com/ws, ws { if (ws.Protocols.Contains(chat.v2)) ws.OnMessage(frame ws.Send($v2:{frame.Text})); else ws.CloseAsync(new() { Code 1002, Reason Unsupported protocol }); });注意 Python 的protocols是属性ws.protocols而非方法与其他语言的方法调用形式不同真实连接时测试用例should pass through the required protocol验证了协议会在握手时正确透传给服务器。七、测试用例佐证的行为边界tests/library/route-web-socket.spec.ts 为上述文档语义提供了完整的行为验证可以据此确认适用边界测试用例验证的语义should work with text message文本消息往返should work with binaryTypeblob/arraybuffer页面不同binaryType下二进制消息正确到达 handlershould work when connection errors out上游连接失败时的行为should work with client-side close页面侧ws.close的关闭传播should observe upstream handshake failure when connectToServer is used拦截模式下握手失败可被观察should observe multiple concurrent routed WebSockets with connectToServer多个并发路由连接互不串扰依赖 dispatcher 的_idToDispatcher映射should pattern matchURL 模式匹配globshould emit close upon frame navigation/frame detach帧导航/分离时自动派发 closeshould route on contextBrowserContext.routeWebSocket生效于上下文内页面should not throw when connecting twice的对应用例二次connectToServer抛错should work with baseURL/baseURL regardless of scheme casing相对 URL 与大小写不敏感匹配should expose protocols to the route handler/on server-side routeprotocols在两侧均可用此外tests/library/multiclient.spec.ts 验证了多客户端场景下路由的共存约束与 dispatcher 中 “Another client is already routing WebSockets” 的源码约束相呼应。八、实用建议与限制总结Mock 与拦截的切换点只有一个是否在 handler 中调用connectToServer。不调用则 Playwright 自动把页面内 WebSocket 置为打开态ensureOpened整条链路脱离真实服务器调用了则进入真实握手默认双向转发onMessage/onClose按方向逐步接管。handler 必须显式转发一旦在某一方向设置了onMessage该方向消息即停止自动转发忘记调用对端send会导致消息静默丢失send内部.catch(() {})吞掉了传输错误不会主动报错。适用前提routeWebSocket需要 v1.48protocols需要 v1.60Java/C# 的消息载荷需经WebSocketFrame间接读取同一浏览器上下文的 WebSocket 路由仅能由一个 Playwright 客户端连接建立。生命周期注意页面导航、帧分离或页面关闭都会触发路由侧的关闭派发编写测试时不应假设导航后 mock 连接仍然存活。综合来看WebSocketRoute以“页面侧路由 服务端侧路由”的双实例模型配合按方向可接管的转发规则覆盖了从纯 Mock 到真实链路上的细粒度拦截的完整光谱是 Playwright 处理实时通信类页面测试与自动化如聊天协议、行情推送、协同编辑同步的核心能力。【免费下载链接】playwrightPlaywright is a framework for Web Testing and Automation. It allows testing Chromium, Firefox and WebKit with a single API.项目地址: https://gitcode.com/GitHub_Trending/pl/playwright创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考