How to implement pytest for FastAPI with MongoDB(Motor)
Asked Answered
S

2

5

I want to write tests for my FastAPI endpoints

example for my code:

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/todos")
async def get_todo_by_title(title: str,current_user: User = Depends(get_current_user))
    document = await collection.find_one({"title": title})
    return document

client = TestClient(app)

def test_get_todo_by_title():
    response = client.get("/todos")
    assert response.status_code == 200

What the best way to test my endpoints?

I want to use fake DB for testing, something like json file

db = {
todos: [...]
}
Selfpreservation answered 26/4, 2022 at 7:2 Comment(5)
The best way would be to just test the endpoints with a test database in mongodb, so that you're sure they work as you expect. The next possibility would be to either mock your collection queries or create a fake that allows certain operations following the API of mongodb collections. A better solution would be to move your actual mongdb queries to a dedicated service that you inject into your views with Depends, then mock this service to give back a set of data as defined in a json file. But first: is all this indirection really necessary? Start by running your tests with the real mongodbCumuliform
so if I run my tests on real MongoDB, you know how I ca to skip depends? use fake token for test?Selfpreservation
Are you thinking about the user dependency? Or something else?Cumuliform
about the user dependencySelfpreservation
You can either make your application allow adding users and authenticate as the user (i.e. the real way), or you can use app.dependency_overrides to provide a custom function that returns a fake/static user in your tests. fastapi.tiangolo.com/advanced/testing-dependencies - app.dependency_overrides[get_current_user] = lambda: return {'id': 1, 'username': 'foo'}Cumuliform
D
4

It is not the best option to use fake data from a JSON file. Instead, you can use a testing DB (based on the env you are running your app through), or any other DB (dev, stg, ..etc), and delete your test data after running all unit tests.

Here is how to simply apply the latter approach in FastAPI;

  • Assume you have 3 simple tables (or collections) X, Y, and Z
  • MongoDB is the DB service
  • PyTest is the test engine

conftest.py

from pytest import fixture
from starlette.config import environ
from starlette.testclient import TestClient
from config.db_connection import database, X_collection, Y_collection, Z_collection


@fixture(scope="session")
def test_X():
    return {
        "_id": "10",
        "name": "test",
        "description": "test",
        "type": "single value",
        "crt_at": "2022-06-27T12:23:15.143Z",
        "upd_at": "2022-06-27T12:23:15.143Z"
    }

//test_Y and test_Z fixtures should be like test_X


@fixture(scope="session", autouse=True)
def test_client(test_X, test_Y, test_Z):
    import main
    application = main.app
    with TestClient(application) as test_client:
        yield test_client

    db = database
    //Here, delete any objects you have created for your tests
    db[X_collection].delete_one({"_id": test_X["_id"]})
    db[Y_collection].delete_one({"_id": test_Y["_id"]})
    db[Z_collection].delete_one({"_id": test_Z["_id"]})


environ['TESTING'] = 'TRUE'

Unit tests should look like the following sample.

test_X_endpoints.py

def test_create_X(test_client, test_X):
    response = test_client.post("/create_X_URI/", json=test_X)
    assert response.status_code == 201
    //Add assertions as needed

def test_list_X(test_client):
    response = test_client.get("/list_X_objects_URI/?page=1&size=1")
    assert response.status_code == 200
    //Add assertions as needed
Dreddy answered 14/7, 2022 at 16:17 Comment(0)
S
2

Using a fake database and tearing it down at the end of your tests will definitely work, but I find it is usually better to implement transactional testing when testing routes that rely on databases. The process looks something like this:

  • In the session scope, create a PyTest fixture that returns a Motor client for all our tests to use.
  • In the function scope, create a Pytest fixture that yields a Mongo session and transaction for each test to use. This transaction should rollback when the test ends to maintain a pure state of the underlying database.
  • Use FastAPI overrides so that our application uses our custom Motor client and our custom session for the tests.

This approach has some benefits, namely that we can manipulate data in each individual test case without affecting other tests. It is also faster than actually writing and deleting data from the database since all the data manipulation is done in a transaction that is never committed.

To get this all working, the trick is to make sure that FastAPI and Motor are using the same event loop. Otherwise, you may run into issues described in this note in the FastAPI docs. Also, make sure to install a PyTest Async extension like pytest-asyncio and either mark your tests and fixtures appropriately or use --asyncio-mode=auto when calling PyTest or in your pytest.ini.

Here is a basic example of an implementation that works for me:

The Motor test client

import pytest
import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

@pytest.fixture(scope="session")
def mongodb_client():
    client = AsyncIOMotorClient(
        ...your connection settings
    )
    # make sure the client always uses the current event loop instead of creating a new one
    # this will ensure the application and the test are running in the same loop
    client.get_io_loop = asyncio.get_event_loop 
    yield client
    client.close()

The Mongo test session

import pytest
from motor.core import AgnosticClient

@pytest.fixture
async def mongo_session(mongodb_client: AgnosticClient):
    async with await mongodb_client.start_session() as session: # wrap each test in its own session
        session.start_transaction() # wrap each test in its own transaction
        yield session
        await session.abort_transaction() # abort at the end of the test so nothing is written to our database

Now that we have these fixtures, we can override dependencies in our app. This part will vary depending on how you structured your own FastAPI application. My application has dependencies that return my client and session so I can just override them. I find this works well, since then each of my routes in my application can use the client and session like so:

@router.get("/{some_id}")
async def get_some_thing(some_id: str, mongo_client: Annotated[AgnosticClient, Depends(get_mongo_client], mongo_session: Annotated[AgnosticClientSession, Depends(get_mongo_session)]):
    collection = mongo_client["some database"].get_collection("some collection")
    return await collection.find_one({"some_id": some_id}, session=mongo_session)

Using this approach to my application, the TestClient fixture can just override the relevant dependencies:

import pytest
from motor.core import AgnosticClient, AgnosticClientSession
from fastapi.testclient import TestClient

@pytest.fixture
def client(
    mongodb_client: AgnosticClient,
    mongo_session: AgnosticClientSession,
):
    def override_get_mongo_client():
        return mongodb_client

    def override_get_mongo_db_session():
        return mongo_session


    app.dependency_overrides[get_mongo_client] = override_get_mongo_client
    app.dependency_overrides[get_mongo_db_session] = override_get_mongo_db_session

    return TestClient(app)

With all this in place, we can use the client and session in our tests and know that the FastAPI application is also using the same client and session. This allows us to insert fake documents in each test, and then use those fake documents when accessing our CRUD routes, etc.

Straightaway answered 28/6 at 15:59 Comment(0)

© 2022 - 2024 — McMap. All rights reserved.