使用pytest+anyio时为何会收集到重复测试用例?
背景信息
我正在使用pytest和asyncio为MongoDB + FastAPI后端编写测试。在搭建过程中遇到了诸多问题,例如Event Loop Close、Task Pending <name=以及AttributeError: 'async_generator' object has no attribute 'post',最终搭建出了可行的测试环境。以下是最小复现示例,因需完整上下文故篇幅较长,敬请谅解。
项目配置
main.py:
from fastapi import FastAPI from contextlib import asynccontextmanager from app.database import close_mongo_connection, connect_to_mongo from app.routes.auth_routes import router as auth_router app = FastAPI() @asynccontextmanager async def lifespan(app: FastAPI): print("Connecting to MongoDB") await connect_to_mongo() yield print("Closing connection to MongoDB") await close_mongo_connection() app.router.lifespan_context = lifespan app.include_router(auth_router, prefix="/auth", tags=["auth"])
database.py:
import os from motor.motor_asyncio import AsyncIOMotorClient from app.config import DATABASE_NAME, MONGO_DETAILS, TEST_DATABASE_NAME client: AsyncIOMotorClient = None db = None async def connect_to_mongo(): global client, db client = AsyncIOMotorClient(MONGO_DETAILS) if os.getenv("IS_TESTING") == "True": db = client[TEST_DATABASE_NAME] else: db = client[DATABASE_NAME] async def close_mongo_connection(): if os.getenv("IS_TESTING") == "True": await client.drop_database(TEST_DATABASE_NAME) client.close() print("Disconnected from MongoDB") async def get_database(): if db is None: await connect_to_mongo() return db
routes/auth_routes.py:
from app.database import get_database from fastapi import APIRouter, Depends from pydantic import BaseModel from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def get_password_hash(password): return pwd_context.hash(password) class UserCreate(BaseModel): email: str password: str class UserModel(BaseModel): email: str hashed_password: str router = APIRouter() async def create_user(user: UserCreate, db) -> UserModel: users_collection = db["users"] user_dict = user.model_dump() user_dict["hashed_password"] = get_password_hash(user_dict.pop("password")) result = await users_collection.insert_one(user_dict) return UserModel(**user_dict) @router.post("/register", response_model=UserModel) async def register_user(user: UserCreate, db=Depends(get_database)): print("GET DB", db) db_user = await create_user(user, db) return db_user
相关依赖库:
anyio 4.4.0 httpcore 1.0.5 httptools 0.6.1 httpx 0.27.0 motor 3.4.0 pip 24.0 pytest 8.2.2 pytest-asyncio 0.23.7 python-dotenv 1.0.1
测试配置
tests/conftest.py:
import pytest from app.database import close_mongo_connection, connect_to_mongo @pytest.fixture(scope="module", autouse=True) async def setup_and_teardown(): await connect_to_mongo() print("This function does not run, but the testing setup does not work without it") yield await close_mongo_connection()
tests/test_auth.py:
import pytest from httpx import AsyncClient, ASGITransport from app.main import app from app.database import connect_to_mongo, close_mongo_connection @pytest.fixture(scope="module", autouse=True) async def setup_and_teardown(): await connect_to_mongo() print("this runs") yield await close_mongo_connection() client = AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") @pytest.mark.anyio async def test_register_user(): user_data = { "email": "test@example.com", "password": "password123", } response = await client.post("/auth/register", json=user_data)
问题描述
功能与测试执行均正常,但每次执行pytest命令时,收集到的测试用例数量会翻倍。例如示例中仅1个测试用例,却会收集并执行2个(可通过创建了2个用户验证);实际项目中有6个测试用例,却会收集执行12个。请问该问题成因是什么,如何解决?
解决方案
问题成因
- Anyio多后端自动运行:
pytest.mark.anyio默认会同时使用asyncio和trio两个异步后端执行测试,导致每个测试用例被执行两次,这是测试数量翻倍的核心原因。 - 重复的自动Fixture:
conftest.py和test_auth.py中都定义了同名的setup_and_teardown自动Fixture(autouse=True),虽不会直接导致测试重复收集,但会造成不必要的数据库连接/关闭操作,加重问题影响。 - 全局AsyncClient:全局定义的
AsyncClient会在多后端环境中被重复初始化,进一步加剧重复执行的问题。
解决步骤
1. 指定Anyio仅使用Asyncio后端
在项目根目录创建或修改pytest.ini文件,添加以下配置:
[pytest] anyio_backends = asyncio
或在pyproject.toml中配置:
[tool.pytest.ini_options] anyio_backends = ["asyncio"]
此配置会强制Anyio只使用asyncio后端执行测试,直接解决测试翻倍的问题。
2. 移除重复的自动Fixture
保留conftest.py中的setup_and_teardownFixture(conftest中的Fixture对所有测试文件生效),删除test_auth.py中的同名Fixture,避免重复初始化数据库。
3. 将AsyncClient改为Fixture
把全局的client改为Fixture,使用async with上下文管理器确保客户端正确关闭,避免跨测试上下文的资源泄漏:
修改后的tests/test_auth.py:
import pytest from httpx import AsyncClient, ASGITransport from app.main import app @pytest.fixture(scope="module") async def client(): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as ac: yield ac @pytest.mark.anyio async def test_register_user(client): user_data = { "email": "test@example.com", "password": "password123", } response = await client.post("/auth/register", json=user_data) assert response.status_code == 200
内容的提问来源于stack exchange,提问作者Ahmet-Salman
相关产品推荐
相关产品推荐

