很难使用 fastAPI 和 ODM beanie 运行 pytest，我收到错误：beanie.exceptions.CollectionWasNotInitialized

Question

每当我使用此代码执行 pytest 时，我总是遇到错误：FAILEDtests/test_main.py::test_create_todo - beanie.exceptions.CollectionWasNotInitialized。您对如何在运行测试之前初始化数据库有任何见解吗？

# tests/test_main.py
import pytest
from fastapi.testclient import TestClient
from src.core.database import init_db
from src.main import app

@pytest.fixture(autouse=True)
async def initialize_test_database():
    # Initialize test database before running any tests
    yield await init_db("mongodb://root:example@localhost:27017", "test_todoapp")  # Assuming you have a separate test database

@pytest.fixture
def client():
    return TestClient(app)

@pytest.mark.asyncio
async def test_create_todo(client):
    data = {"title": "Todo 1", "description": "Description Todo 1"}
    response = await client.post("/todos", json=data)
    assert response.status_code == 201

和我的数据库.py：

import os
from beanie import init_beanie
import motor.motor_asyncio

from src.models.todo import Todo


async def init_db(mongo_uri: str, database_name: str):
    client = motor.motor_asyncio.AsyncIOMotorClient(mongo_uri)
    await init_beanie(database=client[database_name], document_models=[Todo])


def get_mongo_uri():
    return os.environ.get("MONGO_URI", "mongodb://root:example@localhost:27017")

def get_mongo_db_name():
    return os.environ.get("MONGO_DB_NAME", "todoapp")

async def initialize_database():
    await init_db(get_mongo_uri(), get_mongo_db_name())

和我的 main.py:

from fastapi import FastAPI
from src.core.config import settings
from src.core.database import initialize_database
from src.routes.todo_routes import router as todo_router
import os

def init_router(app: FastAPI):
    app.include_router(todo_router, tags=["Todo"], prefix="/todos")

async def lifespan(app: FastAPI):
    await initialize_database()
    yield

def create_app() -> FastAPI:
    _app = FastAPI(
        title=settings.PROJECT_NAME,
        version=settings.PROJECT_VERSION,
        lifespan=lifespan
    )
    init_router(_app)
    return _app

app = create_app()

您对如何在运行测试之前初始化数据库有任何见解吗？

Answer 1

我找到了解决方案

def test_create_todo(client):
    with TestClient(app) as client:
        data = {"title": "Todo 1", "description": "Description Todo 1"}
        response = client.post("/todos", json=data)
        assert response.status_code == 201