[Python] How to Drop MongoDB Test Database After Async Fixture Teardown Without Event Loop...

Discussão em 'Python' iniciado por Stack, Junho 8, 2025.

  1. Stack

    Stack Membro Participativo

    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": "test@gmail.com", "password": "test123", "is_superuser": True})
    # Login and return token
    login_response = client.post("/api/v1/login", json={"email": "test@gmail.com", "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

    Continue reading...

Compartilhe esta Página