中文
devexamples

Pytest Asyncio Fixture Lifespan

Share async resources across tests with pytest-asyncio fixtures, scoping engine creation per session and cleaning up deterministically.

PythonCli Tooling

Code

snippet
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

Related Examples