显式优于隐式:掌握 Pytest fixture 与异步测试

发布: (2025年12月3日 GMT+8 17:03)
6 min read
原文: Dev.to

Source: Dev.to

问题:事件循环让人困惑

说到异步 Python:你的 async/await 代码需要一个事件循环来运行。这不是可选的。而且市面上有多种事件循环实现:

  • asyncio – Python 内置的标准库解决方案
  • trio – 第三方库,提供结构化并发
  • curio – 另一个第三方选项(如今较少使用)

在使用 pytest 测试异步代码时,你需要某种东西来管理这些事件循环。于是 pytest-anyio 出场了——一个本应让你更轻松的插件。剧透一下:它并不总是如此。

那么,Pytest Fixture 到底是什么?

在我们深入异步的疯狂之前,先聊聊 fixture。如果你用过 pytest,应该见过它们。如果你还没弄懂它们,你并不孤单。

Fixture 只是为你的测试准备环境的函数。 它们在测试之前运行,提供测试所需的东西,之后再进行清理。把它们想象成房间里负责的成年人。

依赖注入(好的一种)

def test_something(database_connection):
    # pytest 会自动给你连接对象
    result = database_connection.query("SELECT * FROM users")
    assert result is not None

无需手动设置。无需全局变量。只要声明你需要的东西,pytest 就会处理。

无需样板代码的 Setup 与 Teardown

import pytest

@pytest.fixture
async def create_test_tables():
    # Setup: 在测试前运行
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield  # 这里执行你的测试

    # Teardown: 在测试后运行
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)

yield 之前的代码是准备工作,之后的代码是清理工作。每次测试都会得到一个干净的数据库。

作用域:因为并非所有东西都需要每次都运行

  • function(默认) – 每个测试都运行一次
  • class – 每个测试类运行一次
  • module – 每个测试文件运行一次
  • session – 整个测试套件运行一次

选择合适的作用域对性能至关重要。

anyio 后端问题

当你使用 pytest-anyio 测试异步代码时,它需要知道使用哪个事件循环。默认情况下,它会尝试“帮助”你,针对 所有可用的后端 进行测试。

实际发生的情况:

  1. 你使用 asyncio 编写测试(因为 FastAPI 使用它)。
  2. 你安装了 pytest-anyio 来测试异步代码。
  3. pytest-anyio 发现你系统里装有 trio(可能是别的项目的依赖)。
  4. 它尝试用 trio 运行你的测试。
  5. 你的 aiosqlite 驱动炸掉了,因为它只能在 asyncio 环境下工作。

于是出现 ModuleNotFoundError: No module named 'trio'RuntimeError: no running event loop 等错误,让你开始怀疑自己的职业选择。

解决方案:明确指定

别再让 pytest-anyio 猜测了。明确告诉它你想要的后端。

步骤 1:创建 anyio_backend Fixture

# tests/conftest.py
import pytest

@pytest.fixture(scope="session")
def anyio_backend():
    """
    所有异步测试都使用 asyncio。就这么简单。
    """
    return "asyncio"

该 fixture 在整个测试会话期间运行一次,告诉 pytest-anyio:“使用 asyncio。别再聪明了。”

步骤 2:确保 trio 不会干扰

如果你不需要 trio,直接卸载它:

pip uninstall trio

步骤 3:正确标记你的异步测试

import pytest

@pytest.mark.anyio(backend='asyncio')
async def test_create_and_get_example(client, create_test_tables):
    # 你的测试代码
    pass

即使有了 anyio_backend fixture,backend='asyncio' 参数也是多余的,但它能让意图更加明确。

实际案例:使用 FastAPI 与 SQLModel 进行测试

数据库准备 Fixture

# tests/conftest.py
import pytest
from sqlmodel import SQLModel
from tests.test_db import engine

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture
async def create_test_tables():
    """
    在每个测试前创建表,测试后删除。
    每个测试都得到一个干净的数据库。
    """
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)

测试客户端 Fixture

# tests/features/test_examples_api.py
import pytest
from fastapi.testclient import TestClient
from src.app import app
from src.features.examples.api import get_example_service
from src.features.examples.service import ExampleService
from tests.test_db import TestingSessionLocal

async def override_get_example_service():
    """覆盖依赖以使用测试数据库"""
    async with TestingSessionLocal() as session:
        yield ExampleService(session=session)

app.dependency_overrides[get_example_service] = override_get_example_service

@pytest.fixture(scope="module")
def client():
    with TestClient(app) as c:
        yield c

真正的测试

import pytest
from fastapi.testclient import TestClient

@pytest.mark.anyio(backend='asyncio')
async def test_create_and_get_example(client: TestClient, create_test_tables):
    # 创建一个示例
    create_response = client.post(
        "/api/v1/examples/",
        json={"name": "Test Example", "description": "A test description"},
    )
    assert create_response.status_code == 200
    created_example = create_response.json()
    assert created_example["name"] == "Test Example"
    assert "id" in created_example

    # 再次获取
    example_id = created_example["id"]
    get_response = client.get(f"/api/v1/examples/{example_id}")
    assert get_response.status_code == 200
    fetched_example = get_response.json()
    assert fetched_example["id"] == example_id
    assert fetched_example["name"] == "Test Example"

可以看到,测试只声明了它需要的东西(clientcreate_test_tables),其余的交给 pytest 自动完成。数据库干净、测试客户端准备好,所有的搭建与拆卸都自动完成。

我们解决了什么(故障排查故事)

刚开始配置时,测试因为 trio 错误而失败。问题及解决办法如下:

  • 问题 1: pytest-anyio 试图使用 trio。
  • 解决方案: 添加返回 "asyncio"anyio_backend fixture,并移除不必要的 trio 依赖。

(如有其他问题和对应的修复,可自行补充。)

Back to Blog

相关文章

阅读更多 »