直接用 pytest-asyncio 测试 WebSocket 客户端会卡住,因 websockets.connect() 返回协程且未被正确 await 或超时控制;需用 async def 测试函数、websockets.serve() 启轻量服务端、await recv/send 并加超时,异常如 ConnectionClosedOK 属正常协议行为。

为什么直接用 pytest-asyncio 测试 WebSocket 客户端会卡住
因为 websockets.connect() 返回的是一个协程,而 pytest-asyncio 默认不等待它完成——你写了 await websocket,但没等它真正建立连接就结束了测试函数。更麻烦的是,如果服务端没启动,connect() 会挂起或抛出 OSError: [Errno 61] Connection refused,而不是立刻失败。
- 必须用
async def test_*函数,且 pytest 配置中启用asyncio_mode = auto - 不能只写
websocket = await websockets.connect(...)就完事,要加超时和异常捕获 - 真实服务端依赖(如
uvicorn启动的 ws server)会让测试变慢、不可靠;应优先用websockets.serve()在测试内启一个临时服务端
如何用 websockets.serve() 在测试内跑一个轻量服务端
不用额外进程、不占端口、不污染环境——关键是把服务端逻辑写成 async 函数,并在测试里用 await websockets.serve() 启动它。注意:这个调用本身不会阻塞,得配合 asyncio.create_task() 或 asyncio.wait_for() 控制生命周期。
- 服务端 handler 必须是
async def handler(websocket, path)形式,否则websockets.serve()会报TypeError: expected a coroutine - 测试中启动服务端后,要给客户端留出连接时间(哪怕只是
await asyncio.sleep(0.01)),否则可能连不上 - 服务端异常(如 handler 内未捕获的
ConnectionClosed)默认会被吞掉,建议在 handler 里加try/except Exception并 print,方便调试
async def echo_handler(websocket, path):
try:
async for message in websocket:
await websocket.send(f"echo: {message}")
except Exception as e:
print("Handler error:", e)
pytest-asyncio 下如何正确断言 WebSocket 消息收发
WebSocket 是双向流,recv() 和 send() 都是协程,必须 await。常见错误是忘记 await、或在 send 后没 recv 就结束测试,导致连接被强制关闭,服务端抛 ConnectionClosedError。
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
- 用
asyncio.wait_for(ws.recv(), timeout=2)替代裸await ws.recv(),避免死等 - 发送后别立刻断开,至少等一次 recv 完成,否则服务端可能来不及处理
- 若需验证多条消息,注意
recv()是 FIFO,顺序错会导致断言失败;可加assert "expected" in await ws.recv() - 不要在测试末尾手动调用
await websocket.close()—— pytest-asyncio 的 event loop 会在函数退出时自动清理,强行 close 可能触发RuntimeError: Event loop is closed
遇到 ConnectionClosedOK / ConnectionClosedError 怎么办
这不是测试失败,而是协议正常行为:客户端主动 close → 服务端收到 ConnectionClosedOK;网络中断或对方先关 → ConnectionClosedError。问题在于你的 handler 或测试逻辑是否提前终止了连接。
- 检查服务端 handler 是否在 recv 空时直接 return(这是正常结束),还是因未处理
ConnectionClosed异常而崩溃 - 客户端测试中,如果用
async with websockets.connect(...) as ws:,退出上下文会自动 close,此时服务端必然收到ConnectionClosedOK—— 这是预期行为,不用 assert 报错 - 若想模拟异常断连,可用
transport.close()(需从websocket.transport获取),但多数场景没必要;真要测容错,应在 handler 里故意 raise 异常并观察日志
测试里最易被忽略的,是服务端 handler 的异常传播机制和客户端 recv 超时控制——它们不报错,但会让消息收发看起来“随机失败”。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










