生产环境优先选websockets库，因其轻量、无依赖、RFC6455兼容性好、文档清晰、错误提示直接，是asyncio生态中事实标准的WebSocket服务端实现。
用 websockets 库写一个可断言的 WebSocket 客户端测试
Python 标准库不支持 WebSocket，必须依赖第三方库；websockets 是目前最稳定、异步友好、且自带测试友好接口的选择。它提供 connect() 的同步上下文管理（配合 asyncio.run()）和真正的异步客户端，适合在 pytest 中直接驱动连接、收发、断言。
常见错误是试图用 requests 或 urllib 发起 WebSocket 请求——它们根本无法完成握手，会直接报 400 Bad Request 或卡在 Upgrade 头失败。
- 安装：
pip install websockets - 测试时避免硬编码地址，用
pytest --ws-url="ws://localhost:8000/ws"传参或读取环境变量WS_URL - 务必用
async with websockets.connect(url) as ws:，手动调close()容易漏掉异常导致连接残留 - 接收消息建议设超时：
await asyncio.wait_for(ws.recv(), timeout=3)，否则测试可能无限挂起
模拟服务端做双向交互验证（不用启动真实后端）
很多团队卡在“没后端怎么测”，其实 websockets 自带 serve()，可在测试中快速启一个轻量 mock 服务，控制响应逻辑。关键不是复刻业务逻辑，而是验证客户端能否正确处理 ping/pong、重连、消息格式、错误帧等边界情况。
注意：mock 服务必须运行在独立 event loop 或用 asyncio.create_task() 启动，不能和测试协程共用同一个 run() 调用，否则会阻塞。
立即学习“Python免费学习笔记（深入）”；
- 用
async def echo_handler(ws, path):写一个回显 handler，收到什么就发什么，用于验证基础通路 - 主动发错格式消息（如非 JSON 字符串）触发客户端解析异常，检查是否捕获
json.JSONDecodeError - 在 handler 中调用
await ws.close(4001, "custom reason")模拟服务端主动断连，验证客户端 on_close 回调是否触发 - 别忘了在测试结束时调用
server.close()和await server.wait_closed()，否则 pytest 可能报ResourceWarning: unclosed transport
捕获并断言 WebSocket 连接生命周期事件
真实场景里，连接建立、重连、关闭原因、ping 超时这些都不是“有没有消息”能覆盖的。Python 客户端通常封装了自动重连或心跳逻辑，测试必须能观测底层事件。
websockets 不提供类似浏览器 onopen/onerror 的回调注册，所有状态需通过异常类型和连接对象属性判断。
- 连接失败：捕获
websockets.exceptions.ConnectionClosedError或OSError（如 DNS 失败），而不是只看ConnectionRefusedError - 服务端静默断连：
await ws.recv()抛ConnectionClosedOK，此时ws.close_code和ws.close_reason可断言 - 心跳失效：设置
ping_interval=5, ping_timeout=2，然后在 mock 服务端不响应 ping，观察客户端是否抛ConnectionClosedError - 不要依赖
ws.open属性轮询判断状态——它在 close 过程中可能短暂为True，应以 recv/send 是否抛异常为准
在 pytest 中组织异步测试函数
pytest 默认不支持 async def 测试函数，直接运行会跳过或报 TypeError: object async_generator can't be used in 'await' expression。必须用插件或适配层。
推荐用 pytest-asyncio，但要注意它的默认模式是 function 级 event loop，多个测试并发跑时可能因共享 loop 导致状态污染。
- 安装：
pip install pytest-asyncio - 在
pytest.ini中声明：asyncio_mode = auto，并确保测试函数加@pytest.mark.asyncio - 避免在
setup_method里创建连接——pytest 的类级 setup 不兼容 async，一律改用async def test_xxx()+async with - 若需复用连接（如连续发多条消息），把整个交互链写在一个 test 函数里，不要拆成多个
await步骤分散在不同 test 中
WebSocket 测试最难的不是发消息，而是时间维度上的确定性：什么时候该收到、什么时候该断开、超时是否被正确归因。所有 await 都得配 timeout，所有 close 都要等 wait_closed，所有 mock 服务都要显式关——少一步，CI 就可能偶发失败。
