skip to Main Content

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 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

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"]

I have tried:

  • replacing @pytest.mark.asyncio with @pytest.mark.anyio
  • changing the fixture 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.

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.

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

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

3

Answers


  1. Add session fixture in conftest.py. This fixture ensures that a new event loop is created for the session.

    @pytest.fixture(scope="session")
    def event_loop():
        loop = get_event_loop_policy().new_event_loop()
        yield loop
        loop.close()
    
    Login or Signup to reply.
  2. You need to use decorator pytest_asyncio.fixture from pytest_asyncio

    Login or Signup to reply.
  3. As Badmajor said, you should use pytest_asyncio that will provide you a decorator to create async fixtures.

    You won’t need to use the event_loop fixture provided by Dmitri Galkin since this issue has been fixed.

    Login or Signup to reply.
Please signup or login to give your own answer.
Back To Top
Search