别堆代码了！Pytest Fixture优雅搞定测试数据管理，附 CURD 全接口代码直接抄

深夜，你正在为明天的上线做最后的测试验证。

突然，一个本应通过的测试用例失败了。

你检查代码逻辑，一切正常；查看数据库，却发现前一个测试残留的数据干扰了当前测试的结果。

测试数据管理的困境

在软件开发中，“单元测试” 和 “集成测试” 是保障代码质量的基石。

但当多个测试共享同一个数据库时，一个常见的问题出现了：测试数据污染。

前一个测试创建的数据，会影响后一个测试的执行结果。

尤其在进行集成测试时，每个测试都希望有一个干净的测试环境，但现实中却往往要面对其他测试留下的“烂摊子”。

传统的解决方案是在每个测试开始前手动清理数据库，或在所有测试完成后统一清理。

但这两种方法都有明显缺陷：前者增加了大量重复代码，后者会导致测试之间的相互干扰。

Pytest Fixture：优雅的测试数据管理方案

Pytest 框架提供的 Fixture 功能，正是解决这一痛点的利器。

Fixture 允许我们在测试执行前后执行特定代码，完美实现了测试数据的自动创建与清理。

创建一个基础 Fixture

@pytest.fixture
def test_todo():
    todo = Todos(
        title="Learn to code",
        description="Need to learn everyday",
        priority=4,
        complete=False,
        owner_id=1
    )

    db = TestSessionLocal()
    db.add(todo)
    db.commit()
    yield todo

    with engine.connect() as connection:
        connection.execute(text("DELETE FROM todos;"))
        connection.commit()

Fixture 是 Pytest 的核心，能实现 “测试前准备数据，测试后清理数据”，完美解决数据隔离问题。

核心逻辑：

• 测试前：自动往测试数据库添加一条待办数据；
• 测试后：自动删除数据库中所有待办数据，保证下次测试是 “干净环境”；
• 注意：必须使用测试数据库，避免误删生产数据！

实战：完整测试示例

01、测试待办数据读取接口

在 test 目录下新建 test_todos.py 文件

# 查询存在的待办（正常场景）
def test_read_one_authenticated(test_todo):
    response = client.get("/todos/1")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == {
        "title": "Learn to code",
        "description": "Need to learn everyday",
        "priority": 4,
        "complete": False,
        "owner_id": 1,
        "id": 1
    }


# 查询不存在的待办（异常场景）
def test_read_one_authenticated_not_found():
    response = client.get("/todos/999")
    assert response.status_code == status.HTTP_404_NOT_FOUND
    assert response.json() == {"detail": "Todo not Found"}

02、测试待办数据创建接口

# 创建待办数据
def test_create_todo(test_todo):
    request_data = {
        "title": "New Todo",
        "description": "New todo description",
        "priority": 5,
        "complete": False,
    }
    response = client.post("/todos/", json=request_data)
    assert response.status_code == status.HTTP_201_CREATED

    # 验证数据确实被创建
    db = TestSessionLocal()
    model = db.query(Todos).filter(Todos.title == request_data["title"]).first()

    assert model.title == request_data["title"]
    assert model.description == request_data["description"]
    assert model.priority == request_data["priority"]
    assert model.complete == request_data["complete"]

03、测试待办数据更新接口

# 测试正常更新接口
def test_update_todo(test_todo):
    request_data = {
        "title": "Change the title of the todo",
        "description": "New todo description",
        "priority": 5,
        "complete": False,
    }
    response = client.put("/todos/1", json=request_data)
    assert response.status_code == status.HTTP_204_NO_CONTENT

    db = TestSessionLocal()
    model = db.query(Todos).filter(Todos.id == 1).first()
    assert model.title == request_data.get("title")
    assert model.description == request_data.get("description")


# 测试异常更新接口（数据没找到）
def test_update_todo_not_found(test_todo):
    request_data = {
        "title": "Change the title of the todo",
        "description": "New todo description",
        "priority": 5,
        "complete": False,
    }
    response = client.put("/todos/999", json=request_data)
    assert response.status_code == status.HTTP_404_NOT_FOUND
    assert response.json() == {"detail": "Todo not found."}

04、测试待办数据删除接口

# 测试正常删除接口
def test_delete_todo(test_todo):
    response = client.delete("/todos/1")
    assert response.status_code == status.HTTP_204_NO_CONTENT

    db = TestSessionLocal()
    model = db.query(Todos).filter(Todos.id == 1).first()
    assert model is None

# 测试异常删除接口（数据没找到）
def test_delete_todo_not_found():
    response = client.delete("/todos/999")
    assert response.status_code == status.HTTP_404_NOT_FOUND
    assert response.json() == {"detail": "Todo not found."}

05、测试管理员接口

在 test 目录下新建 test_admin.py 文件

from .utils import *
from ..routers.admin import get_db, get_current_user
from fastapi import status

app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_current_user] = override_get_current_user

# 测试获取所有待办数据接口
def test_admin_read_all_authenticated(test_todo):
    response = client.get("/admin/todos")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == [{
        "title": "Learn to code",
        "description": "Need to learn everyday",
        "priority": 4,
        "complete": False,
        "owner_id": 1,
        "id": 1
    }]


# 测试删除待办数据接口
def test_admin_delete_todo(test_todo):
    response = client.delete("/admin/todos/1")
    assert response.status_code == status.HTTP_204_NO_CONTENT

    db = TestSessionLocal()
    model = db.query(Todos).filter(Todos.id == 1).first()
    assert model is None

06、测试命令汇总

# 运行所有测试（禁用警告）
pytest --disable-warnings

# 运行特定测试文件
pytest test_todos.py --disable-warnings

# 显示详细测试信息
pytest -v --disable-warnings

拆分 utils.py，告别重复代码

我们把 “数据库连接、测试客户端、依赖覆盖、Fixture” 这些可复用的代码，统一放到test/utils.py中，后续所有测试文件直接导入即可。

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from ..main import app
from ..models import Base, Todos, Users
from passlib.context import CryptContext

SQLALCHEMY_DATABASE_URL = 'sqlite:///./testdb.db'

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def override_get_db():
    db = TestSessionLocal()
    try:
        yield db
    finally:
        db.close()


def override_get_current_user():
    return {"username": "wangerge", "id": 1, "user_role": "admin"}


client = TestClient(app)


@pytest.fixture
def test_todo():
    todo = Todos(
        title="Learn to code",
        description="Need to learn everyday",
        priority=4,
        complete=False,
        owner_id=1
    )

    db = TestSessionLocal()
    db.add(todo)
    db.commit()
    yield todo

    with engine.connect() as connection:
        connection.execute(text("DELETE FROM todos;"))
        connection.commit()

测试技巧总结

1. 数据隔离用 Fixture：通过yield关键字实现 “测试前准备、测试后清理”，再也不用担心数据库污染；
2. **重复代码放 utils.py **：把数据库连接、测试客户端、依赖覆盖集中管理，测试文件只关注业务逻辑；
3. 场景覆盖要全面：每个接口都测 “正常场景” 和 “异常场景”。

我们还有，用户接口应该怎么测试？认证功能怎么测试？token 创建功能怎么测试？

下期，我将掰开了，揉碎了，把它们一次性讲清楚。

想要获取本章完整代码，请在评论区回复 【FastAPI】，代码直接复制就能跑。

关于 FastAPI 的其他疑问

测试别踩坑！FastAPI隔离数据库+Mock用户，守住职场安全线

新功能上线就崩？Pytest三步测试法，让你的FastAPI稳如老狗，Bug率直降80%

加个字段，服务崩了？FastAPI新手避坑，Alembic三步搞定表结构变更！方案闭眼抄

别等着被骂：API上线前，一定要把SQLite换成MySQL，附 FastAPI对接代码

别等被骂才后悔：APP上线前，一定要把SQLite换成PostgreSQL，附 FastAPI对接代码

相关内容我都给大家做好了，感兴趣的朋友来「我的主页」找一找，直接就可以看到。

欢迎关注 「王二哥的技术笔记」，每天分享「Python」、「职场」有趣干货，千万不要错过！