如何用Pytest在FastAPI+SQLAlchemy测试且不影响本地数据库数据

你当前的核心问题是：测试请求（比如/auth/register）仍在操作本地真实数据库，因为FastAPI应用本身的数据库会话依赖没有被替换为测试环境的会话。虽然你创建了独立测试库和会话，但应用端依然在用原有连接访问本地库。

核心修改思路 #

• 通过依赖注入覆盖应用的数据库会话获取器，让测试用TestClient使用测试数据库的会话
• 用事务包裹每个测试，结束后自动回滚，确保测试数据互不干扰且不残留

第一步：修改conftest.py配置 #

替换原有client fixture，添加依赖覆盖逻辑，同时优化会话清理机制：

from typing import AsyncIterator

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.main import app
from app.setup_db import Base
# 导入你项目中实际的session_getter依赖函数
from app.auth.models import session_getter

pytest_plugins = ("anyio",)

TEST_DB_URL = "postgresql+asyncpg://test:test@localhost:5432/test"


@pytest.fixture(scope="session")
def anyio_backend() -> str:
    return "asyncio"


@pytest.fixture(scope="session")
def test_engine() -> AsyncEngine:
    return create_async_engine(TEST_DB_URL, echo=True)


@pytest.fixture(scope="session")
def async_session_maker(test_engine) -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(bind=test_engine, expire_on_commit=False)


@pytest.fixture(scope="function")
async def session(async_session_maker) -> AsyncIterator[AsyncSession]:
    # 用事务包裹单个测试，结束后自动回滚
    async with async_session_maker() as session:
        async with session.begin():
            yield session
        await session.rollback()


@pytest.fixture(scope="function")
def client(session: AsyncSession) -> TestClient:
    # 覆盖应用的session_getter依赖，返回测试会话
    def override_get_session():
        yield session
    
    app.dependency_overrides[session_getter] = override_get_session
    with TestClient(app) as _client:
        yield _client
    # 测试结束后清除依赖覆盖，避免影响其他测试
    app.dependency_overrides.clear()


@pytest.fixture(scope="session", autouse=True)
async def setup_database(test_engine) -> AsyncIterator[None]:
    # 会话开始时创建测试库表结构
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 会话结束后销毁测试库表
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await test_engine.dispose()

第二步：确认User模型的依赖定义 #

确保你的session_getter是一个可被覆盖的独立依赖函数，示例如下：

# app/auth/models.py
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends
from app.setup_db import async_session  # 原项目的会话生成器
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

class User(SQLAlchemyBaseUserTable[int], Base):
    id: Mapped[int] = mapped_column(Integer, primary_key=True)

# 独立的会话获取依赖，用于被测试覆盖
async def session_getter() -> AsyncIterator[AsyncSession]:
    async with async_session() as session:
        yield session

@classmethod
async def get_user_db(cls, session: AsyncSession = Depends(session_getter)):
    yield SQLAlchemyUserDatabase(session, User)

第三步：优化测试代码（可选） #

测试时可直接验证测试库中的数据，确保隔离生效：

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import User

pytestmark = pytest.mark.anyio


async def test_database_setup(session: AsyncSession):
    result = await session.execute(text("SELECT 1"))
    assert result.scalar() == 1


async def test_client_request(client):
    response = client.get("/")
    assert response.status_code == 200


async def test_register(client, session: AsyncSession):
    register_data = {
        "email": "user@example.com",
        "password": "string",
        "is_active": True,
        "is_superuser": False,
        "is_verified": False,
    }
    response = client.post("/auth/register", json=register_data)
    assert response.status_code == 201
    
    # 从测试会话查询，确认用户创建在测试库而非本地库
    user_result = await session.execute(select(User).where(User.email == "user@example.com"))
    user = user_result.scalar_one_or_none()
    assert user is not None
    assert user.email == "user@example.com"

关键说明 #

1. 依赖隔离：通过app.dependency_overrides强制替换应用的数据库会话，确保所有测试请求都访问独立的测试库
2. 事务清理：每个测试用事务包裹，结束后自动回滚，避免测试数据残留，保证测试独立性
3. 环境隔离：测试库的表结构在会话开始时创建、结束时销毁，完全与本地真实数据库物理隔离

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