使用Pytest测试时出现Runtime Error: Event loop is closed的解决求助

测试FastAPI应用时，test_register()和test_login()均可正常通过，但调用test_logout()并传递cookies时，出现RuntimeError: Event loop is closed错误。

相关代码 #

conftest.py #

...............

@pytest_asyncio.fixture(autouse=True, scope='session')
async def prepare_database():
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.create_all)
    yield
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.drop_all)


@pytest.fixture(scope='session')
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


client = TestClient(app)

...............

test_auth.py #

import asyncio
import pytest
import pytest_asyncio

from conftest import client


def test_register():
    response = client.post("/auth/register", json={
        "email": "user@example.com",
        "password": "string",
        "is_active": True,
        "is_superuser": False,
        "is_verified": False
    })
    assert response.status_code == 201


def test_login():
    data = {
        "username": "user@example.com",
        "password": "string",
    }
    encoded_data = "&".join([f"{key}={value}" for key, value in data.items()])
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    response = client.post("/auth/login", data=encoded_data, headers=headers)
    assert response.status_code == 204


def test_logout():
    cookies = {k: client.cookies.get(k) for k in client.cookies}
    print(cookies)
    headers = {
        "Content-Type": "application/json"
    }
    response = client.post("/auth/logout", headers=headers, cookies=cookies)
    print(response)
    assert response.status_code == 201

报错堆栈信息 #

platform linux -- Python 3.11.3, pytest-7.4.0, pluggy-1.2.0 -- /home/djamal/PycharmProjects/green/venv/bin/python
cachedir: .pytest_cache
rootdir: /home/djamal/PycharmProjects/green/server
plugins: anyio-3.7.1, asyncio-0.21.0
asyncio: mode=Mode.STRICT
collected 3 items

tests/test_auth.py::test_register PASSED                                                                         [ 33%]
tests/test_auth.py::test_login PASSED                                                                            [ 66%]
tests/test_auth.py::test_logout FAILED                                                                           [100%]

FAILURES ========================================================================================
test_logout ______________________________________________________________________________________

self = Connection<host=localhost,port=6379,db=0>, disable_decoding = False, timeout = None

async def read_response(
    self,
    disable_decoding: bool = False,
    timeout: Optional[float] = None,
    *,
    disconnect_on_error: bool = True,
):
    """Read the response from a previously sent command"""
    read_timeout = timeout if timeout is not None else self.socket_timeout
    host_error = self._host_error()
    try:
        if read_timeout is not None:
            async with async_timeout(read_timeout):
                response = await self._parser.read_response(
                    disable_decoding=disable_decoding
                )
        else:
            response = await self._parser.read_response(
                disable_decoding=disable_decoding
            )

../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:782:
_
../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:262: in read_response
    response = await self._read_response(disable_decoding=disable_decoding)
../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:270: in _read_response
    raw = await self._readline()
../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:344: in _readline
    data = await self._stream.readline()
/usr/lib/python3.11/asyncio/streams.py:545: in readline
    line = await self.readuntil(sep)
/usr/lib/python3.11/asyncio/streams.py:637: in readuntil
    await self._wait_for_data('readuntil')
_

self = <StreamReader transport=<_SelectorSocketTransport closing fd=18>>, func_name = 'readuntil'

async def _wait_for_data(self, func_name):
    """Wait until feed_data() or feed_eof() is called.

    If stream was paused, automatically resume it.
    """
    StreamReader uses a future to link the protocol feed_data() method
    to a read coroutine. Running two read coroutines at the same time
    would have an unexpected behaviour. It would not possible to know
    which coroutine would get the next data.
    if self._waiter is not None:
        raise RuntimeError(
            f'{func_name}() called while another coroutine is '
            f'already waiting for incoming data')

    assert not self._eof, '_wait_for_data after EOF'

    Waiting for data while paused will make deadlock, so prevent it.
    This is essential for readexactly(n) for case when n > self._limit.
    if self._paused:
        self._paused = False
        self._transport.resume_reading()

    self._waiter = self._loop.create_future()
    try:
        await self._waiter
E           RuntimeError: Task <Task pending name='anyio.from_thread.BlockingPortal._call_func' coro=<BlockingPortal._call_func() running at /home/djamal/PycharmProjects/green/venv/lib/python3.11/site-packages/anyio/from_thread.py:217> cb=[TaskGroup._spawn..task_done() at /home/djamal/PycharmProjects/green/venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py:661]> got Future  attached to a different loop

/usr/lib/python3.11/asyncio/streams.py:522: RuntimeError

During handling of the above exception, another exception occurred:

def test_logout():
    cookies = {k: client.cookies.get(k) for k in client.cookies}
    print(cookies)
    headers = {
        "Content-Type": "application/json"
    }
    response = client.post("/auth/logout", headers=headers, cookies=cookies)

tests/test_auth.py:38:
_
..........................
_

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

def _check_closed(self):
    if self._closed:
        raise RuntimeError('Event loop is closed')
E           RuntimeError: Event loop is closed

/usr/lib/python3.11/asyncio/base_events.py:519: RuntimeError
Captured stdout call ----------------------------------------------------------------------------------
{'fastapiusersauth': 'X5LEEnDGUSGXIhA5gOTQJSyJQ0g7xsKXvx4v2xBFCv8'}
short test summary info ================================================================================
FAILED tests/test_auth.py::test_logout - RuntimeError: Event loop is closed
1 failed, 2 passed in 1.00s ==============================================================================

修复方案 #

1. 避免全局创建TestClient，改用Fixture #

全局client会提前绑定到错误的事件循环，导致后续请求出现循环不匹配问题。修改conftest.py，将client改为function级别的fixture：

# conftest.py
...............

@pytest_asyncio.fixture(autouse=True, scope='session')
async def prepare_database():
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.create_all)
    yield
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.drop_all)


@pytest.fixture(scope='function')
def client():
    with TestClient(app) as test_client:
        yield test_client

# 移除原来的全局 client = TestClient(app)
...............

2. 简化test_logout的cookie传递 #

TestClient会自动维护会话cookie，无需手动提取和传递。修改test_auth.py：

# test_auth.py
import pytest

# 移除不需要的导入：asyncio, pytest_asyncio

def test_register(client):
    response = client.post("/auth/register", json={
        "email": "user@example.com",
        "password": "string",
        "is_active": True,
        "is_superuser": False,
        "is_verified": False
    })
    assert response.status_code == 201


def test_login(client):
    data = {
        "username": "user@example.com",
        "password": "string",
    }
    encoded_data = "&".join([f"{key}={value}" for key, value in data.items()])
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    response = client.post("/auth/login", data=encoded_data, headers=headers)
    assert response.status_code == 204


def test_logout(client):
    # 先执行登录获取cookie
    data = {
        "username": "user@example.com",
        "password": "string",
    }
    encoded_data = "&".join([f"{key}={value}" for key, value in data.items()])
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    client.post("/auth/login", data=encoded_data, headers=headers)
    
    # 直接调用logout，client会自动带上cookie
    headers = {
        "Content-Type": "application/json"
    }
    response = client.post("/auth/logout", headers=headers)
    print(response)
    assert response.status_code == 201

3. 调整事件循环配置（可选） #

如果仍有循环问题，移除自定义的event_loop fixture，改用pytest-asyncio的默认循环：

# conftest.py中移除以下代码：
# @pytest.fixture(scope='session')
# def event_loop(request):
#     loop = asyncio.get_event_loop_policy().new_event_loop()
#     yield loop
#     loop.close()

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