FastAPI & Pytest. Got Future attached to a different loop

The problem is this. I get an error when I try to make tests for my FastAPI application.

FAILED tests/test_users_api.py::test_create_jwt - RuntimeError: Task <Task pending name='Task-153' coro=<test_create_jwt() running at /backend/tests/test_users_api.py:22> cb=[_run_until_complete_cb() at /usr/local/lib/python3.12/asyncio/base_events.py:182]> got Future <Future pending cb=[BaseProtocol._on_waiter_completed()]> attached to a different loop
FAILED tests/test_users_api.py::test_create_user_with_valid_data - RuntimeError: Task <Task pending name='Task-179' coro=<test_create_user_with_valid_data() running at /backend/tests/test_users_api.py:122> cb=[_run_until_complete_cb() at /usr/local/lib/python3.12/asyncio/base_events.py:182]> got Future <Future pending cb=[BaseProtocol._on_waiter_completed()]> attached to a different loop

The error occurs when testing those functions where I work with the database via sqlalchemy.

Here is the code for my db sessions and dependency_override for FastAPI:

@pytest_asyncio.fixture(scope="session")
async def engine(app_database_url, migrations) -> AsyncEngine:
    engine = create_async_engine(app_database_url)
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def db_session(engine) -> AsyncGenerator[AsyncSession, None]:
    async with engine.connect() as connection:
        transaction = await connection.begin()
        session = AsyncSession(bind=connection, join_transaction_mode="create_savepoint", expire_on_commit=False)
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()
            await transaction.rollback()
            await connection.close()


@pytest.fixture
def session_override(db_session):

    async def get_session_override() -> AsyncGenerator[AsyncSession, None]:
        yield db_session

    main_app.dependency_overrides[db_helper.get_session] = get_session_override

My AsyncClient

@pytest_asyncio.fixture
async def client(session_override) -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient(app=main_app, base_url="http://localhost:8000") as ac:
        yield ac

I tried making my fixture for event_loop

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

But this did not change the situation

My pyproject.toml section with pytest settings

[tool.pytest.ini_options]
addopts = [
    "-vvv",
    "--cov=app",                   
    "--cov-report=term-missing", 
    "--cov-config=pyproject.toml"
]
python_files = "test_*.py"
filterwarnings = [
    "ignore::DeprecationWarning",
    "ignore::SyntaxWarning",
]
pythonpath = [
  ".", "backend",
]

asyncio_mode="auto"
asyncio_default_fixture_loop_scope = "session"

Project github https://github.com/DenisMaslennikov/to-do-list-FastAPI

Thanks to Boris's advice, I managed to find a bug in the tests. An async fixture with session scope was declared. Which created another event_loop and led to the problem.

Here is the code with the error:

@pytest_asyncio.fixture(scope="session")
async def engine(app_database_url, migrations) -> AsyncEngine:
"""Creates a SQLAlchemy engine to interact with the database."""

engine = create_async_engine(app_database_url)
yield engine
await engine.dispose()

To fix it, just need to remove scope="session" from the decorator.

And then it will become a foonction scope like all other fixtures and will not create its own separate event_loop

UPD No need to add decorator to all tests as Boris says. It just helped to find a bug in fixtures. Now I have function scope event_loop.