使用redis.asyncio与pytest-asyncio时遇Event loop is closed错误

我正在为基于asyncio的Python Redis应用编写测试，参照Redis官方《Redis作为文档数据库快速入门指南》中的Python代码改写为测试用例后，运行抛出两类RuntimeError：

• Event loop is closed
• Task got Future attached to a different loop

测试代码 #

import pytest

from redis.asyncio import Redis
from redis.commands.json.path import Path
from redis.commands.search import AsyncSearch
from redis.commands.search.field import NumericField, TagField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from redis.exceptions import ResponseError

@pytest.fixture(scope="session")
async def redis():
    redis = Redis(host="localhost", port=6379, db=0, decode_responses=True)
    yield redis
    await redis.aclose()

bicycle_ = {
    "brand": "Velorim",
    "model": "Jigger",
    "price": 270,
    "description": (
        "Small and powerful, the Jigger is the best ride "
        "for the smallest of tikes! This is the tiniest "
        "kids’ pedal bike on the market available without"
        " a coaster brake, the Jigger is the vehicle of "
        "choice for the rare tenacious little rider "
        "raring to go."
    ),
    "condition": "new",
}

bicycles_ = [
    bicycle_,
    {
        "brand": "Bicyk",
        "model": "Hillcraft",
        "price": 1200,
        "description": (
            "Kids want to ride with as little weight as possible."
            " Especially on an incline! They may be at the age "
            'when a 27.5" wheel bike is just too clumsy coming '
            'off a 24" bike. The Hillcraft 26 is just the solution'
            " they need!"
        ),
        "condition": "used",
    },
    {
        "brand": "Nord",
        "model": "Chook air 5",
        "price": 815,
        "description": (
            "The Chook Air 5  gives kids aged six years and older "
            "a durable and uberlight mountain bike for their first"
            " experience on tracks and easy cruising through forests"
            " and fields. The lower  top tube makes it easy to mount"
            " and dismount in any situation, giving your kids greater"
            " safety on the trails."
        ),
        "condition": "used",
    },
]

schema = (
    TextField("$.brand", as_name="brand"),
    TextField("$.model", as_name="model"),
    TextField("$.description", as_name="description"),
    NumericField("$.price", as_name="price"),
    TagField("$.condition", as_name="condition"),
)

@pytest.fixture(scope="session")
async def create_bicycle_index(redis: Redis):
    index = redis.ft("idx:bicycle")
    try:
        await index.dropindex()
    except ResponseError as e:
        pass
    await index.create_index(
        schema,
        definition=IndexDefinition(prefix=["bicycle:"], index_type=IndexType.JSON),
    )

@pytest.fixture(scope="session")
async def bicycles(redis: Redis):
    for bid, bicycle in enumerate(bicycles_):
        await redis.json().set(f"bicycle:{bid}", Path.root_path(), bicycle)

@pytest.fixture()
async def bicycle_idx(create_bicycle_index, redis: Redis):
    index = redis.ft("idx:bicycle")
    return index

async def test_search_all_bicycles(bicycles, bicycle_idx: AsyncSearch, redis: Redis):
    res = await bicycle_idx.search(Query("*"))
    assert res.total == 3

async def test_search_jigger_bicycle(bicycles, bicycle_idx: AsyncSearch, redis: Redis):
    res = await bicycle_idx.search(Query("@model:Jigger"))
    assert res.total == 1

错误详情 #

E           RuntimeError: Event loop is closed

/usr/lib/python3.12/asyncio/base_events.py:539: RuntimeError

E           RuntimeError: Task <Task pending name='Task-5' coro=<test_search_all_bicycles() running at /home/duranda/devel/redis-pytest/test_redis.py:94> cb=[_run_until_complete_cb() at /usr/lib/python3.12/asyncio/base_events.py:180]> got Future <Future pending> attached to a different loop

/usr/lib/python3.12/asyncio/streams.py:542: RuntimeError

已尝试的无效方案 #

• 添加自定义event_loop fixture未生效：

@pytest.fixture(scope="session")
def event_loop():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
    yield loop
    loop.close()

• 在pytest.ini配置asyncio_mode = auto，也尝试给所有测试函数添加@pytest.mark.asyncio标记，均无效。
• 环境依赖已正确配置：

pip install pytest-asyncio redis
docker run -d --name redis-stack-server -p 6379:6379 redis/redis-stack-server:latest

错误核心原因是会话级fixture绑定了早期创建的事件循环，但后续测试用例使用了不同循环，或循环被提前关闭导致资源无法访问，按以下步骤修复：

1. 强制使用全局会话级事件循环 #

修改event_loop fixture，创建独立的会话级循环，并延迟关闭以等待后台任务完成：

import asyncio

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    # 等待所有异步任务收尾后再关闭循环
    loop.run_until_complete(asyncio.sleep(0.1))
    loop.close()

2. 统一fixture作用域与依赖关系 #

• 将bicycle_idx的作用域改为session，避免每次测试重新创建索引实例绑定不同循环：

@pytest.fixture(scope="session")
async def bicycle_idx(create_bicycle_index, redis: Redis):
    index = redis.ft("idx:bicycle")
    return index

• 调整bicycles fixture的依赖顺序，确保索引创建后再插入数据，并添加清理逻辑：

@pytest.fixture(scope="session")
async def bicycles(redis: Redis, create_bicycle_index):
    # 插入测试数据
    for bid, bicycle in enumerate(bicycles_):
        await redis.json().set(f"bicycle:{bid}", Path.root_path(), bicycle)
    yield
    # 测试完成后清理数据
    for bid in range(len(bicycles_)):
        await redis.delete(f"bicycle:{bid}")

3. 显式绑定Redis连接到会话级循环 #

在redis fixture中指定循环，确保连接始终绑定到同一个会话级循环：

@pytest.fixture(scope="session")
async def redis(event_loop):
    redis = Redis(host="localhost", port=6379, db=0, decode_responses=True, loop=event_loop)
    yield redis
    await redis.aclose()

4. 升级pytest-asyncio版本 #

旧版本可能存在会话级fixture的循环管理bug，升级到最新版：

pip install --upgrade pytest-asyncio

最终完整测试代码 #

import asyncio
import pytest

from redis.asyncio import Redis
from redis.commands.json.path import Path
from redis.commands.search import AsyncSearch
from redis.commands.search.field import NumericField, TagField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from redis.exceptions import ResponseError

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.run_until_complete(asyncio.sleep(0.1))
    loop.close()

@pytest.fixture(scope="session")
async def redis(event_loop):
    redis = Redis(host="localhost", port=6379, db=0, decode_responses=True, loop=event_loop)
    yield redis
    await redis.aclose()

bicycle_ = {
    "brand": "Velorim",
    "model": "Jigger",
    "price": 270,
    "description": (
        "Small and powerful, the Jigger is the best ride "
        "for the smallest of tikes! This is the tiniest "
        "kids’ pedal bike on the market available without"
        " a coaster brake, the Jigger is the vehicle of "
        "choice for the rare tenacious little rider "
        "raring to go."
    ),
    "condition": "new",
}

bicycles_ = [
    bicycle_,
    {
        "brand": "Bicyk",
        "model": "Hillcraft",
        "price": 1200,
        "description": (
            "Kids want to ride with as little weight as possible."
            " Especially on an incline! They may be at the age "
            'when a 27.5" wheel bike is just too clumsy coming '
            'off a 24" bike. The Hillcraft 26 is just the solution'
            " they need!"
        ),
        "condition": "used",
    },
    {
        "brand": "Nord",
        "model": "Chook air 5",
        "price": 815,
        "description": (
            "The Chook Air 5  gives kids aged six years and older "
            "a durable and uberlight mountain bike for their first"
            " experience on tracks and easy cruising through forests"
            " and fields. The lower  top tube makes it easy to mount"
            " and dismount in any situation, giving your kids greater"
            " safety on the trails."
        ),
        "condition": "used",
    },
]

schema = (
    TextField("$.brand", as_name="brand"),
    TextField("$.model", as_name="model"),
    TextField("$.description", as_name="description"),
    NumericField("$.price", as_name="price"),
    TagField("$.condition", as_name="condition"),
)

@pytest.fixture(scope="session")
async def create_bicycle_index(redis: Redis):
    index = redis.ft("idx:bicycle")
    try:
        await index.dropindex()
    except ResponseError:
        pass
    await index.create_index(
        schema,
        definition=IndexDefinition(prefix=["bicycle:"], index_type=IndexType.JSON),
    )
    yield
    # 测试完成后清理索引
    try:
        await index.dropindex()
    except ResponseError:
        pass

@pytest.fixture(scope="session")
async def bicycles(redis: Redis, create_bicycle_index):
    for bid, bicycle in enumerate(bicycles_):
        await redis.json().set(f"bicycle:{bid}", Path.root_path(), bicycle)
    yield
    # 清理测试数据
    for bid in range(len(bicycles_)):
        await redis.delete(f"bicycle:{bid}")

@pytest.fixture(scope="session")
async def bicycle_idx(create_bicycle_index, redis: Redis):
    return redis.ft("idx:bicycle")

async def test_search_all_bicycles(bicycles, bicycle_idx: AsyncSearch):
    res = await bicycle_idx.search(Query("*"))
    assert res.total == 3

async def test_search_jigger_bicycle(bicycles, bicycle_idx: AsyncSearch):
    res = await bicycle_idx.search(Query("@model:Jigger"))
    assert res.total == 1

内容的提问来源于stack exchange，提问作者DurandA