import asyncio
import pytest
import pytest_asyncio
from motor.motor_asyncio import AsyncIOMotorClient
from odmantic import AIOEngine
import os
from typing import AsyncGenerator
from app.main import app
from app.db.database import get_database
from fastapi.testclient import TestClient
# Test database configuration
TEST_DATABASE_URL = os.getenv("TEST_DATABASE_URL", "mongodb://localhost:27017")
TEST_DATABASE_NAME = "test_db"
@pytest_asyncio.fixture(scope="session")
async def event_loop():
"""Create an instance of the default event loop for the test session."""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
yield loop
loop.close()
@pytest_asyncio.fixture(scope="function")
async def test_engine() -> AsyncGenerator[AIOEngine, None]:
"""Create a test database engine and clean up after each test."""
client = AsyncIOMotorClient(TEST_DATABASE_URL)
engine = AIOEngine(client=client, database=TEST_DATABASE_NAME)
yield engine
try:
print(f"🗑️ Dropping database: {TEST_DATABASE_NAME}")
await client.drop_database(TEST_DATABASE_NAME) # <-- Fails here if event loop is closed
except Exception as e:
print(f"Error dropping database: {e}")
finally:
client.close()
@pytest_asyncio.fixture
async def client(test_engine):
"""Create a test client with dependency override."""
app.dependency_overrides.clear()
app.dependency_overrides[get_database] = lambda: test_engine
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
@pytest.fixture
def auth_headers(client):
"""Fixture to get authenticated headers for testing protected routes."""
# Register test user
client.post("/api/v1/users", json={"email": "[email protected]", "password": "test123", "is_superuser": True})
# Login and return token
login_response = client.post("/api/v1/login", json={"email": "[email protected]", "password": "test123"})
if login_response.status_code == 200:
return {"Authorization": f"Bearer {login_response.json()['access_token']}"}
return None
I'm trying to set up test fixtures for my FastAPI application using MongoDB (via motor) and ODMantic (AIOEngine). I'm running into an issue during fixture teardown when trying to drop the test database:
Error dropping database: This event loop is already closed