我正在使用 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
:
from app.database import get_database
from fastapi import APIRouter
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"))
return await users_collection.insert_one(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": "[email protected]",
"password": "password123",
}
response = await client.post("/auth/register", json=user_data)
功能良好,测试行为正确。但每当我输入
pytest
命令时,都会收集双倍的测试数量。例如,在我提供的示例设置中,pytest 将收集 2 个测试,并且它们都会执行(可以通过创建 2 个用户来验证)。这适用于多个文件中的所有测试。在我的完整设置中,我有 6 个测试,但它始终收集并执行 12 个测试。为什么会发生这种情况以及如何解决它。
我想这会解决你的问题:
@pytest.fixture
def anyio_backend():
return 'asyncio'
我从官方文档中获取它,
通过这种方式,AnyIO 仅使用 asyncio 进行测试。