直接用 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。
- 用
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 或测试逻辑是否提前终止了连接。
立即学习“Python免费学习笔记（深入）”；
- 检查服务端 handler 是否在 recv 空时直接 return（这是正常结束），还是因未处理
ConnectionClosed异常而崩溃 - 客户端测试中，如果用
async with websockets.connect(...) as ws:，退出上下文会自动 close，此时服务端必然收到ConnectionClosedOK—— 这是预期行为，不用 assert 报错 - 若想模拟异常断连，可用
transport.close()（需从websocket.transport获取），但多数场景没必要；真要测容错，应在 handler 里故意 raise 异常并观察日志
