Pytest 异步夹具生命周期

示例说明

异步测试的资源管理与同步测试同源但语法不同：夹具本身是协程，yield 前是准备、后是清理。会话级夹具让昂贵资源全程只建一次——数据库引擎、消息客户端都适用；用例级隔离夹具在其上叠加快照回滚，每个用例看到干净的收件箱，测试顺序彻底无关化。

技术要点

• session 作用域让昂贵资源只建一次
• 快照切片实现用例级状态隔离
• yield 之后的收尾断言必然执行

代码解读

mailer 夹具在会话开始创建伪造邮件客户端，结束时断言所有发出的邮件都有收件人——收尾自检把资源使用的不变式也纳入测试。clean_mailer 记录当前长度，用例结束后删除快照之后的记录。三个用例分别验证单发、隔离与 gather 并发发送。

使用建议

session 夹具里的状态会被全部用例共享，隔离逻辑要显式。event_loop 策略在新版本已由框架自动管理。清理代码别吞异常，失败要让测试红。

代码示例

import asyncio

import pytest
import pytest_asyncio


class FakeMailer:
    def __init__(self):
        self.sent: list[dict] = []

    async def send(self, to: str, subject: str) -> None:
        self.sent.append({'to': to, 'subject': subject})


@pytest_asyncio.fixture(scope='session')
async def mailer():
    client = FakeMailer()
    yield client
    assert all(item['to'] for item in client.sent)  # 收尾自检
    print(f'mailer sent {len(client.sent)} messages in total')


@pytest_asyncio.fixture
async def clean_mailer(mailer):
    snapshot = len(mailer.sent)
    yield mailer
    # 用例级隔离：只保留快照之前的历史
    del mailer.sent[:snapshot]


@pytest.mark.asyncio
async def test_welcome_email(clean_mailer):
    await clean_mailer.send('[email protected]', 'Welcome')
    assert clean_mailer.sent[-1]['subject'] == 'Welcome'


@pytest.mark.asyncio
async def test_reset_email(clean_mailer):
    await clean_mailer.send('[email protected]', 'Reset password')
    assert len(clean_mailer.sent) == 1  # 上一个用例的信件被隔离


@pytest.mark.asyncio
async def test_concurrent_sends(clean_mailer):
    await asyncio.gather(
        clean_mailer.send('[email protected]', 'One'),
        clean_mailer.send('[email protected]', 'Two'),
    )
    assert len(clean_mailer.sent) == 2