1
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

1 Answer 1

2

pytest-asyncio manages the event loop. You don't have to create a new event loop for your tests. So, your event_loop fixture is not required. To set up an event loop with the session scope, set asyncio_default_fixture_loop_scope and asyncio_default_test_loop_scope to session. See How to change the default event loop scope of all fixtures and How to change the default event loop scope of all tests.

Note: When the configuration option asyncio_default_fixture_loop_scope is unset, the event loop scope for asynchronous fixtures defaults to the scope of the fixture. But the future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function.

When asyncio_default_test_loop_scope configuration option is unset, it already defaults to function scope.

You can also override default loop scope of your fixture by decorating it with:

@pytest_asyncio.fixture(loop_scope="module")

See, How to change the event loop scope of a fixture.

Additionally, set asyncio_mode to auto. auto mode automatically adds asyncio marker to async test functions.

Sign up to request clarification or add additional context in comments.

Comments

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.