测试 WebSocket 接口：一发一收只适用于回显服务

行情推送接口的自动化测试一直很稳定：连接、发订阅请求、收一条消息、断言是订阅成功。后来服务端在连接建立后先推一条「系统公告」，测试开始随机失败；再后来有人把两个订阅并发发出去，断言又对不上号。最难查的一次是测试卡住不动，直到 CI 的总超时把整个任务杀掉。

WebSocket 和普通的 HTTP 接口很不一样：HTTP 是一个请求对应一个响应，WebSocket 是一条长期存在的双向通道，双方都可以随时发消息。测试代码如果沿用「发一条、收一条、断言」的写法，只有在服务端是回显服务时才可靠。

本文用 Python 3.14 和 websockets 17.2，在本地起一个可控的 WebSocket 服务，把消息匹配、超时、握手鉴权、关闭码和连接清理逐个跑一遍。实验见文末。

图 1 · 示意，各行结果为实测（Python 3.14.8、websockets 17.2）：服务端可以随时推送，处理快的请求先返回，所以「下一条消息就是响应」只在回显服务上成立。按 id 分派时每个请求拿到自己的结果，推送单独入队。recv 本身不会超时，每次等待都要套上超时；连接结束时断言具体的关闭码。

一、连接的生命周期

测试 HTTP 接口时，连接对测试代码几乎是透明的。WebSocket 的测试要显式处理四个阶段：

1. 握手：客户端发一个带 Upgrade: websocket 的 HTTP 请求，服务端回 101 后连接升级。鉴权通常发生在这里，失败时服务端直接回普通的 HTTP 状态码。
2. 收发消息：双方各自发送文本或二进制消息，没有「这条是那条的响应」这种协议层面的关系。
3. 保活：库会定期发 ping，对方要回 pong；长时间不回就认为连接已死。
4. 关闭：一方发关闭帧（带关闭码和原因），另一方回关闭帧，然后断开 TCP。没有走这一步就断开，属于异常关闭。

每个阶段都有自己的失败方式，测试要分别覆盖。

二、消息匹配：响应不一定是下一条

在回显服务上，「发一条、收一条」没有问题：

echo 服务一发一收：收到 {"id": 1, "result": "hi"}

只要服务端会主动推送，这个写法就会拿错消息：

服务端先推送了一条通知：一发一收拿到的是 {'type': 'notice', 'text': '系统维护'}，断言 result=='hi' 失败

请求的处理时间不同时，响应还可能乱序：

先发 id=1（慢）再发 id=2（快）：第一条收到的是 id=2；按发送顺序配对会把 2 的结果当成 1 的

这两种情况在生产环境里都是正常行为，并不是服务端的缺陷。测试客户端要做的是按关联 ID 匹配：每个请求带一个 id，由一个后台任务持续读取所有消息，有 id 的交给对应的等待者，没有 id 的推送放进单独的队列：

class Client:
    def __init__(self, ws):
        self.ws = ws
        self.pending = {}               # id -> Future
        self.pushes = asyncio.Queue()   # 服务端主动推送
        self.next_id = 0
        self.reader = asyncio.create_task(self._read())

    async def _read(self):
        async for raw in self.ws:
            msg = json.loads(raw)
            if msg.get("id") in self.pending:
                self.pending.pop(msg["id"]).set_result(msg)
            else:
                await self.pushes.put(msg)

    async def call(self, timeout=2, **payload):
        self.next_id += 1
        mid = self.next_id
        fut = asyncio.get_running_loop().create_future()
        self.pending[mid] = fut
        await self.ws.send(json.dumps({"id": mid, **payload}))
        async with asyncio.timeout(timeout):
            return await fut

按 id 分派：请求 1 得到 done-1，请求 2 得到 done-2；推送单独收到 notice:系统维护

推送消息进了自己的队列，测试可以单独断言「连接建立后收到了一条公告」，而不是让它干扰请求的断言。如果协议本身没有关联 ID，就要按消息类型或业务键（订阅的主题、订单号）匹配，并在测试里写明这个假设。

为什么一定要有一个统一的读取任务？因为 websockets 不允许两个协程同时读同一个连接：

两个协程同时 recv：ConcurrencyError：cannot call recv while another coroutine is already running recv or recv_streaming

两个测试步骤各自调用 recv() 的写法，在并发场景下直接报错；即便库允许，哪个协程拿到哪条消息也是不确定的。

三、超时：recv 会一直等

客户端默认值：open_timeout=10，ping_interval=20，ping_timeout=20，close_timeout=10，max_size=1048576；recv() 本身没有超时参数
服务端不回复：recv 本身没有超时，套上 asyncio.timeout(1) 后约 1 秒抛出 TimeoutError；连接仍是 OPEN

websockets 给握手（open_timeout）、保活（ping_interval、ping_timeout）和关闭（close_timeout）都设了默认超时，但等一条消息没有。服务端只要还在回 pong，连接就是健康的，recv() 就会一直等下去。开头那个卡住的测试，就是在等一条永远不会来的消息。

每一次等待消息都要套上超时，超时时间按这条消息的业务预期来定。超时之后连接仍然是 OPEN，这个测试应当失败并关闭连接，而不是继续在这条连接上做后面的断言。

四、握手鉴权：失败发生在连接之前

握手不带 Token：InvalidStatus，HTTP 401；连接根本没有建立
握手时在请求头带 Token：连接成功，收到 ok

在握手阶段鉴权时，失败表现为一个普通的 HTTP 响应，客户端拿到的是握手异常，而不是某条消息里的错误码。测试要覆盖两种情况：不带凭证时握手被拒绝，带上凭证时连接成功。

另一种常见做法是先建立连接、再发一条认证消息。这时未认证的连接也能建立，测试要断言的是「认证之前发的业务请求被拒绝」或「超过时限没有认证就被服务端关闭」。两种设计的测试方式不同，先确认接口用的是哪一种。浏览器里的 WebSocket API 不能自定义请求头，所以面向浏览器的接口常用查询参数、Cookie 或首条消息传递凭证；这些方式各有泄露和校验上的问题，不在本文范围。

五、关闭码：区分正常结束、被踢和出错

正常关闭后 recv：ConnectionClosedOK，关闭码 1000
服务端以 1008 踢掉连接：ConnectionClosedError，关闭码 1008，原因 'token expired'
发送 2 MiB 消息（服务端 max_size 1 MiB）：服务端以关闭码 1009 断开，原因 'frame exceeds limit of 1048576 bytes'
服务端直接断开 TCP：ConnectionClosedError，没有收到关闭帧（rcvd=None）；RFC 6455 把这种情况记作 1006，它不会出现在关闭帧里

关闭码是 WebSocket 接口契约的一部分。常用的几个：

关闭码	含义	测试关注
1000	正常关闭	业务结束后双方正常断开
1001	一方离开（服务重启、页面关闭）	客户端是否重连
1008	违反策略（凭证过期、权限不足）	原因文本、客户端是否停止重连
1009	消息过大	服务端的上限与客户端的处理
1011	服务端内部错误	告警与重连
1006	没有收到关闭帧（异常断开）	只在本地出现，不能由对方发送

websockets 用 ConnectionClosedOK 和 ConnectionClosedError 区分正常与异常关闭。测试被踢下线的场景时，要断言具体的关闭码和原因，而不是只断言「连接断了」：1008 和 1006 在客户端上应当触发完全不同的处理（前者不应自动重连，后者通常要重连）。

消息大小的上限也是契约：websockets 两端默认都是 1 MiB，超过时连接被直接关闭，而不是只拒绝这一条消息。

六、清理：忘记关闭的连接留在服务端

测试里 connect 之后没有关闭：服务端仍有 1 个打开的连接
显式 close（或用 async with）之后：服务端打开的连接 0 个

HTTP 测试忘记关闭连接，通常只是浪费一点资源；WebSocket 测试忘记关闭，服务端会一直认为这个客户端在线。如果服务端按用户限制连接数、按在线状态推送，后面的测试就会受到影响。用 async with 或测试框架的清理钩子关闭连接，不要依赖对象被回收时的析构。

七、测试分几层

层次	验证什么	做法
消息处理逻辑	收到某条消息后的业务结果	不建立连接，直接调用处理函数
协议契约	握手、鉴权、消息格式、关闭码、大小上限	本地服务 + 真实客户端库，本文的做法
连接行为	断线重连、保活、多连接下的推送	能控制网络的环境（断开、延迟）

用同一个团队写的客户端去测服务端，只能证明两边的理解一致，不能证明符合协议或别的客户端能用；协议契约层最好也用一个独立实现（另一种语言的库、浏览器）跑一遍关键场景。

八、检查清单

• 每个请求带关联 ID，由一个读取任务统一分派；推送消息单独断言。
• 每一次等待消息都有超时。
• 握手鉴权的成功与失败分别测试。
• 断言具体的关闭码与原因，区分 1000、1008、1006 等。
• 测试消息大小上限。
• 每个连接都在测试结束时关闭。

---

配套实验

• codesphere-labs/testing/websocket-testing：Python 3.14.8、websockets 17.2：推送与乱序响应下一发一收出错、按 id 分派、并发 recv、recv 无超时、握手鉴权、关闭码 1000/1008/1009/1006 与连接清理（验证记录）

参考资料

• RFC 6455：The WebSocket Protocol（7.4 Status Codes）
• websockets 文档：asyncio API（client、server、exceptions）
• websockets 文档：Keepalive and latency
• MDN：WebSocket