基于FastAPI与MongoDB的异步API接口Pytest测试问题排查

1. 重构数据库连接逻辑 #

原代码在模块加载阶段就初始化了MongoDB连接，导致测试时无法动态切换到测试数据库。改为依赖注入+FastAPI生命周期管理的方式，实现数据库连接的动态初始化与切换：

修改database.py：

from motor.motor_asyncio import AsyncIOMotorClient
from starlette.config import Config

config = Config()
MONGO_DETAILS = config("MONGO_DETAILS", cast=str)
DATABASE_NAME = config("DATABASE_NAME", cast=str, default="app_db")

client: AsyncIOMotorClient | None = None
db = None

async def get_database():
    global db, client
    if db is None:
        client = AsyncIOMotorClient(MONGO_DETAILS)
        try:
            await client.admin.command("ismaster")
            print("Connected to MongoDB")
        except Exception as e:
            print(f"Server not available: {e}")
            raise
        db = client[DATABASE_NAME]
    return db

async def close_db_connection():
    global client
    if client is not None:
        client.close()
        print("MongoDB connection closed")

在main.py中注册生命周期事件：

from fastapi import FastAPI
from app.database import close_db_connection
from app.auth_route import router as auth_router

app = FastAPI()

app.include_router(auth_router, prefix="/auth")

@app.on_event("shutdown")
async def shutdown_event():
    await close_db_connection()

修改auth_route.py，通过依赖注入获取数据库集合：

from pydantic import BaseModel
from passlib.context import CryptContext
from fastapi import APIRouter, Depends
from app.database import get_database

router = APIRouter()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class UserBase(BaseModel):
    name: str

class UserCreate(UserBase):
    password: str

async def get_users_collection(db=Depends(get_database)):
    return db["users"]

def get_password_hash(password: str):
    return pwd_context.hash(password)

async def create_user(user: UserCreate, users_collection):
    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 await users_collection.find_one({"_id": result.inserted_id})

@router.post("/register")
async def register_user(user: UserCreate, users_collection=Depends(get_users_collection)):
    db_user = await create_user(user, users_collection)
    return db_user

2. 修复测试Fixtures与测试用例 #

解决环境变量生效晚、测试数据库隔离、测试用例逻辑错误的问题：

修改conftest.py：

import pytest
from httpx import AsyncClient
from starlette.config import environ
from app.main import app
from app.database import get_database, close_db_connection, db as global_db, client as global_client

@pytest.fixture(scope="function")
async def test_db():
    # 切换到测试数据库
    original_db = environ.get("DATABASE_NAME")
    environ["DATABASE_NAME"] = "test_db"
    
    # 重置全局数据库连接
    global global_db, global_client
    global_db = None
    global_client = None
    
    test_db = await get_database()
    yield test_db
    
    # 测试结束后删除测试库并恢复配置
    await test_db.client.drop_database("test_db")
    if original_db:
        environ["DATABASE_NAME"] = original_db
    await close_db_connection()

@pytest.fixture(scope="function")
async def test_client(test_db):
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

修改test_auth.py（修正原测试中不存在的email断言）：

import pytest

@pytest.mark.asyncio
async def test_register_user(test_client):
    user_data = {
        "name": "Test User",
        "password": "testpassword",
    }
    response = await test_client.post("/auth/register", json=user_data)
    
    assert response.status_code == 200, f"请求失败：{response.text}"
    
    response_data = response.json()
    assert response_data["name"] == user_data["name"]
    assert "hashed_password" in response_data
    assert "_id" in response_data

3. 关键问题说明 #

• 模块级连接初始化问题：原代码在模块加载时就创建了数据库连接，此时测试环境变量尚未生效，导致测试使用生产数据库。重构后通过依赖注入动态初始化连接，确保测试时能切换到独立的测试库。
• 事件循环错误：使用pytest-asyncio的默认事件循环，确保所有异步操作（HTTP请求、数据库操作）在同一个循环中执行，避免Event Loop Closed或任务挂起错误。
• 测试数据隔离：每个测试函数创建独立的测试数据库，测试完成后自动删除，彻底避免测试数据污染。

运行测试注意事项 #

确保安装依赖：pytest, pytest-asyncio, httpx, motor, fastapi, passlib[bcrypt]
执行测试命令：pytest tests/ -v

内容的提问来源于stack exchange，提问作者Ahmet-Salman