pytest-asyncio测试全通过但报RuntimeError: Event loop is closed错误

Windows环境下使用Python 3.10、pytest-asyncio对sqlalchemy[asyncmy]驱动的异步数据库逻辑进行测试时，所有测试用例均正常通过，但测试流程退出时抛出RuntimeError: Event loop is closed异常，报错信息如下：

================================ 2 passed in 0.32s ======================================
Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x00000183137D28C0>
Traceback (most recent call last):
  File "D:\Programs\development\Python\Python310\lib\asyncio\proactor_events.py", line 116, in __del__
    self.close()
  File "D:\Programs\development\Python\Python310\lib\asyncio\proactor_events.py", line 108, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "D:\Programs\development\Python\Python310\lib\asyncio\base_events.py", line 750, in call_soon
    self._check_closed()
  File "D:\Programs\development\Python\Python310\lib\asyncio\base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

涉及的原有配置和代码如下：

• pytest.ini配置

[pytest]
testpaths = test
asyncio_mode=strict

• 原有conftest.py中event_loop fixture

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

• 原有测试用例session fixture

@pytest.fixture(scope='session')
def session():
    default_mysql_config = mysql_config.get('default')
    db_url: str = f"mysql+asyncmy://{default_mysql_config.username}:" \
                  f"{urlquote(default_mysql_config.password)}@" \
                  f"{default_mysql_config.host}:{default_mysql_config.port}/" \
                  f"{default_mysql_config.database}"
    async_engine = create_async_engine(db_url, echo=True, future=True)
    async_session = sessionmaker(bind=async_engine, class_=AsyncSession, expire_on_commit=False, autoflush=False,
                                 autocommit=False)
    return async_session

• 资源未主动释放：创建的异步SQLAlchemy引擎持有数据库连接池，测试结束后没有主动调用dispose()方法释放连接资源，底层数据库连接对应的管道对象在事件循环关闭后才被垃圾回收，此时尝试调度清理逻辑就会访问已关闭的事件循环抛出异常。
• 事件循环关闭时序错误：自定义的session级event_loop fixture在yield后直接关闭循环，没有等待所有剩余异步清理任务执行完成，导致资源回收逻辑没有执行时机。
• Windows平台ProactorEventLoop已知兼容问题：Python 3.8+在Windows默认使用ProactorEventLoop，其底层管道传输对象的析构逻辑本身存在事件循环关闭后仍尝试调用循环方法的已知问题，该异常属于被解释器忽略的清理阶段异常，本身不影响测试结果正确性，但会污染命令行输出。

1. 改造session fixture为异步fixture，测试结束后主动释放数据库引擎

import pytest
import asyncio
from src.repository import user_repositroy
from src.database.dbconfig import mysql_config
from urllib.parse import quote_plus as urlquote
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker


@pytest.fixture(scope='session')
async def session():
    default_mysql_config = mysql_config.get('default')
    db_url: str = f"mysql+asyncmy://{default_mysql_config.username}:" \
                  f"{urlquote(default_mysql_config.password)}@" \
                  f"{default_mysql_config.host}:{default_mysql_config.port}/" \
                  f"{default_mysql_config.database}"
    async_engine = create_async_engine(db_url, echo=True, future=True)
    async_session = sessionmaker(
        bind=async_engine, 
        class_=AsyncSession, 
        expire_on_commit=False, 
        autoflush=False,
        autocommit=False
    )
    yield async_session
    # 测试会话结束后主动释放连接池所有资源
    await async_engine.dispose()

2. 调整event_loop fixture的关闭逻辑，先跑完所有待执行任务再关闭循环

@pytest.fixture(scope="session")
def event_loop():
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    # 收集所有未完成任务，等待执行完成后再关闭循环
    pending_tasks = asyncio.all_tasks(loop)
    loop.run_until_complete(asyncio.gather(*pending_tasks, return_exceptions=True))
    loop.close()

3. （兜底兼容方案）如果上述修改后仍有报错，在conftest.py最开头添加配置，将Windows下的事件循环策略切换为SelectorEventLoop，规避ProactorEventLoop的已知问题：

import sys
import asyncio

# Windows平台切换事件循环策略
if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

修改完成后重新执行pytest命令，测试通过后不会再出现该异常输出。

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