首页 / 资讯中心 / 文章详情

Selenium自动化测试中WebSocket消息监听:用Playwright旁路实现

Selenium自动化测试中WebSocket消息监听:用Playwright旁路实现 ★ FEATURED ARTICLE
做 Selenium 自动化时有个绕不开的痛点WebDriver 协议只能看到 DOM、请求和响应WebSocket 消息对测试框架来说是透明的。页面究竟收到多少条推送、消息内容是否符合预期、二进制数据有没有丢帧——这些想验证标准 Selenium API 完全做不到。上一篇讲过大致方向这一篇把“用 Playwright 旁路监听 Selenium 自动化测试中的 WebSocket 消息”的方案完整落地从架构设计、控制模式选型到可跑通的全套代码以及我实际项目中踩过的坑。适合已经在用 Selenium 维护测试集、又恰好需要校验实时推送场景的人。哪怕你之前没碰过 Playwright照着本文也能把监听器搭起来。1. 为什么 Selenium 测不到 WebSocket而 Playwright 能1.1 WebDriver 的协议边界只能看“表面”很多人第一次遇到这个问题时会愣一下Selenium 不是能执行 JavaScript 吗用execute_script拿到 WebSocket 对象、再包装一下不就行了理论上确实可以比如把window.WebSocket替换成一个记录消息的代理类。但真在测试里这么干你很快就会撞上几堵墙页面里的 WebSocket 连接可能发生在脚本注入之前这些早期消息拿不到前端框架如果直接引用原生WebSocket构造器替换window.WebSocket不一定生效WebSocket实例上的onmessage回调可以偷听但服务器主动推送的二进制帧、CLOSE 帧、PING/PONG 帧你只能拿到一部分更麻烦的是真实项目里 WebSocket 服务端可能不止一个连接你没法轻易把页面里所有 Socket 的状态汇总起来。这些问题的根源在 WebDriver 协议本身。WebDriver 是面向 DOM 和用户代理操作的协议它没有定义“读取帧数据”的能力。Selenium 作为 WebDriver 客户端自然没有现成 API。Chrome 浏览器本身有一套更底层的 DevTools 协议CDP里面有Network.webSocketFrameReceived、Network.webSocketFrameSent这类事件能拿到完整的帧内容但 Selenium 只封装了 CDP 的一小部分WebSocket 帧监听恰恰没公开。1.2 替代方案横向对比为什么绕道 Playwright为什么最后选了 Playwright而不是更直接的抓包代理或者裸写 CDP我把几种方案放在一起对比过方案能拿到什么改造成本典型问题代理抓包如 mitmproxy / BrowserMob全部 HTTP/WS 流量高改代理、装证书、处理 HTTPS 证书受信任问题WS 升级后的二进制帧解析麻烦且本地代理对性能影响明显浏览器扩展注入通过 content_script 包装 WebSocket中要适配前端构建产物页面 CSP 限制多生产构建可能失效自己写 CDP 客户端全部 CDP 事件高要自己维护 WebSocket 客户端、事件订阅、重连Playwright 已经封装了同样的事重复造轮子Playwright 旁路监听page.on(websocket) 帧事件低一个客户端库搞定需要接受“多一个浏览器进程/连接”的复杂度Playwright 最吸引我的一点是它原生支持page.on(websocket)并且每个 WebSocket 对象上都有framereceived、framesent事件。对于自动化测试来说这意味着监听代码可以非常短而且能拿到文本帧和二进制帧。相比抓包代理Playwright 不需要给浏览器配代理、不需要处理证书信任代码侵入也最小。你可以把它理解成测试里的“旁路取证设备”Selenium 负责操作页面Playwright 负责记录页面背后的 WS 消息两者各管一摊互不干扰。2. 旁路监听的总体架构谁启动浏览器、谁拿消息2.1 一条消息从服务器到断言的完整链路先建立一个整体画面。假设你要测试的是一个带实时行情面板的页面页面通过 WebSocket 订阅价格变化你要在自动化测试里断言“收到过 seq3 的行情消息”。用这套旁路方案完整链路是这样的Playwright 启动一个带远程调试端口的 Chromium 实例Selenium 通过debuggerAddress连接上这个已经跑起来的浏览器Selenium 驱动页面打开被测系统页面 JS 里创建 WebSocket 连接Playwright 提前注册好的page.on(websocket)捕获到新建的 WebSocket 对象再给这个对象注册framereceived监听服务器推送的每条帧进入 Playwright 的回调函数回调把内容丢进一个线程安全的队列Selenium 测试代码从队列里取消息做关键字/字段断言。值得强调的是第 4 步的“提前”。如果 Playwright 是在 WebSocket 握手之后才开始监听你只能收到握手完成后到达的帧如果监听器还没创建整个连接都收不到。所以架构上必须保证监听器先启动Selenium 后操作页面。2.2 三种浏览器控制模式怎么选浏览器到底由谁来启动是这套方案里最容易翻车的决策点。我试过三种模式这里直接给结论控制模式浏览器启动方Selenium 接入方式优点限制ASelenium 启动Playwright 连接Selenium带--remote-debugging-port正常 new Driver对旧代码改动最小Playwright 需要处理“浏览器已经存在”的 attach 逻辑新开的 tab 需要额外注册监听BPlaywright 启动Selenium 连接Playwright持久化上下文 调试端口debuggerAddress回连浏览器生命周期归监听器管理新页面统一捕获Selenium 不能主动决定浏览器启动参数环境初始化顺序要固定C双浏览器 CDP 直连各自独立完全不共享隔离性最好拿不到真实被测页面里的 WS 连接失去监听意义实际操作中我推荐 B。理由很现实Selenium 在模式 A 下启动浏览器后Playwright 需要靠connect_over_cdp去连连线过程会多一层复杂度。而模式 B 里Playwright 用持久化上下文启动浏览器等于天然拥有context.on(page)能力——不管 Selenium 后续driver.get打开几个标签页每个新页面都会触发这个事件我只需要在这个事件里统一挂 WebSocket 监听器。消息漏掉的可能性小很多。补充一句如果你是在别人写好的超大测试集里加监听不想动原有 Selenium 启动逻辑那模式 A 也完全可以。只是下面代码示例我按模式 B 来写因为它的监听覆盖最全。2.3 消息队列跨线程不丢消息的地基Playwright 的事件回调跑在它自己的事件循环里而 Selenium 测试代码跑在主测试线程。如果回调里直接往一个普通list写消息另一个线程同时读会出现两种典型问题要么读到不完整的半条消息要么两个线程同时操作导致崩溃。Python 标准库queue.Queue就是为这种情况准备的——内部自带锁put和get都是线程安全的。我在项目里一般会把队列包一层做成一个极简的消息总线import queue import threading class WsMessageBus: def __init__(self, maxsize10000): self._queue queue.Queue(maxsizemaxsize) self._lock threading.Lock() def put(self, message): try: self._queue.put_nowait(message) except queue.Full: # 队列满就是消息消费跟不上丢旧保新或直接丢弃并记录 pass def drain(self): messages [] while True: try: messages.append(self._queue.get_nowait()) except queue.Empty: break return messages def wait_for(self, keyword, timeout5): # 带超时的关键词匹配测试断言用 deadline threading.Timer(timeout, lambda: None) deadline.start() while deadline.is_alive(): messages self.drain() for msg in messages: if keyword in msg: deadline.cancel() return msg time.sleep(0.1) raise AssertionError(ftimeout waiting for {keyword})队列容量要设上限否则长时间测试里如果没人消费消息回调把内存堆爆只是时间问题。这个我后面在踩坑部分还会展开。3. 从零实现Playwright 拉浏览器、Selenium 连浏览器3.1 环境准备装什么、版本怎么配先列依赖pip install playwright selenium websockets pytest playwright install chromium虽然 Playwright 默认会下载它自己的 Chromium但为了让 Selenium 能无缝连接同一个浏览器实例我建议让 Playwright 直接使用系统安装的 Chrome用channelchrome指定。这样你在 Selenium 里看到的浏览器就是你日常用的 Chrome不会出现调试端口起了一个但两边版本对不上的问题。版本匹配是最大的坑。Selenium 连接已启动的 Chrome 时依然需要 chromedriver。chromedriver 的大版本必须和 Chrome 版本一致否则会报session not created: This version of ChromeDriver only supports Chrome version X。你可以用下面的命令快速确认版本google-chrome --version # Linux # 或者 Windows 下看浏览器“关于”页面 chromedriver --version两个版本对不上就先升级或降级其中一边。用 Selenium 4 的 Selenium Manager 有时能自动拉取匹配驱动但自动拉取也可能拉到不匹配的版本所以我在正式跑之前都会手动核对一次。3.2 第一步用 Playwright 启动带调试端口的 Chrome代码用 Python 的 Playwright 同步 API原因只有一个同步 API 在复杂测试流程里更好控制事件回调注册的时机更直观。import time import queue import json from playwright.sync_api import sync_playwright WS_QUEUE queue.Queue(maxsize10000) def start_browser_and_listener(debug_port9222, user_data_dir./ws_playwright_profile): p sync_playwright().start() context p.chromium.launch_persistent_context( user_data_diruser_data_dir, channelchrome, headlessFalse, args[ f--remote-debugging-port{debug_port}, --no-sandbox, --disable-dev-shm-usage, ], ) def wire_page(page): # 对新出现的页面统一挂上 WebSocket 监听 page.on(websocket, wire_ws) print(f[listener] wiring page: {page.url}) def wire_ws(ws): print(f[listener] new websocket: {ws.url}) ws.on(framereceived, on_ws_frame) def on_ws_frame(payload): # 不同 Playwright 版本对帧数据的封装略有差异这里做防御式解析 if isinstance(payload, dict): raw payload.get(payload) else: raw payload if isinstance(raw, bytes): text raw.decode(utf-8, errorsreplace) elif isinstance(raw, str): text raw else: text repr(raw) try: WS_QUEUE.put_nowait(text) except queue.Full: # 队列满了就丢弃最旧的一条保证新消息有位置 try: WS_QUEUE.get_nowait() WS_QUEUE.put_nowait(text) except queue.Empty: pass # 监听将来所有新页面 context.on(page, wire_page) # 处理已经存在的页面 for page in context.pages: wire_page(page) return context, p这里有几个细节需要解释launch_persistent_context比launch更适合这个场景。它把浏览器会话持久化到一个用户数据目录Selenium 连接后操作的所有状态都留在同一个浏览器实例里。如果你用launch新建一个默认 profileSelenium attach 过去时可能会出现 profile 目录不一致的诡异问题。channelchrome指定用本机安装的 Chrome而不是 Playwright 自带的 Chromium。如果你机器上没有 Chrome可以把这行去掉改用它自带的 Chromium但那样 Selenium 连接时依赖的 Chromedriver 就得和 Chromium 版本严格对齐麻烦一些。--remote-debugging-port9222是整套方案的钥匙。没有这个参数Selenium 无法通过 CDP 连上来。启动完成后先访问 http://127.0.0.1:9222/json/version 看一眼如果能看到webSocketDebuggerUrl字段说明端口正常开放了。这一步可以用来排查后续 Selenium 连不上的问题。3.3 第二步Selenium 通过 debuggerAddress 回连浏览器已经在跑了Selenium 侧就非常简单from selenium import webdriver from selenium.webdriver.chrome.options import Options chrome_options Options() chrome_options.debugger_address 127.0.0.1:9222 driver webdriver.Chrome(optionschrome_options) driver.get(https://example.com) # 这里页面里创建的任何 WebSocket 都会被 Playwright 捕获注意debugger_address只写ip:port不要带http://。Selenium 4 支持直接设置这个属性如果你用的是旧版 Selenium可以用chrome_options.add_experimental_option(debuggerAddress, 127.0.0.1:9222)效果一样。这一步执行时如果报错优先排查端口是否真的开放其次检查 chromedriver 和 Chrome 版本。还有一个容易忽略的点Chrome 启动参数里如果没有--remote-allow-origins*某些版本下 Selenium 通过 CDP 连接时会被跨源策略拦掉此时可以在 Playwright 的args里加上这个参数再启动。3.4 第三步模拟 WebSocket 服务与断言为了让你能直接跑一遍我提供一个最小可运行的模拟场景一个本地 WebSocket 服务每隔 0.2 秒推送一条 JSON 消息一个简单测试页面页面加载后用 JS 连接这个 WebSocket并把消息渲染到 DOM 上Selenium 打开页面Playwright 在后台记录所有 WS 帧断言关键字段。先起一个 WebSocket 服务端用websockets库import asyncio import json import time import threading import websockets def run_ws_server(port8765): async def handler(websocket): for i in range(5): message {seq: i, price: 100.0 i, ts: time.time()} await websocket.send(json.dumps(message)) await asyncio.sleep(0.2) async def main(): async with websockets.serve(handler, 127.0.0.1, port): # 挂起服务直到进程退出 await asyncio.Future() thread threading.Thread(targetlambda: asyncio.run(main()), daemonTrue) thread.start() return thread这里用线程跑服务是因为测试进程里同时有 Selenium 和 Playwright主线程需要继续执行测试逻辑。被测页面我直接用 data URL 内嵌避免依赖文件读取HTML_PAGE !DOCTYPE html html body div idmessages/div script const ws new WebSocket(ws://127.0.0.1:8765); ws.onmessage (event) { const div document.getElementById(messages); div.textContent event.data; }; /script /body /html测试主流程def test_ws_capture_via_playwright(): run_ws_server() context, p start_browser_and_listener() chrome_options Options() chrome_options.debugger_address 127.0.0.1:9222 driver webdriver.Chrome(optionschrome_options) try: driver.get(data:text/html;charsetutf-8, HTML_PAGE) # 等待 Playwright 捕获到消息 found wait_for_ws_message(WS_QUEUE, price, timeout5) assert json.loads(found[payload])[seq] 0 # 你甚至可以验证收到的条数 messages WS_QUEUE.drain() assert len(messages) 5 finally: driver.quit() context.close() p.stop()如果wait_for_ws_message是从 B 队列取payload字段的你会注意到on_ws_frame里我并没有把一个包装好的字段塞进去而是塞了原始文本。这里我建议根据实际需求调整我通常塞两个字段eventreceived/sent和payload因为测试里有时要判断服务器是否收到过客户端的 ping。Playwright 的framereceived回调在 Python 里传的是一个 dict常见字段是payload。二进制帧的 payload 视版本可能是 base64 字符串也可能是 bytes所以上面代码里做了类型判断。如果你发现payload是 base64 字符串要解码再断言import base64 try: raw base64.b64decode(raw).decode(utf-8, errorsreplace) except (binascii.Error, ValueError): pass这段代码我一般放进on_ws_frame的类型判断分支里。4. 实战踩坑从端口起不来到消息凭空消失4.1 调试端口相关的一串问题最常见的报错是 Selenium 连不上 9222 端口。我踩过的实际场景有四种端口被上次残留的 Chrome 进程占用。测试失败后Chrome 窗口没关干净playwright profile 目录还被锁着。下次启动直接报 “user data directory is already in use”。解决方式是每次测试前删掉 profile 目录或者在finally里保证context.close()一定执行。--remote-debugging-port没生效。在 Windows 上Chrome 新版本对远程调试支持摇摆不定哪怕参数写了访问/json/version也可能 404。我的习惯是启动后立即做一次健康检查import urllib.request def check_debug_port(port9222): try: with urllib.request.urlopen(fhttp://127.0.0.1:{port}/json/version, timeout2) as resp: return resp.status 200 except Exception: return FalseChromedriver 版本与 Chrome 不一致。这个报错信息很直白把 chromedriver 换成一致版本即可。跨源策略拦截。某些自动测试框架里CDP 客户端来自非 Chrome 原生源会被--remote-allow-origins限制。启动参数里主动加上--remote-allow-origins*能解决一大部分。4.2 监听事件注册时机晚一步消息就没了这是整个方案里最隐蔽、也最容易误判的坑。我一开始是 Selenium 先driver.get打开页面然后再启动 Playwrightconnect_over_cdp。结果发现页面里已经建立的 WebSocket 一条都收不到。原因很简单Playwright 只能监听“它已经知道”的页面上的 WebSocket。如果页面和 WebSocket 连接都在 Playwright attach 之前就建好了Playwright 不会收到任何websocket事件。解决方案就是我在代码结构里强调的——先启动 Playwright 并注册事件再让 Selenium 打开页面。模式 B 下浏览器由 Playwright 启动天然不会踩这个坑。还有一类场景是页面是 SPA前端路由切换时会新建 WebSocket 连接。Playwright 的page.on(websocket)能覆盖同一页面内的所有连接只要页面对象没变事件就不会丢。但如果是 iframe 里的 WebSocketpage.on(websocket)默认是否覆盖取决于具体版本。我在项目里遇到过 iframe 内支付组件建了独立 WS 的情况最后加了一层context.on(page)里对 iframe 的 frame 事件做额外监听才补上。如果你的被测页面也有重 iframe建议先用抓包确认消息来源再决定要不要扩监听范围。4.3 队列堆积与内存失控WebSocket 帧是高频事件行情服务一秒推送几十条很常见。如果测试跑了半小时队列里的消息没人消费queue.Queue会无限增长。我最初用无界队列跑到第 40 分钟时测试进程内存吃掉 800MB直接 OOM。现在我在队列初始化时就设maxsize满了之后丢最旧的消息。这个策略在断言类测试里几乎不受影响——你要断言的都是“新消息”或“特定 seq 的消息”而不是历史全量消息。如果是做回归回放我会把消息同时写一份到本地文件这样队列轻量历史的还能离线查。4.4 高频问题速查表现象原因解决方案Selenium 连接报WebDriverException: unable to connect调试端口没开健康检查 9222确认 Playwright 启动参数里有--remote-debugging-portPlaywright 收不到任何 WS 消息页面在监听注册前就打开了 WS调整启动顺序Playwright 先启动并注册事件Selenium 再driver.get只收到文本帧二进制帧乱码Playwright 版本对二进制帧封装不同对 payload 做base64.b64decode兜底测试运行 30 分钟后内存暴涨无界队列积压给queue.Queue设maxsize或消费后落盘Chrome 和 chromedriver 版本不匹配环境升级不同步用channelchrome指定系统 Chrome锁定两边版本Selenium 连上了但页面操作很卡Selenium 通过 CDP attach 后性能开销大只把监听器用于抓消息不要混用 Selenium 和 Playwright 同时操作同一个页面最后分享一点实操经验这套方案里Playwright 更适合当一个“旁路取证”工具而不是第二个操作控制器。我见过有人试图像 Selenium 一样用 Playwright 直接点页面结果两边同时对同一个元素操作把状态搞乱了测试出现偶发失败。我的原则是——Selenium 负责所有业务操作Playwright 只负责看 WebSocket 帧两者完全分离。消息落到队列后我还会批量写入一个 JSONL 文件这样除了断言还能在测试失败后回放消息快速判断是前端渲染问题还是服务器数据问题。一次排查线上偶现丢消息时正是回放了录制下来的 WS 日志才定位到是服务端某个推送批次缺了一条页面侧一切正常。所以别小看监听器日志的价值——它既是断言依据也是问题定位的底稿。
阅读完成 · 觉得有帮助?
咨询建站