【捕获WebSocket】基于CDP与Playwright增强Selenium测试中的实时消息验证

1. 为什么我们需要在Selenium里监听WebSocket如果你做过Web自动化测试尤其是那种带实时功能的比如在线文档编辑、股票行情看板或者在线聊天室你肯定遇到过这个头疼的问题UI操作做完了页面也变了但你怎么知道后台的实时通信真的发生了用Selenium我们擅长的是“看”。我们能找到按钮点击它能找到输入框填上文字能断言页面上某个元素出现了或者文字变了。这些都是基于浏览器最终渲染出来的“结果”进行验证。但是很多现代应用的核心逻辑尤其是实时协作、即时通知这类功能它们的“动作”发生在用户看不见的地方——WebSocket连接里。想象一下这个场景你在测试一个在线文档工具。用户A输入了一段文字这个操作除了要在页面上显示出来更应该通过WebSocket立刻、准确地发送到服务器再由服务器广播给正在协作的用户B和C。如果你只用Selenium你只能断言用户A的页面上文字出现了用户B和C的页面也更新了。但这中间的黑盒子里发生了什么消息真的发出去了吗发了几条消息的格式和内容对吗有没有发多余的消息这些传统的基于UI的断言是无能为力的。这就是我们面临的测试盲区。UI测试告诉你“结果”对了但无法验证“过程”是否正确。而很多隐蔽的Bug恰恰藏在这个“过程”里。比如可能因为前端代码的缺陷一次点击误发了两次相同的WebSocket消息导致服务器处理异常或者消息的action字段拼写错误导致后端无法识别。这些光看页面是看不出来的。所以我们需要一双能“听”的耳朵嵌入到我们的Selenium测试流程里去监听浏览器和服务器之间那些“窃窃私语”般的WebSocket消息。这不仅能将测试覆盖到更深层的业务逻辑更能帮助我们精准定位问题是出在前端、网络还是后端协议上。而实现这双“耳朵”的关键就是Chrome DevTools Protocol以及一个能优雅使用它的工具——Playwright。2. 理解我们的工具箱CDP与Playwright的角色要解决监听问题我们得先搞清楚手头有什么工具以及它们各自擅长什么。别被那些术语吓到我们打个简单的比方。Selenium就像是一个经验丰富的机器人操作员。它有一套标准的指令集WebDriver API可以命令浏览器这个“机器”“点击这里”、“在那输入文字”、“看看那个标题是不是‘成功’”。它很强大生态成熟但它的工作层面主要在“操作”和“观察最终状态”。Chrome DevTools Protocol简称CDP则是浏览器这里特指Chrome或Chromium内核的浏览器暴露出来的底层调试接口。你可以把它理解为浏览器的“后台管理系统”或“工程模式”。通过这个协议你能干很多高级活儿监控网络请求、分析内存使用、执行JavaScript、当然也包括监听所有WebSocket帧的发送与接收。CDP能力极强但直接使用它的原始JSON-RPC消息比较繁琐。这时候Playwright登场了。你可以把它看作一个更现代、更全能的“机器人操作员”。它原生支持CDP并且用非常友好的API把它包装了起来。Playwright自己当然也能做自动化操作但在这里我们看中的是它作为“CDP客户端”的便捷性。我们不需要自己去建立WebSocket连接、发送复杂的CDP命令、再解析返回的流数据。Playwright提供了像page.on(websocket, ...)这样直观的事件监听器让我们几乎可以用写前端事件监听的方式来捕获底层的网络流量。那么我们的方案就清晰了不替换现有的、稳定的Selenium测试框架而是引入Playwright作为一个专门的“网络监听器”。让Selenium继续干它擅长的UI驱动和操作让Playwright通过CDP连接同一个浏览器实例专心负责“窃听”WebSocket通信。两者各司其职强强联合。这个架构的核心挑战不在于API调用而在于如何让这两个独立工具有序地协同工作这就是接下来要解决的生命周期问题。3. 核心挑战让Selenium和Playwright和谐共处把Playwright引入Selenium项目听起来就是加几行代码连接一下但实际一跑你可能马上就会撞上第一个大坑ECONNREFUSED—— 连接被拒绝。这个问题看似是网络错误实则根源是生命周期不同步。我们来拆解一下浏览器启动和连接的过程。当你用Selenium的new ChromeDriver()启动浏览器时浏览器进程才开始运行。而要让Playwright能通过CDP连接必须在启动浏览器时就告诉它“请打开一个调试端口等着别人来连接。” 这需要通过Selenium的ChromeOptions添加--remote-debugging-port9222这样的参数。但即使端口打开了问题也没完。因为浏览器启动需要时间从进程启动到CDP服务在端口上准备好监听有一个细微的延迟。如果你的测试代码顺序是启动Selenium ChromeDriver。立刻调用Playwright的chromium.ConnectOverCDPAsync(“http://localhost:9222”)。那么极有可能代码执行到第2步时浏览器进程还在初始化9222端口虽然被占用了但CDP服务还没准备好接受连接Playwright一尝试连接自然就被拒绝了。所以“什么时候连”就成了关键。我踩过这个坑后的经验是必须确保连接动作发生在浏览器完全启动并稳定运行之后。在我的实践中一个相对稳妥的调用顺序是这样的测试类初始化时只做准备工作。生成一个空闲的调试端口号比如9222创建好我们的WebSocket监听器类实例但先不进行连接。单个测试用例初始化时首先调用基类的TestInitialize方法或者你的测试框架的setup方法让Selenium启动带有调试端口参数的浏览器。然后显式地、等待性地调用监听器实例的EnsureConnected()方法。这个方法内部会尝试连接CDP并且最好包含简单的重试逻辑或延迟确保浏览器已经就绪。// 伪代码示例 public class MyTestClass { private IWebDriver _driver; private WebSocketListener _wsListener; [ClassInitialize] public static void ClassSetup() { // 确定调试端口创建监听器对象未连接 _wsListener new WebSocketListener(port: 9222); } [TestInitialize] public void TestSetup() { // 1. Selenium 启动浏览器携带--remote-debugging-port var options new ChromeOptions(); options.AddArgument(--remote-debugging-port9222); _driver new ChromeDriver(options); // 2. 显式连接Playwright到CDP _wsListener.EnsureConnected(); } }这种“先启动再连接”的顺序虽然看起来多了一步但能极大地提高连接的稳定性。它明确了责任的边界Selenium负责创造环境启动带调试功能的浏览器Playwright负责在环境准备好后接入。这避免了因线程调度或启动速度差异导致的竞态条件。4. 关键设计如何精准捕获“操作相关”的消息连接稳了能收到消息了但你会发现下一个问题消息太多了而且很“吵”。页面一加载可能就有一堆初始化消息、心跳包、其他无关的通知。如果你在测试中直接断言“从测试开始到现在收到的所有EditRecord消息等于1条”那测试会脆弱不堪因为任何无关的消息都会干扰结果。我们测试的核心诉求不是“页面收没收到过某种消息”而是“某个特定的UI操作是否触发了预期的消息”。比如“点击保存按钮应该且只应该发送一条SaveDocument的WebSocket消息。” 我们需要一种方法从连续的消息流中精确地切出与本次操作相关的那一小段。我尝试过用时间戳但发现不可靠。消息的产生、传递、被CDP捕获、再到我们代码处理中间有微小的延迟且可能不稳定。用固定时间窗口去截取要么可能漏消息要么可能包含进无关消息。后来我采用了一个非常简单但异常有效的思路基于消息序号的切片。原理如下每捕获到一条WebSocket消息无论是发送还是接收都给它打上一个全局递增的序号。在执行我们要测试的UI操作之前记录下当前的序号比如是100。执行UI操作。操作完成后等待一小段合理的时间例如200-500毫秒让网络消息有足够时间传递和被捕获。最后从所有捕获的消息中过滤出序号大于100的所有消息。这些消息理论上就是在我们记录序号之后产生的也就是我们的UI操作可能触发的消息。// 核心捕获方法的简化逻辑 public ListWsMessage CaptureMessagesRelatedToAction(Action uiAction, int waitMsAfterAction 300) { // 1. 记录操作开始前的消息序号 long startSequenceNumber _currentMaxSequenceNumber; // 2. 执行UI操作例如点击按钮、输入文本 uiAction.Invoke(); // 3. 等待一段时间让网络消息得以传递和捕获 Thread.Sleep(waitMsAfterAction); // 4. 过滤出序号大于开始序号的消息 var relatedMessages _allCapturedMessages .Where(msg msg.SequenceNumber startSequenceNumber) .OrderBy(msg msg.SequenceNumber) .ToList(); return relatedMessages; }这个方法妙在哪里它不依赖于绝对时间只依赖于消息事件的相对顺序。只要Playwright通过CDP捕获消息是基本有序的这一点通常能保证这个切片就是准确的。它完美地将测试关注点隔离到了“本次操作”的上下文中排除了页面加载期或其他异步任务产生的噪音让断言变得清晰和稳定。当然这不是银弹如果被测应用在后台有非常频繁的、与操作无关的定时消息可能还需要结合其他过滤条件但在绝大多数交互场景下这个方法已经足够好用了。5. 处理消息细节方向、空消息与业务字段提取抓到了我们关心的那批消息后接下来就是分析和验证它们。这里有几个细节处理能让我们的工具更健壮、更好用。首先是消息方向。通过CDP我们能区分消息是浏览器发送给服务器的还是服务器发送给浏览器的。在Playwright的事件里对应着FrameSent和FrameReceived。这个信息很有用。比如你可能只想断言某个操作发出了特定的请求而不关心接收到的响应或者反之。在我的监听器里我会把方向抽象成一个属性方便后续过滤。public class WebSocketMessage { public long SequenceNumber { get; set; } public string Direction { get; set; } // Sent 或 Received public string RawText { get; set; } }其次是处理“空消息”或“无意义消息”。在实际抓包中你经常会看到一些{}、空字符串或者只包含心跳标识的消息。这些消息对于验证业务逻辑通常没有价值反而会成为断言时的干扰项。因此我在工具方法里提供了一个可选开关允许在提取消息文本时过滤掉它们。public IEnumerablestring GetReceivedMessageTexts(ListWebSocketMessage messages, bool includeEmpty false) { return messages .Where(m m.Direction Received) // 只收关心的方向 .Where(m includeEmpty || !IsEmptyOrNoise(m.RawText)) // 可选过滤空消息 .Select(m m.RawText); } private bool IsEmptyOrNoise(string text) { return string.IsNullOrWhiteSpace(text) || text {} || text.Contains(heartbeat); }最后也是最重要的是从消息中提取业务逻辑字段。我们很少需要对整个JSON消息体做字符串匹配断言那样太脆弱。通常业务语义体现在某个特定的字段上比如action、type或event。我们需要解析JSON并提取出这些关键字段进行统计和断言。例如一个典型的协作消息可能是{action: CursorMove, userId: 123, position: 50}。我们关心的是“有没有发生CursorMove这个动作”。我会写一个辅助方法专门用来统计一批消息中各个action出现的次数。public Dictionarystring, int CountActions(IEnumerablestring messageTexts) { var actionCounts new Dictionarystring, int(); foreach (var text in messageTexts) { if (IsEmptyOrNoise(text)) continue; try { // 使用如Newtonsoft.Json或System.Text.Json解析 var jsonDoc JsonDocument.Parse(text); if (jsonDoc.RootElement.TryGetProperty(action, out var actionElement)) { var action actionElement.GetString(); if (!string.IsNullOrEmpty(action)) { // 统计次数 actionCounts[action] actionCounts.GetValueOrDefault(action) 1; } } } catch (JsonException) { // 非JSON或格式错误忽略这条消息 continue; } } return actionCounts; }这样在测试断言中我们就可以写出非常清晰、意图明确的验证代码// 1. 执行一个输入文字的操作并捕获操作后产生的消息 var messages _wsListener.CaptureMessagesRelatedToAction(() { textEditor.SendKeys(Hello World); }); // 2. 提取并统计这些消息中的 Action var receivedTexts GetReceivedMessageTexts(messages, includeEmpty: false); var actionStats CountActions(receivedTexts); // 3. 断言期望触发一次 TextEdit 动作 Assert.AreEqual(1, actionStats.GetValueOrDefault(TextEdit)); // 还可以断言不应该出现的动作 Assert.IsFalse(actionStats.ContainsKey(Undo)); // 例如输入操作不应触发撤销动作这种断言方式的力量在于它不仅能验证“期望发生的发生了”还能意外地发现“不该发生的也发生了”这对于定位一些隐蔽的Bug非常有效。6. 实战从搭建到运行一个完整的例子说了这么多理论我们来串一个完整的、可运行的例子。假设我们要测试一个简易的在线笔记应用验证“输入文本会触发WebSocket同步消息”。第一步环境准备与项目搭建。创建一个新的测试项目比如用NUnit或xUnit。通过NuGet安装必要的包Selenium.WebDriver用于UI自动化。Microsoft.Playwright用于连接CDP和监听WebSocket。当然还有对应的浏览器驱动。第二步创建核心的WebSocket监听器类。这个类将封装所有与Playwright CDP连接和消息处理的逻辑。using Microsoft.Playwright; using System.Collections.Concurrent; public class WebSocketCaptureTool { private IPlaywright _playwright; private IBrowser _browser; private IPage _page; private CDPSession _cdpSession; private readonly string _cdpUrl; private long _messageSequence 0; private readonly ConcurrentBagWebSocketMessage _capturedMessages new(); public WebSocketCaptureTool(string cdpUrl) { _cdpUrl cdpUrl; // 例如http://127.0.0.1:9222 } public async Task EnsureConnectedAsync() { if (_cdpSession ! null) return; _playwright await Playwright.CreateAsync(); // 通过CDP连接到已存在的浏览器 _browser await _playwright.Chromium.ConnectOverCDPAsync(_cdpUrl); // 获取第一个页面通常就是Selenium控制的那个 _page _browser.Contexts[0].Pages[0]; // 创建CDP会话用于监听网络事件 _cdpSession await _page.Context.NewCDPSessionAsync(_page); // 启用网络域监听WebSocket事件 await _cdpSession.SendAsync(Network.enable); await _cdpSession.SendAsync(Network.setCacheDisabled, new { cacheDisabled true }); // 订阅WebSocket帧事件 _cdpSession.On(Network.webSocketFrameSent, HandleWebSocketEvent(Sent)); _cdpSession.On(Network.webSocketFrameReceived, HandleWebSocketEvent(Received)); } private ActionIDictionarystring, object HandleWebSocketEvent(string direction) { return (payload) { var frameData payload[response] as IDictionarystring, object; if (frameData ! null frameData.ContainsKey(payloadData)) { var message new WebSocketMessage { SequenceNumber Interlocked.Increment(ref _messageSequence), Direction direction, RawText frameData[payloadData].ToString() }; _capturedMessages.Add(message); } }; } // 核心的切片捕获方法 public async TaskListWebSocketMessage CaptureForActionAsync(FuncTask uiAction, int waitMs 500) { var startSeq Interlocked.Read(ref _messageSequence); await uiAction.Invoke(); await Task.Delay(waitMs); // 使用异步等待避免阻塞线程 return _capturedMessages .Where(m m.SequenceNumber startSeq) .OrderBy(m m.SequenceNumber) .ToList(); } // ... 其他辅助方法如 GetReceivedMessageTexts, CountActions 等 } public class WebSocketMessage { public long SequenceNumber { get; set; } public string Direction { get; set; } public string RawText { get; set; } }第三步在测试类中集成Selenium和监听器。[TestFixture] public class RealTimeNoteTest { private IWebDriver _driver; private WebSocketCaptureTool _wsTool; private const int DebugPort 9222; [SetUp] public async Task Setup() { // 1. 配置Selenium启动带调试端口的Chrome var options new ChromeOptions(); options.AddArgument($--remote-debugging-port{DebugPort}); // 可能还需要其他配置如禁用沙箱等 options.AddArgument(--no-sandbox); options.AddArgument(--disable-dev-shm-usage); _driver new ChromeDriver(options); // 确保ChromeDriver路径已设置 // 2. 创建并连接监听工具 _wsTool new WebSocketCaptureTool($http://127.0.0.1:{DebugPort}); await _wsTool.EnsureConnectedAsync(); // 3. 导航到被测页面 _driver.Navigate().GoToUrl(http://localhost:3000/notes); } [Test] public async Task TextInput_ShouldTriggerSingleEditMessage() { // 准备找到输入框 var textArea _driver.FindElement(By.CssSelector(.note-editor)); // 使用监听工具捕获操作期间的消息 var relatedMessages await _wsTool.CaptureForActionAsync(async () { // 执行UI操作 textArea.SendKeys(这是测试文本); // 如果UI操作涉及等待可以在这里进行 await Task.Delay(100); }, waitMs: 800); // 等待消息稳定 // 分析与断言 var receivedTexts _wsTool.GetReceivedMessageTexts(relatedMessages, includeEmpty: false); var actionCounts _wsTool.CountActions(receivedTexts); // 关键断言应该恰好触发一次 textEdit 动作 Assert.That(actionCounts.GetValueOrDefault(textEdit), Is.EqualTo(1)); // 可选断言没有触发其他非预期的动作如 cursorMove如果输入不应该触发光标移动消息 Assert.That(actionCounts.ContainsKey(cursorMove), Is.False); } [TearDown] public void Teardown() { _driver?.Quit(); _wsTool?.Dispose(); // 需要实现IDisposable来清理Playwright资源 } }第四步运行与调试。运行这个测试。如果一切顺利Selenium会打开浏览器并操作Playwright在后台监听。当你在输入框打字时监听器会捕获到对应的WebSocket消息测试将通过。如果没收到消息或者收到了多条textEdit消息测试就会失败帮助你立刻发现问题。在调试时如果遇到连接问题请务必检查浏览器启动参数是否正确添加了--remote-debugging-port。Playwright连接时使用的IP和端口是否正确建议用127.0.0.1而非localhost避免IPv6问题。确保EnsureConnectedAsync在浏览器启动之后被调用并且有足够的重试或等待逻辑。7. 避坑指南常见问题与方案边界在实际集成过程中除了生命周期问题你还可能遇到其他一些“坑”。这里分享几个我遇到的典型问题和处理思路。1. CDP连接不稳定或失败。除了前面提到的启动顺序问题网络环路也可能导致连接失败。在本地开发时使用localhost有时会解析到IPv6地址::1而浏览器可能只监听在IPv4的127.0.0.1上。最稳妥的办法是在连接CDP时显式指定使用127.0.0.1。此外在ConnectOverCDPAsync后添加一个短暂的延迟或简单的重试循环也能提升初次连接的稳定性。2. 消息捕获不全或顺序错乱。虽然CDP流通常有序但在高频率消息场景下偶尔可能出现顺序问题。如果你的断言对消息顺序有严格要求可以考虑在消息对象中加入高精度时间戳DateTime.UtcNow.Ticks作为序号之外的辅助排序依据。但绝大多数业务验证场景基于序号的切片已经足够。3. 性能开销。启用CDP监听网络尤其是捕获所有WebSocket帧的完整负载会对浏览器性能有轻微影响。在长时间运行或性能基准测试中需要留意。通常的UI自动化测试中这个开销是可以接受的。如果确实成为瓶颈可以考虑在CDP命令中只启用必要的域或者在非验证阶段暂停监听。4. 方案边界与取舍。必须清醒认识到这个方案是一个增强补丁而不是一个替代方案。它的目标是在最小化改动现有Selenium测试的前提下增加对实时通信的验证能力。因此它有明确的边界浏览器限制它严重依赖Chrome DevTools Protocol因此主要适用于Chromium内核的浏览器Chrome, Edge, Opera。如果你需要测试Firefox或Safari这个方案不直接适用。复杂度它引入了额外的组件Playwright和更复杂的生命周期管理增加了测试框架的复杂度。不是性能测试工具虽然能捕获消息但它不适合做精确的性能计时如消息延迟因为CDP事件本身有处理延迟。如果你的项目满足以下条件这个方案会非常合适主要测试Chromium浏览器、已有成熟的Selenium测试体系、迫切需要增强对WebSocket行为的断言。如果你的项目正在考虑全新的测试框架且对跨浏览器有强需求那么直接评估和迁移到Playwright可能是更一劳永逸的选择。但对于那些“船大难掉头”的遗留Selenium项目这个基于CDP和Playwright的“外挂式”监听方案无疑是一个成本低、收益高的精准增强手段。