FastAPI测试报错AssertionError: DatabaseBackend未在测试中运行

问题根源拆解 #

1. 异步环境不匹配：用同步的create_engine和sessionmaker创建测试会话，但后端是异步架构，导致会话未正确绑定到异步请求上下文，触发后端的测试模式校验失败。
2. 依赖未注入：定义了test_db fixture，但没有将其替换为接口依赖的get_db，请求上下文里仍然使用生产环境的数据库会话，后端检测不到测试状态。
3. Fixture跨文件未加载：pytest只会自动加载conftest.py中的fixture，普通文件里的fixture无法被其他测试文件识别。

分步解决方案 #

1. 适配异步测试数据库会话 #

如果后端使用SQLAlchemy异步模式，必须用异步引擎和会话工厂替换同步实现：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# 替换为你的异步测试数据库URL（比如postgresql+asyncpg://xxx）
TEST_DATABASE_URL = "你的测试数据库地址"

# 创建异步引擎和会话工厂
test_engine = create_async_engine(TEST_DATABASE_URL, echo=True)
TestAsyncSessionLocal = sessionmaker(
    bind=test_engine, class_=AsyncSession, expire_on_commit=False
)

2. 替换接口依赖的数据库会话 #

将生产环境的get_db依赖替换为测试用的异步会话，确保请求上下文注入测试DB：

from your_app_module import app, get_db

async def override_get_db():
    async with TestAsyncSessionLocal() as session:
        yield session

# 覆盖原依赖
app.dependency_overrides[get_db] = override_get_db

3. 修正test_db Fixture #

调整fixture为异步模式，确保会话正确管理：

import pytest

@pytest.fixture(scope="function")
async def test_db():
    async with TestAsyncSessionLocal() as session:
        # 可选：测试前初始化数据
        # await session.execute(...)
        yield session
        # 测试后回滚数据，避免污染
        await session.rollback()

4. 修复异步路由测试 #

如果接口是异步路由，测试函数需改为异步并添加pytest.mark.asyncio装饰：

from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

@pytest.mark.asyncio
async def test_read_company(test_db):
    company_id = "1"
    response = client.get(f"/companies/{company_id}/")
    assert response.status_code == 200

5. 解决Fixture跨文件问题 #

将test_db fixture及依赖替换代码放到项目根目录的conftest.py文件中，pytest会自动加载该文件中的所有fixture，其他测试文件无需手动导入即可直接使用。

示例conftest.py：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import pytest
from fastapi.testclient import TestClient
from your_app_module import app, get_db

TEST_DATABASE_URL = "你的异步测试数据库URL"

test_engine = create_async_engine(TEST_DATABASE_URL)
TestAsyncSessionLocal = sessionmaker(
    bind=test_engine, class_=AsyncSession, expire_on_commit=False
)

async def override_get_db():
    async with TestAsyncSessionLocal() as session:
        yield session

app.dependency_overrides[get_db] = override_get_db

@pytest.fixture(scope="function")
async def test_db():
    async with TestAsyncSessionLocal() as session:
        yield session
        await session.rollback()

@pytest.fixture(scope="module")
def client():
    return TestClient(app)

测试文件简化为：

import pytest

@pytest.mark.asyncio
async def test_read_company(client, test_db):
    company_id = "1"
    response = client.get(f"/companies/{company_id}/")
    assert response.status_code == 200

额外注意事项 #

• 确保已安装异步数据库驱动（如PostgreSQL需pip install asyncpg，MySQL需pip install aiomysql）。
• 如果后端有自定义DatabaseBackend类，需检查其测试模式校验逻辑，确保测试时通过app.state或环境变量标记测试状态（如app.state.is_test = True）。

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