0

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.

2 Answers 2

2

You need to use decorator pytest_asyncio.fixture from pytest_asyncio.

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

2 Comments

Could you explain why its needed?
This is needed on all async fixtures - instead of standard pytest.fixture.
0

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

1 Comment

I'm getting the error AttributeError: 'async_generator' object has no attribute 'post' now. I have gotten all sorts of errors such as event loop closed, Pending Task <name=...., and the one I just mentioned

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.