
本文详解如何通过正确配置 pytest-asyncio 的 loop_scope 与 fixture 作用域,实现跨测试复用同一个异步 TCP 服务器,避免重复启停开销,并解决因事件循环不匹配导致的 RuntimeError: Task got Future attached to a different loop 错误。
本文详解如何通过正确配置 `pytest-asyncio` 的 loop_scope 与 fixture 作用域,实现跨测试复用同一个异步 tcp 服务器,避免重复启停开销,并解决因事件循环不匹配导致的 `runtimeerror: task got future attached to a different loop` 错误。
在使用 pytest 编写异步网络测试时,一个常见需求是:让多个测试共享同一个 TCP 服务器实例,而非每个测试都启动/关闭一次——这不仅能显著提升测试执行速度,还能更真实地模拟长连接服务场景。但直接将 @pytest.fixture(scope="module") 应用于异步 fixture(如 async def tcp_server())会导致运行时错误:
RuntimeError: Task ... got Future attached to a different loop
该错误的根本原因是:pytest-asyncio 默认为每个 async fixture 创建独立的事件循环(loop),而 scope="module" 的 fixture 生命周期跨越多个测试函数,若其内部 asyncio.Event、asyncio.Task 等对象绑定到某个特定 loop,后续测试中若在另一个 loop 中 await 它们,就会触发跨 loop 引用异常。
✅ 正确解法是显式统一 fixture 的事件循环作用域(loop_scope) 与 fixture 自身的作用域(scope),确保二者协同一致。
✅ 推荐配置(稳定可靠)
# test_tcp.py
import pytest
import pytest_asyncio
# 全局设置:所有 async 测试使用 session 级别事件循环
pytestmark = pytest.mark.asyncio(loop_scope="session")
@pytest_asyncio.fixture(loop_scope="session", scope="module")
async def tcp_server():
server = TCPServer('localhost', 3102)
# 启动服务器任务(注意:不 await,仅创建 task)
task = asyncio.create_task(server.start())
# 确保服务器已就绪(可选:加简短等待或健康检查)
await asyncio.sleep(0.05)
yield server
# 清理:取消任务并等待优雅退出
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
⚠️ 关键点说明:
- 必须使用
@pytest_asyncio.fixture(而非原生@pytest.fixture);loop_scope="session"表示整个测试会话共用同一个事件循环;scope="module"表示 fixture 在模块内只初始化一次,供该模块下所有测试复用;loop_scope必须 ≥scope(即"session" > "module" > "function"),否则 pytest-asyncio 无法保证 loop 复用一致性。
? 测试函数写法(保持简洁)
async def test_motor_identification_dataFL(tcp_server):
tcp_server.set_header_function(lambda: create_motor_identification_header(110))
try:
await asyncio.wait_for(tcp_server.data_received_event.wait(), timeout=5.0)
except asyncio.TimeoutError:
pytest.fail("Timeout waiting for client data")
assert tcp_server.last_received_data == b'\x00\x06'
无需额外装饰器(因已通过 pytestmark 全局声明),也无需手动管理 loop。
? 常见误区与规避建议
- ❌ 错误:混合使用
@pytest.fixture和async def—— 必须改用@pytest_asyncio.fixture; - ❌ 错误:
loop_scope="function"+scope="module"—— loop 生命周期短于 fixture,必然报错; - ⚠️ 注意:
TCPServer.start()是无限serve_forever(),务必通过task.cancel()主动终止,否则测试进程无法退出; - ? 提升健壮性:可在
yield前添加简易连接探测(如await asyncio.open_connection(...)),确保端口已监听再继续; - ? 环境要求:确保安装
pytest-asyncio>=0.24.0(支持loop_scope参数),并禁用旧版兼容警告(在pyproject.toml中添加):[tool.pytest.ini_options] asyncio_default_fixture_loop_scope = "session"
通过以上配置,你就能安全、高效地在多个 pytest 测试间复用同一个异步 TCP 服务器,兼顾性能与稳定性。










