如何正确配置Pytest以实现FastAPI应用测试前后的PostgreSQL测试数据库全生命周期管理

看起来你遇到的核心问题有两个：一是setup_test_db fixture完全没触发，二是测试没用到测试数据库而是直接走了生产库。我来一步步帮你排查解决：

一、先解决Fixture不触发的核心问题 #

你的setup_test_db是异步Fixture，但当前代码有两个致命问题：

1. 异步函数调用错误：Fixture最后一行drop_test_db()是直接调用异步函数，没有用await，这会导致删除操作完全不执行，还可能埋下异步上下文隐患。
2. 缺少异步测试支持：Pytest本身不原生支持异步Fixture，需要安装pytest-asyncio插件才能识别并执行异步Fixture。

修复步骤： #

1. 先安装必要依赖：

pip install pytest-asyncio

2. 修改setup_test_db的清理逻辑，把同步调用改成异步等待：

# 原代码最后一行
drop_test_db()
# 改为
await drop_test_db()

二、确保测试用的是测试数据库 #

你的FastAPI应用里肯定有一个获取数据库会话的依赖（比如get_session），默认指向生产库的Engine。测试时必须覆盖这个依赖，让它指向测试库的Engine。

1. 确认配置正确性 #

检查config.py里的TEST_DATABASE_URI是否正确指向你要创建的测试库，格式示例：

# config.py
TEST_DATABASE_URI = f"postgresql+asyncpg://{DB_USER}:{DB_PASSWD}@{DB_HOST}/{DB_NAME}_test"

2. 在conftest.py中添加依赖覆盖Fixture #

假设你的FastAPI应用的数据库依赖是这样的：

# app/main.py 中的依赖
from sqlmodel.ext.asyncio.session import AsyncSession
from config import engine  # 生产库的Engine

async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

那你需要在conftest.py中添加Fixture来覆盖这个依赖：

# 新增到conftest.py
from app.main import get_session  # 导入你应用里的get_session依赖

@pytest.fixture(scope="function")
async def async_session(async_test_engine: AsyncEngine):
    # 创建测试用的异步会话
    async_session = sessionmaker(
        async_test_engine, class_=AsyncSession, expire_on_commit=False
    )
    async with async_session() as session:
        yield session

@pytest.fixture(scope="function")
async def client(async_session: AsyncSession):
    # 覆盖应用的get_session依赖，改用测试会话
    async def override_get_session():
        yield async_session
    app.dependency_overrides[get_session] = override_get_session
    
    # 创建异步测试客户端
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client
    
    # 测试结束后清除依赖覆盖
    app.dependency_overrides.clear()

3. 修改测试用例使用新的Client Fixture #

修改test_app.py的测试函数，直接使用client Fixture，无需手动创建：

# test_app.py
@pytest.mark.asyncio
async def test_root_route(client: AsyncClient):
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": config.WELCOME_MESSAGE}

三、完整修改后的conftest.py参考 #

import pytest
import asyncpg
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from alembic.config import Config
from alembic import command
from config import config
from app.main import app, get_session  # 导入你的app和数据库依赖

TEST_DATABASE_URI = config.SQLALCHEMY_DATABASE_URI_UNIT_TEST

# 1. 创建测试库
async def create_test_db():
    print("Creating test database")
    # 推荐连接到postgres默认库创建测试库，避免生产库不存在的问题
    conn = await asyncpg</think_never_used_51bce0c785ca2f68081bfa7d91973934>
        user=config.DB_USER, password=config.DB_PASSWD, database="postgres", host=config.DB_HOST
    )
    try:
        await conn.execute(f"CREATE DATABASE {config.DB_NAME}_test;")
        print("Database created successfully")
    except asyncpg.exceptions.DuplicateDatabaseError as e:
        print(f"Test DB already exists: {e}")
    finally:
        await conn.close()

# 2. 删除测试库
async def drop_test_db():
    print("Dropping test database")
    conn = await asyncpg.connect(
        user=config.DB_USER, password=config.DB_PASSWD, database="postgres", host=config.DB_HOST
    )
    try:
        # 先断开所有到测试库的连接，否则无法删除
        await conn.execute(f"""
            SELECT pg_terminate_backend(pg_stat_activity.pid)
            FROM pg_stat_activity
            WHERE pg_stat_activity.datname = '{config.DB_NAME}_test'
              AND pid <> pg_backend_pid();
        """)
        await conn.execute(f"DROP DATABASE IF EXISTS {config.DB_NAME}_test;")
        print("Test DB dropped successfully")
    finally:
        await conn.close()

# 3. 运行Alembic迁移
def run_migrations(db_uri, direction="upgrade", revision="head"):
    print(f"Running Alembic {direction} to {revision}")
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", db_uri)
    if direction == "upgrade":
        command.upgrade(alembic_cfg, revision)
    elif direction == "downgrade":
        command.downgrade(alembic_cfg, revision)

# 4. 测试库生命周期管理Fixture
@pytest.fixture(scope="session", autouse=True)
async def setup_test_db():
    print("Starting test DB setup")
    # 先清理旧的测试库
    await drop_test_db()
    # 创建新测试库
    await create_test_db()
    # 运行迁移
    run_migrations(TEST_DATABASE_URI, "upgrade", "head")

    yield  # 所有测试在这里执行

    # 清理工作
    print("Starting test DB cleanup")
    run_migrations(TEST_DATABASE_URI, "downgrade", "base")
    await drop_test_db()
    print("Test DB cleanup completed")

# 5. 测试用异步Engine Fixture
@pytest.fixture(scope="function")
async def async_test_engine():
    engine = create_async_engine(url=TEST_DATABASE_URI, echo=False)
    yield engine
    await engine.dispose()

# 6. 测试用异步会话Fixture
@pytest.fixture(scope="function")
async def async_session(async_test_engine: AsyncEngine):
    async_session = sessionmaker(
        async_test_engine, class_=AsyncSession, expire_on_commit=False
    )
    async with async_session() as session:
        yield session

# 7. 测试客户端Fixture（覆盖数据库依赖）
@pytest.fixture(scope="function")
async def client(async_session: AsyncSession):
    async def override_get_session():
        yield async_session
    app.dependency_overrides[get_session] = override_get_session
    
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client
    
    app.dependency_overrides.clear()

四、额外优化建议 #

1. 权限问题：确保你的数据库用户有创建/删除数据库、终止连接的权限，否则创建或删除测试库会失败。
2. 测试数据隔离：如果多个测试之间会互相影响数据，可以在function级别的fixture里用事务回滚（比如每个测试开启事务，结束时回滚），比每次重建库效率更高。
3. Alembic配置：如果你的alembic迁移脚本不在默认位置，需要在run_migrations函数里指定script_location参数，比如alembic_cfg.set_main_option("script_location", "alembic")。

现在运行pytest，你应该能看到终端打印出创建测试库、运行迁移的日志，测试会正确使用测试库，结束后自动清理。

备注：内容来源于stack exchange，提问作者Daniel