如何使用peewee_async与pytest创建事务性测试Fixture

要在pytest中为peewee_async搭建事务性测试Fixture，核心逻辑是给每个测试用例单独开启异步事务，测试结束后自动回滚，以此保证测试间的数据库状态完全隔离，不会留下脏数据影响后续测试。

结合你的代码思路，我整理了完整可运行的实现方案：

import pytest
from app.db.clients import (
    get_database,  # 返回PooledMySQLDatabase类的实例
    BaseManager    # 继承自peewee_async.Manager类
)

# 初始化数据库连接与异步操作管理器
db = get_database()
db.set_allow_sync(False)  # 禁用同步操作，强制使用异步模式避免混淆
objects = BaseManager(database=db)

@pytest.fixture(scope="function")
async def transactional():
    """事务性Fixture：测试执行在事务内，结束后自动回滚"""
    async with objects.transaction() as trx:
        yield trx  # 让测试代码在事务上下文里执行
        await trx.rollback()  # 测试完成后回滚，清除所有测试产生的数据

# 示例异步测试用例
@pytest.mark.asyncio
@pytest.mark.usefixtures('transactional')
async def test_create_test_user():
    # 假设你有一个基于peewee_async.AsyncModel的User模型
    from app.models import User
    
    # 在事务内创建测试数据
    test_user = await objects.create(User, username="temp_user", email="temp@test.com")
    
    # 验证数据创建成功
    fetched_user = await objects.get(User, id=test_user.id)
    assert fetched_user.username == "temp_user"
    
    # 测试结束后，Fixture会自动回滚，这条临时数据不会留在数据库中

关键细节说明： #

• Fixture作用域：这里设置scope="function"，意味着每个测试函数都会启动独立事务，实现最严格的测试隔离。如果需要在测试类内共享事务，可以改成scope="class"。
• 异步事务上下文：objects.transaction()是peewee_async提供的异步上下文管理器，能确保yield前后的所有数据库操作都处于同一个事务中。
• 强制异步模式：db.set_allow_sync(False)可以避免误写同步API导致的事务失效问题，确保所有操作都是异步的。
• 测试标记：异步测试必须加上@pytest.mark.asyncio，pytest才能正确识别并执行异步代码。

注意事项： #

• 测试中所有数据库操作必须通过objects这个Manager执行，否则操作会脱离Fixture的事务上下文，导致回滚失效。
• 测试代码里不要主动调用await trx.commit()，否则回滚操作无法恢复已提交的数据，破坏测试隔离性。

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