如何在Python中测试基于WebSocket的长连接_利用pytest-asyncio模拟客户端-Python教程-PHP中文网

不能直接用 pytest-asyncio 启动真实 websocket 客户端，它仅负责协程调度；需配合 websockets 等异步客户端库，通过 @pytest.mark.asyncio 标记 async 测试函数并利用其管理的事件循环完成连接、收发与异常验证。

如何在python中测试基于websocket的长连接_利用pytest-asyncio模拟客户端

pytest-asyncio 能否直接驱动 WebSocket 客户端测试？

不能直接用 pytest-asyncio 启动真实 WebSocket 客户端，它只负责协程调度，不提供网络客户端能力。你真正需要的是一个支持异步、可被 pytest 控制的 WebSocket 客户端库（比如 websockets），再配合 pytest-asyncio 的事件循环管理。

常见错误现象：RuntimeError: There is no current event loop in thread 或测试函数不 await 就返回 —— 本质是没正确标记 async 测试函数，或没启用 asyncio 插件。

必须在 pytest.ini 或 pyproject.toml 中启用 pytest-asyncio 插件（推荐配置 asyncio_mode = auto）
测试函数必须用 async def 声明，且 pytest 会自动为你注入运行中的 event loop
不要手动调用 asyncio.run()，否则会与 pytest 的 loop 冲突

用 websockets.connect() 模拟客户端连接并收发消息

websockets 是目前最轻量、最适配 pytest-asyncio 的 WebSocket 客户端库。它原生 async/await，无回调陷阱，连接、发送、接收都可精确控制。

典型使用场景：验证服务端是否正确响应 ping/pong、是否广播消息、是否在异常时关闭连接。

import pytest
import asyncio
import websockets

@pytest.mark.asyncio
async def test_ws_echo():
    uri = "ws://localhost:8000/ws"
    async with websockets.connect(uri) as ws:
        await ws.send("hello")
        resp = await ws.recv()
        assert resp == "hello"

@pytest.mark.asyncio 是关键标记，告诉插件这是 async 测试
async with websockets.connect(...) 自动处理握手、心跳、异常断连
await ws.recv() 默认阻塞直到有消息，适合单次交互；若需持续监听，改用 async for msg in ws:
注意：服务端必须已启动，否则测试会超时失败（默认 10 秒）

如何模拟多个并发客户端并验证广播行为？

WebSocket 测试的复杂点常在于状态共享和时序控制。比如验证“用户 A 发消息，B 和 C 都收到”——这需要同时运行多个 client 协程，并协调它们的收发时机。

Python 3.14.2

Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本，属于3.14系列的第二个维护更新。该版本包含了18项修复，重点解决了多进程、数据类及正则表达式等模块的回归问题，并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式（移除GIL）正式获得官方支持，是Python发展的重要里程碑。

下载

性能影响：每个 websockets.connect() 创建独立 TCP 连接，开销不大，但并发数过高可能触发本地端口耗尽（OSError: [Errno 24] Too many open files）。

用 asyncio.gather() 并发启动多个 client 协程，避免串行等待
用 asyncio.wait_for() 包裹 ws.recv()，防止某个 client 卡住拖垮整个测试
服务端若依赖 session ID 或连接顺序，测试前需清空状态（例如重置 mock 数据库或内存 dict）
不要依赖 sleep 等待消息到达；改用 asyncio.Event 或队列做信号同步

测试异常路径：连接拒绝、主动断连、协议错误怎么写？

真实长连接必然遇到网络抖动、认证失败、帧格式错误等边界情况。pytest-asyncio + websockets 可以精准触发这些异常，但写法容易出错。

常见错误现象：websockets.exceptions.ConnectionClosedError 没被捕获导致测试直接报错退出，而不是走断言逻辑。

连接拒绝（如服务未启动）：await websockets.connect("ws://localhost:9999") 会抛 ConnectionRefusedError，需用 pytest.raises(ConnectionRefusedError)
服务端主动关闭：在 await ws.recv() 后捕获 websockets.exceptions.ConnectionClosedOK
发送非法帧（如超长字符串）：某些服务端会立即断连，此时 recv 可能返回空或抛 ConnectionClosedError
注意：websockets 的异常类型较细，别用宽泛的 Exception 捕获，否则掩盖真实问题