Context

Hello everyone, I have checked the following links for a potential solution for my problem:

• How to implement pytest for FastAPI with MongoDB(Motor)
• What causes error ‘async_generator’ object has no attribute ‘add’?
• AttributeError in pytest with asyncio after include code in fixtures
• pytest-asyncio has a closed event loop, but only when running all tests

And none of them are working, unless I missed something in their solution.

Problem

So my issue is the following, I have an API using FastAPI as the backend and MongoDB as the database. I want to write tests for the API endpoints I have, I have been trying for hours to get a single test up and running but with no luck. My minimum reproducible setup is the following:

database.py

client = AsyncIOMotorClient(MONGO_DETAILS)
db = client[DATABASE_NAME]
try:
    # The ismaster command is cheap and does not require auth.
    client.admin.command("ismaster") # type: ignore
    print("Connected to MongoDB")
except Exception as e:
    print("Server not available")

print("Database:", db)


def close_mongo_connection():
    client.close()

auth_route.py:

class UserCreate(UserBase):
    name: str
    password: str

async def create_user(user: UserCreate):
    user_dict = user.model_dump()
    user_dict["hashed_password"] = get_password_hash(user_dict.pop("password"))
    result = await users_collection.insert_one(user_dict)
    return result

@router.post("/register")
async def register_user(user: UserCreate):
    db_user = await create_user(user)
    return db_user

Test Setup

My tests are inside a folder called tests and it has 2 files:

conftest.py:

from httpx import AsyncClient
from pytest import fixture
from starlette.config import environ
import app.main as main

@fixture(scope="function", autouse=True)
async def test_client(monkeypatch):
    original_env = environ.get('ENVIRONMENT')
    monkeypatch.setenv("ENVIRONMENT", "test")
    application = main.app
    async with AsyncClient(app=application, base_url="http://test") as test_client:
        yield test_client
    monkeypatch.setenv("ENVIRONMENT", original_env)

test_auth.py

import pytest
@pytest.mark.asyncio
async def test_register_user(test_client):
    user_data = {
        "name": "Test User",
        "password": "testpassword",
    }
    response = await test_client.post("/auth/register", json=user_data)
    print(response)
    assert response.status_code == 200
    assert response.json()["email"] == user_data["email"]

What I have tried

So I have tried:

• replacing @pytest.mark.asyncio with @pytest.mark.anyio
• changing the picture scope from function to session
• using TestClient instead of AsyncClient
• setting asyncio_mode=auto when executing the tests

The errors I’m getting range from RuntimeError: Task <Task pending name='Task-3' coro=<test_register_user() to Event Loop Closed. Whenever I change anything, I get another error.

Expected behavior

• I want the tests to execute without errors, it should send a request to my API and receive back an answer from the live database
• I want the tests to set up a new DB (e.g. called “test”) and apply all the CRUD operations in that DB, then delete the DB after the tests are done