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 ofAsyncClient
- 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:
- 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.
3
Answers
Add session fixture in
conftest.py
. This fixture ensures that a new event loop is created for the session.You need to use decorator pytest_asyncio.fixture from pytest_asyncio
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.