FastAPI异步测试遇RuntimeError: Event loop is closed问题求助

问题根源 #

多个异步测试用例运行时，测试框架（pytest+unittest）可能复用或重建事件循环，但默认数据库连接池（如SQLAlchemy的QueuePool）会绑定到第一个用例的事件循环。当第一个用例结束后循环被关闭，后续用例尝试使用绑定到已关闭循环的连接，就会触发RuntimeError: Event loop is closed和AttributeError: 'NoneType' object has no attribute 'send'异常。NullPool能解决是因为它不会复用连接，但生产环境用它会严重影响性能。

适配生产的解决方案 #

1. 分环境动态配置连接池

通过环境变量区分生产/测试环境，生产用默认高效连接池，测试用NullPool或临时池策略：

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool
import os

def get_db_engine():
    env = os.getenv("APP_ENV", "production")
    if env == "test":
        # 测试环境用NullPool，避免连接复用导致的循环绑定问题
        return create_async_engine(
            "sqlite+aiosqlite:///test.db",
            poolclass=NullPool,
            echo=False
        )
    else:
        # 生产环境保留默认QueuePool，保证连接复用性能
        return create_async_engine(
            "postgresql+asyncpg://user:pass@db-host:5432/dbname",
            pool_size=10,
            max_overflow=20
        )

2. 用pytest Fixture管理测试资源生命周期

为每个测试用例创建独立的数据库引擎和会话，用例结束后彻底销毁资源：

import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.pool import NullPool
from your_app.models import Base

@pytest.fixture(scope="function")
async def test_db_session():
    # 为每个测试用例创建独立的内存SQLite引擎（带NullPool）
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        poolclass=NullPool
    )
    # 初始化测试表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    # 生成数据库会话
    async with AsyncSession(engine) as session:
        yield session
    # 测试结束后销毁引擎，释放所有连接
    await engine.dispose()

测试用例直接注入test_db_session即可，每个用例的资源完全隔离。

3. 修复事件循环复用问题

如果使用pytest-asyncio，确保事件循环配置正确，避免复用已关闭的循环：

• 在项目根目录的pytest.ini中添加：

  [pytest]
  asyncio_mode = auto
  

• 或者在conftest.py中手动管理全局事件循环：

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

4. Unittest异步测试的资源清理

如果用IsolatedAsyncioTestCase，重写asyncTearDown确保每个用例结束后关闭连接：

from unittest import IsolatedAsyncioTestCase
from your_app.db import get_db_engine
from sqlalchemy.ext.asyncio import AsyncSession

class TestAPIEndpoints(IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.engine = get_db_engine()  # 测试环境会自动用NullPool
        self.session = AsyncSession(self.engine)

    async def asyncTearDown(self):
        await self.session.close()
        await self.engine.dispose()

    async def test_create_item(self):
        # 测试逻辑...
        pass

核心思路 #

测试环境让连接池生命周期与单个测试用例绑定，用后即销毁；生产环境保留连接复用的池策略保证性能。通过环境配置和测试资源隔离，既解决测试报错问题，又不影响生产环境的运行效率。

内容的提问来源于stack exchange，提问作者Super Ultra Noob