显式优于隐式:掌握 Pytest fixture 与异步测试
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 测试异步代码时,它需要知道使用哪个事件循环。默认情况下,它会尝试“帮助”你,针对 所有可用的后端 进行测试。
实际发生的情况:
- 你使用
asyncio编写测试(因为 FastAPI 使用它)。 - 你安装了
pytest-anyio来测试异步代码。 pytest-anyio发现你系统里装有trio(可能是别的项目的依赖)。- 它尝试用 trio 运行你的测试。
- 你的
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"
可以看到,测试只声明了它需要的东西(client、create_test_tables),其余的交给 pytest 自动完成。数据库干净、测试客户端准备好,所有的搭建与拆卸都自动完成。
我们解决了什么(故障排查故事)
刚开始配置时,测试因为 trio 错误而失败。问题及解决办法如下:
- 问题 1:
pytest-anyio试图使用 trio。 - 解决方案: 添加返回
"asyncio"的anyio_backendfixture,并移除不必要的trio依赖。
(如有其他问题和对应的修复,可自行补充。)