PyTest-Asyncio: Collectionwasnotinitialisierte Fehler mit Fastapi und Beanie ODM im Test-Setup ⇐ Python

[Anonymous]

Ich erstelle eine Fastapi -Anwendung mit dem Repository -Muster mit Beanie ODM für MongoDB. Ich versuche, Tests mit PyTest-Asyncio einzurichten, aber ich stieß auf eine Beanie.exceptions.Collectionwasnotinitialized Fehler beim Ausführen meiner Tests.
Hier ist meine Projektstruktur : < /p>

Code: Select all

├── product_catalouge
│   ├── __init__.py
│   ├── core
│   │   ├── __init__.py
│   │   └── config.py
│   ├── exceptions
│   │   ├── database_exceptions.py
│   │   └── http_exceptions.py
│   ├── lib
│   │   ├── __init__.py
│   │   ├── db_initializer.py
│   │   ├── models_loader.py
│   │   └── utils.py
│   ├── models
│   │   ├── __init__.py
│   │   ├── attribute.py
│   │   ├── .
│   │   ├── .
│   │   └── .
│   ├── repository
│   │   ├── __init__.py
│   │   ├── attribute_repository.py
│   │   ├── base_repository.py
│   │   ├── .
│   │   ├── .
│   │   ├── .
│   ├── schemas
│   │   ├── __init__.py
│   │   ├── attribute.py
│   │   ├── .
│   │   ├── .
│   │   └── .
│   ├── service
│   │   ├── __init__.py
│   │   ├── attributes_service.py
│   │   ├── .
│   │   ├── .
│   │   ├── .
│   │   ├── base_service.py
│   │   ├── domain.py
│   │   └── unit_of_work.py
│   └── web
│       ├── __init__.py
│       ├── api
│       │   ├── __init__.py
│       │   ├── api.py
│       │   ├── routers
│       │   │   ├── __init__.py
│       │   │   ├── attribute.py
│       │   │   ├── .
│       │   │   ├── .
│       │   │   └── .
│       └── app.py
├── pyproject.toml
└── tests
├── __init__.py
├── conftest.py
└── test_attributes
├── conftest.py
└── test_attributes.py
< /code>
Meine Fastapi -App -Initialisierung (web/api/api.py

):

Code: Select all

# imports are omitted

@asynccontextmanager
async def lifespan(app: FastAPI):
await init_db(settings.db_uri, settings.db_name)
yield

app = FastAPI(title="Product Catalogue Service",
docs_url="/api/docs",
openapi_url="/api/openapi",
lifespan=lifespan)

app.include_router(attribute.router)
< /code>
Datenbankinitialisierung (Initialisierung (lib/db_initializer.py

):

Code: Select all

# imports are omitted

async def init_db(mongo_uri:str, db_name:str):
client = AsyncIOMotorClient(mongo_uri)
db = client[db_name]
await init_beanie(database=db,
document_models=get_beanie_models())
< /code>
Test -Setup in ((tests/conftest.py

):

Code: Select all

# some imports are omitted

from product_catalouge.core.config import settings
from product_catalouge.lib.models_loader import get_beanie_models
from product_catalouge.lib.db_initializer import init_db
from product_catalouge.web.api.routers import attribute

routers = [attribute]

@pytest_asyncio.fixture(autouse=True)
async def testdb():
db_client = AsyncIOMotorClient(settings.db_uri)
try:
await db_client.server_info()
print("[TEST DB CONNECTION] MongoDB connection successful.")
except Exception as e:
print(f"[TEST DB CONNECTION] MongoDB connection failed: {e}")
raise
await init_beanie(database=db_client.get_database(name=settings.test_db_name),
document_models=get_beanie_models())
print("[TEST DB INITIALIZATION] Beanie initialization complete.")
yield db_client
await db_client.drop_database(settings.test_db_name)
print("[TEST DB TEARDOWN] Test database dropped.")

@pytest_asyncio.fixture()
async def test_lifespan():
@asynccontextmanager
async def lifespan(app: FastAPI):
await init_db(settings.db_uri, settings.test_db_name)
yield
return lifespan

@pytest.fixture(autouse=True)
def test_app(test_lifespan):
app_ = FastAPI(
title="Product Catalogue Service (Test)",
docs_url="/api/docs",
openapi_url="/api/openapi",
lifespan=test_lifespan,
)
for router in routers:
app_.include_router(router.router)
return app_
< /code>
Testleuchten in ((tests/test_attributes/conftest.py

):

Code: Select all

# some imports are omitted
from product_catalouge.models.attribute import AttributeModel

@pytest.fixture()
def str_attr():
payload = {
"label": "String",
"internal_code": "String",
"type": "string",
"is_required": True,
"measurement_type":  "distance",
"unit": "cm",
"options": ["option1"]
}
return payload

@pytest.fixture()
def num_attr():
payload = {
"label": "Number",
"internal_code": "Number",
"type": "number",
"is_required": True,
"measurement_type": "distance",
"unit": "m",
"options": ["option1"]
}
return payload

@pytest.fixture()
def select_attr():
payload = {
"label": "Select",
"internal_code": "Select",
"type": "select",
"is_required": True,
"measurement_type": "distance",
"unit": "m",
"options": ["option1"]
}
return payload

@pytest.fixture(params=["str_attr", "select_attr"])
def str_select_payload(request):
return request.getfixturevalue(request.param)

@pytest.fixture()
def attrs(str_attr, num_attr, select_attr):
return [str_attr, num_attr, select_attr]

@pytest_asyncio.fixture()
async def three_attrs(testdb, attrs):
inserted_attrs = []
for attr in attrs:
inserted_attr = await AttributeModel(**attr).insert()
inserted_attrs.append(inserted_attr)
yield inserted_attrs
< /code>
Wenn ich diesen Test ausführe: < /p>
async def test_get_all_attrs(testdb, three_attrs, test_app):
async with AsyncClient(transport=ASGITransport(test_app),
base_url="http://test",
follow_redirects=True) as ac:
res = await ac.get("/api/attributes")
assert res.status_code == 200
assert len(res.json()) == 3
< /code>
Ich erhalte den Fehler: beanie.exceptions.collectionwasnotinitialized < /code> funktioniert sowohl gut als auch andere Tests. Ich richte die Testdatenbank ein oder verwalte den asynchronen Kontext, aber ich bin mir nicht sicher, was ich falsch mache. Beanie odm? Als ich beanie.exceptions.collectionwasnotinitialized 

Fehler bekam, war ich zwar aufgrund mangelnder Funktionalität in mongomock_motor. P>