如何使用pytest测试与数据库交互的CRUD函数？

这类数据库交互函数是否需要测试？ #

必须测试。这类函数直接对接数据库，涉及SQL语句正确性、参数绑定逻辑、结果格式处理等关键环节，一旦出错会直接导致业务数据查询异常。比如SQL拼写错误、参数传递遗漏（像你代码里的(user_id, )多一个逗号是为了避免元组歧义，这个细节如果写错就会报错）、DictCursor是否正确返回字典结构等，都需要通过测试验证。

前置步骤：调用create_user()还是预填充数据？ #

优先选择调用业务层的create_user()，原因如下：

• 对齐真实业务流程：测试场景和生产环境中用户的创建逻辑完全一致，能确保测试结果的可信度。如果预填充数据的逻辑和create_user()不一致（比如create_user会自动处理密码加密、默认字段赋值，但预填充SQL没做），测试就失去了意义。
• 联动测试：如果create_user()本身存在bug，通过这种方式能提前发现，避免后续单独测试create_user()时才暴露问题。
• 维护成本低：后续create_user()逻辑更新时，测试用例无需修改，自动适配新的用户创建规则。

预填充数据（比如直接执行INSERT SQL）仅适合极端场景（比如create_user()依赖复杂外部服务，测试时无法调用），但缺点是容易和业务逻辑脱节，维护成本高。

pytest测试实现示例 #

1. 用Fixture管理测试数据的生命周期 #

使用pytest的fixture来创建测试用户，并在测试结束后清理，避免污染测试数据库：

import pytest
from your_module import get_user_by_id, create_user, get_db

@pytest.fixture(scope="function")
def test_user():
    # 前置：创建测试用户
    user_info = {"name": "测试用户", "email": "test@example.com"}
    user_id = create_user(user_info)
    yield user_id  # 将用户ID传递给测试用例
    
    # 后置：清理测试用户
    db = get_db()
    cursor = db.cursor()
    cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
    db.commit()
    cursor.close()

2. 编写测试用例 #

覆盖正常查询和异常查询两种场景：

def test_get_user_by_id_success(test_user):
    # 测试存在的用户ID
    user = get_user_by_id(test_user)
    
    assert user is not None
    assert user["id"] == test_user
    assert user["name"] == "测试用户"
    assert user["email"] == "test@example.com"

def test_get_user_by_id_nonexistent():
    # 测试不存在的用户ID
    user = get_user_by_id(999999)
    
    assert user is None

3. 优化：用事务回滚替代手动清理 #

如果数据库支持事务，可以用事务回滚来提高测试效率，无需手动删除数据：

@pytest.fixture(scope="function")
def db_transaction():
    db = get_db()
    db.begin()  # 开启事务
    yield db
    db.rollback()  # 测试结束后回滚事务，数据不持久化
    db.close()

# 此时create_user和get_user_by_id会使用这个事务内的连接，测试后自动回滚

关键注意事项 #

• 必须使用测试专用数据库，绝对不能直接操作生产数据库，避免数据风险。
• 测试数据库最好每次测试前初始化（比如清空表结构重建），确保测试环境干净，避免之前的测试数据影响结果。
• 如果是团队协作，可统一配置测试数据库的连接信息，确保所有成员的测试环境一致。

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