You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用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个。请问该问题成因是什么,如何解决?


解决方案

问题成因

  1. Anyio多后端自动运行:pytest.mark.anyio默认会同时使用asyncio和trio两个异步后端执行测试,导致每个测试用例被执行两次,这是测试数量翻倍的核心原因。
  2. 重复的自动Fixture:conftest.py和test_auth.py中都定义了同名的setup_and_teardown自动Fixture(autouse=True),虽不会直接导致测试重复收集,但会造成不必要的数据库连接/关闭操作,加重问题影响。
  3. 全局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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.22 15:17:32