33 使用测试数据库

在上一篇中，我们介绍了如何使用 pytest 进行 FastAPI 应用的测试。在进行测试时，尤其是涉及到数据库的功能时，我们需要一个快速且可控的方式来测试数据库交互。为此，我们将创建一个 测试数据库。

什么是测试数据库？

测试数据库是专门用于测试目的的数据库实例，它通常与生产环境和开发环境数据库分开。通过使用测试数据库，我们可以：

• 避免对生产数据造成影响。
• 使测试结果独立于开发和生产环境。
• 更容易地创建和销毁测试数据。

在本教程中，我们将使用 SQLite 作为我们的测试数据库，因为它简单且易于设置。接下来我们将展示如何将其集成到我们的 FastAPI 测试中。

设置测试数据库

首先，我们需要安装必要的依赖包。确保你的 requirements.txt 中包含以下行：

fastapi
pytest
httpx
sqlalchemy
databases

然后，我们可以创建一个 conftest.py 文件，来配置测试数据库。

conftest.py 示例

在项目根目录下，创建一个名为 conftest.py 的文件：

import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from app.models import Base
from app.main import app  # 确保导入你的FastAPI应用实例
import databases

# 创建一个测试数据库连接
DATABASE_URL = "sqlite+aiosqlite:///./test.db"  # 使用SQLite作为测试数据库
engine = create_async_engine(DATABASE_URL, echo=True)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="session")
async def setup_database():
    # 创建数据库表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    yield

    # 在测试结束后，销毁数据库
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def db_session(setup_database):
    # 创建一个新的数据库会话
    async_session: AsyncSession = TestingSessionLocal()
    yield async_session
    await async_session.close()

这里，我们使用了 pytest 的 fixture 来设置和清理测试数据库。setup_database fixture 会在测试会话开始时创建所有数据库表，并在测试完成后删除它们。

使用测试数据库进行测试

接下来，我们需要编写测试用例，以确保我们的 FastAPI 应用与数据库交互正常。在此例中，我们将测试一个简单的 CRUD 操作。

假设的模型和API

假设我们有一个 User 模型和相应的 FastAPI 路由。首先，我们定义模型：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

然后，我们在 FastAPI 中定义 API 路由：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from app.database import get_db  # 假设有一个获取DB的依赖函数

app = FastAPI()

@app.post("/users/", response_model=User)
async def create_user(user: User, db: Session = Depends(get_db)):
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

编写测试用例

现在，我们来编写一个简单的测试，来验证我们的 create_user 路由。

在你的测试文件中（比如 test_api.py），添加以下内容：

import pytest
from httpx import AsyncClient
from app.main import app
from app.models import User

@pytest.mark.asyncio
async def test_create_user(db_session):
    async with AsyncClient(app=app, base_url="http://test") as client:
        user_data = {"name": "Test User"}

        response = await client.post("/users/", json=user_data)

        assert response.status_code == 200
        assert response.json()["name"] == user_data["name"]

        # 验证数据库中是否插入了该用户
        result = await db_session.execute(select(User).where(User.name == "Test User"))
        user = result.scalars().first()
        assert user is not None
        assert user.name == "Test User"

在这个测试中，我们首先使用 AsyncClient 发送一个 POST 请求来创建用户。然后，我们验证响应的 status_code 和返回的数据。最后，我们通过与数据库会话的交互，检查用户是否实际被插入到测试数据库中。

总结

通过使用一个独立的测试数据库，我们可以确保我们的 FastAPI 应用在运行测试时不会影响生产数据。这种方法也提高了测试的可靠性和可维护性。接下来，在下一篇教程中，我们将讨论如何 选择 FastAPI 应用的部署方式。保持关注！