pytest本身不支持websocket测试，需借助websockets或websocket-client等第三方库，并配合pytest-asyncio插件处理异步逻辑，注意连接管理、超时设置、异常捕获及心跳验证。
pytest本身不支持WebSocket测试，得靠第三方库
pytest只是通用测试框架，没有内置WebSocket客户端或服务端能力。想测长连接，必须引入能发起WebSocket请求、维持连接并收发消息的工具。最常用的是 websockets 库（纯异步）或 websocket-client（同步，适合简单场景）。选哪个取决于你被测服务是异步（如FastAPI/Starlette）还是同步（如Flask-SocketIO），以及你的测试逻辑是否需要并发多个连接。
常见错误现象：RuntimeError: asyncio.run() cannot be called from a running event loop —— 这是因为 pytest 默认在主线程运行，而 websockets.connect() 是协程，直接 await 会出错。必须用 asyncio.run() 包一层，或改用 pytest-asyncio 插件。
- 若项目已用
asyncio，推荐装pytest-asyncio，并在测试函数上加@pytest.mark.asyncio - 若只是简单连一下、发一条、收一条，用
websocket-client更轻量，不用碰事件循环 - 避免在
setup_method里直接await connect()—— pytest 的 setup 方法不是异步的，会静默失败
用 pytest-asyncio 测试 FastAPI WebSocket 路由
以 FastAPI 的 /ws 端点为例，它接受连接、回显消息。测试关键在于：连接不能立刻断开，要保持活跃直到验证完成；消息收发需按序等待；超时必须设，否则测试会卡死。
实操要点：
- 测试函数必须标记为
@pytest.mark.asyncio - 用
async with websockets.connect(...)确保连接自动关闭 - 发送后立即
await websocket.recv()，不要用time.sleep()等待 - 务必设置
timeout=5参数，例如await asyncio.wait_for(websocket.recv(), timeout=5)，否则服务端没响应时测试挂起
示例片段：
图片提示词生成器？不止如此。 马甲系统 —— 把脑海中的画面，翻译成AI能理解的专业表达。 用得越多，它越懂你：首次需要多问几句确认方向，用久了几乎一说就懂。 用得越多，它越快：缓存机制让后续对话越来越省。 RAG进化：成功案例持续入库，越跑越聪明。 输入「新手指南」查看完整功能介绍
@pytest.mark.asyncio
async def test_ws_echo():
async with websockets.connect("ws://localhost:8000/ws") as ws:
await ws.send("hello")
resp = await asyncio.wait_for(ws.recv(), timeout=5)
assert resp == "echo: hello"
测试连接异常与重连逻辑
真实场景中，WebSocket 长连接会遇到网络抖动、服务重启、心跳超时等断连情况。单纯测“通”没意义，要覆盖 ConnectionClosed、InvalidStatusCode 等异常分支。
注意点：
-
websockets.exceptions.ConnectionClosed是正常断连抛出的异常，不是 bug，需显式捕获 - 若服务端返回 401/403，客户端会抛
websockets.exceptions.InvalidStatusCode，而不是 HTTP 异常 - 不要在测试里写 while 循环模拟重连 —— 它会让测试变慢且难调试；应单独写一个“断连后能否重建连接”的用例，用
await asyncio.sleep(0.1)模拟间隔即可 - 用
pytest.raises()断言异常类型，例如with pytest.raises(websockets.exceptions.ConnectionClosed): ...
如何测心跳与超时行为
长连接稳定性核心是心跳机制。服务端通常要求客户端每 30 秒发一次 ping，超时未收到则断连。测试这个，不能只看连接是否建立，要看「空闲 N 秒后是否被踢」。
实操建议：
- 启动连接后，先
await websocket.ping()手动触发一次，确认基础通路 - 然后
await asyncio.sleep(35)，再尝试await websocket.send("test")—— 此时应抛ConnectionClosed - 若服务端用
ping_interval自动发 ping，客户端需调用websocket.pong()响应，否则会被视为失联；测试时可 patch 或 mock 该行为来验证服务端判断逻辑 - 避免依赖真实时间等待：用
asyncio.create_task()启一个后台任务定时发 ping，主流程专注验证响应
容易被忽略的是：WebSocket 协议层的 ping/pong 是控制帧，不会触发你写的 on_message 回调。测试心跳必须检查连接状态或抓包，不能只靠业务消息收发来判断。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
