运行时错误：将 redis.asyncio 与 pytest-asyncio 一起使用时事件循环关闭

Question

我正在尝试为使用 Redis 的基于异步的 Python 应用程序编写测试。

我从官方Redis作为文档数据库快速入门指南复制了Python代码，并把它变成了测试：

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-geospatial/app/tests/test_app.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

我尝试在模块顶部添加以下代码来解决此错误；没有成功。

@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-asyncio

和

redis-py

并运行 redis-stack 服务器才能测试上述代码，例如：

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

我还在

pytest.ini

中添加了以下内容，以便不必使用

@pytest.mark.asyncio

（我也尝试用

@pytest.mark.asyncio

标记我的所有功能）：

[pytest]
asyncio_mode = auto

Answer 1

这个答案存在问题，我仍在寻找更好的解决方案。

我能够通过将 event_loop 夹具传递到

redis

夹具来

解决此问题

：

@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.fixture(scope="session")
async def redis(event_loop):
    redis = Redis(host="localhost", port=6379, db=0, decode_responses=True)
    yield redis
    await redis.aclose()

但是，这会引发一些警告：

venv/lib/python3.12/site-packages/pytest_asyncio/plugin.py:229
  /home/duranda/devel/redis-geospatial/venv/lib/python3.12/site-packages/pytest_asyncio/plugin.py:229: PytestDeprecationWarning: redis is asynchronous and explicitly requests the "event_loop" fixture. Asynchronous fixtures and test functions should use "asyncio.get_running_loop()" instead.
    warnings.warn(

app/tests/test_app.py::test_search_all_bicycles
  /home/duranda/devel/redis-test/venv/lib/python3.12/site-packages/pytest_asyncio/plugin.py:751: DeprecationWarning: The event_loop fixture provided by pytest-asyncio has been redefined in
  /home/duranda/devel/redis-test/app/tests/test_app.py:13
  Replacing the event_loop fixture with a custom implementation is deprecated
  and will lead to errors in the future.
  If you want to request an asyncio event loop with a scope other than function
  scope, use the "scope" argument to the asyncio mark when marking the tests.
  If you want to return different types of event loops, use the event_loop_policy
  fixture.