Mostafa Dekmak

Posted on Nov 19

Writing Integration And Unit Tests for a Simple Fast API application using Pytest

#pytest #unittest #integrationtest #fastapi

Introduction

Python is a great language for building various types of applications, especially in today's landscape where machine learning and AI are rapidly advancing. With this growth in services, there’s a strong need for well-designed, maintainable, and scalable APIs. That’s where FastAPI comes in, a powerful async web API framework for Python that's both simple and robust https://fastapi.tiangolo.com/.

In this article, I won’t be covering the details of how FastAPI works or how to build APIs with it. Instead, the focus will be on writing integration and unit tests for FastAPI applications using Pytest. This guide is ideal for those familiar with Python and frameworks like Flask, Django, or other web frameworks (e.g., NestJS, Express, Spring Boot) who want to dive into building tests with Python and Pytest.

Github repo https://github.com/dkmostafa/fast-api-sample/tree/unit-testing-sample

First thing first lets Install the required requirements and that will be used for the testing purpose :

pytest : The main testing framework for creating integration and unit test in python
pytest-asyncio : async io pytest to test asynchronous calls ( ex : API calls )
pytest-env : imitating environment variables for testing puproses ( optional )
pytest-cov : library to produce the coverage test of our app ( optional if intrested in coverage test )
Faker : a powerful and easy seeding library to seed our database

So lets start by writing our first integration test In Part 1 and Part 2 we will write a Unit Test :

Part 1

I will create and test a simple API endpoint : /user that fetches me all the users in my database.

Before start with writing our unit test that tests that endpoints , we have to make sure that we isolate our testing database from the production one , WE DONT WANT TO PERFORM ANY OPERATIONS ON OUR MAIN DATABASE !!!

Database Preparation :

Creating the database and models :

@pytest.fixture()
def session():

    Base.metadata.create_all(engine)
    try:
        with get_session() as session:
            yield session
    finally:
        Base.metadata.drop_all(engine)

in this fixture we are creating a separate database session for our tests , that's create before running any tests ( by passing the session as an argument ) and dropping it all after our designated test are done .

Seeding our database :

Since isolation is an important part of our testing , again *WE DONT WANT TO USE ANY PRODUCTION DATABASE * so we are going to seed our database with fake data , hence using the Faker library :

@pytest.fixture()
def seeded_users_session(session):
    faker = Faker()

    user_repo = UserRepository()
    for i in range(1, 6):
        seeded_user = {
            "username": faker.user_name(),
            "name": faker.name()
        }
        user_repo.save(seeded_user)

NOTE ! : how I passed the session fixture to the seeded_users_session fixture function.

So now since our database preparation is done , lets dive into the test finally.

First Step :

Defining my test , and what am expecting from my test , a very popular approach when writing tests is following a TDD approach where we write our tests before writing the implementation , From here I want to create an api that do the following :

Fetches me list of users and the API enpoint is /user
make sure that my status code response is 200

our test will be as simple as translating the above text into a test code :

def test_get_users(seeded_users_session):
    response = user_client.get("/user") 
    json_response = response.json()
    assert response.status_code == 200
    assert len(json_response) > 1
    assert isinstance(json_response, list)

In short : passing the seeded_users_sessionfixture to this function so it used the above seeded database.

Calling the API
Parsing the json response
Asserting the status code
asserting that the list is greater than 1.

Full Integration Test File :

import pytest
from faker import Faker
from fastapi.testclient import TestClient
from database.db_engine import Base, get_session, engine
from database.models.user_model import UserModel
from database.repositories.user_repository import UserRepository
from ...models.user_response_models import CreateUserResponse

from ...user_routes import user_router


@pytest.fixture()
def session():

    Base.metadata.create_all(engine)
    try:
        with get_session() as session:
            yield session
    finally:
        Base.metadata.drop_all(engine)


@pytest.fixture()
def seeded_users_session(session):
    faker = Faker()

    user_repo = UserRepository()
    for i in range(1, 6):
        seeded_user = {
            "username": faker.user_name(),
            "name": faker.name()
        }
        user_repo.save(seeded_user)


user_client = TestClient(user_router)


def test_create_user_success(session):
    response = user_client.post("/user", json={
        "name": "test_user_creation",
        "username": "test_user_creation@gmail.com",
    })

    json_response = response.json()
    CreateUserResponse(**json_response)
    assert response.status_code == 200


def test_get_users(seeded_users_session):
    response = user_client.get("/user")
    json_response = response.json()
    assert response.status_code == 200
    assert len(json_response) > 1
    assert isinstance(json_response, list)


def test_get_user_by_id(seeded_users_session):
    response = user_client.get("/user/1")
    json_response = response.json()
    UserModel(**json_response)
    assert response.status_code == 200

Second Step :

Create the Api :

@user_router.get("/")
def get_users():
    res = user_service.get_users()
    return res

user_router is an instance of the APIRouter to organise my FastApi routes.
user_service.get_users() is a simple user fetch query written in sqlalchemy . ( check the repo for more details)

add your logic inside the get_users in iterations till your test is Passed.

Next step will be writing a simple Unit Test :

Part 2 :

Unit test differ from integration , in the integration we are testing the full API journey , while in the unit test we are testing a single very scoped function while we isolate it from external influence and focuses on testing our business inside our services, ( database , or others ) for this purpose we will use Mock .

so lets start by writing our test :
in my test I will test the get_user(self, user_id: int) in 2 scenarios , first scenario if it returns me an existing user and assert that this user in of an instance of the UserModel , second scenario is if the user not found and it throws me an error.

First things first is by preparing my service to the test :

@pytest.fixture
def user_service():
    mock_user_repository = MagicMock()
    mock_user_repository.get_by_id.return_value = UserModel(
        name=faker.name(),
        username=faker.user_name()
    )

    user_service = UserService()
    user_service.user_repository = mock_user_repository

    return user_service

as we did in the integration test , I created a fixture where i mocked the user_repo so it wont call a database call instead it will mock the data am assuming it will return

def test_get_user_success(user_service):
    response = user_service.get_user(1)
    assert isinstance(response, UserModel)

Here am calling my function and asserting its output .

Second scenario if the user is not found . a diffrent mock is given to the user_service , where am assuming no user will be found and asserting that my error is raised.

def test_get_user_not_found():
    mock_user_repository = MagicMock()
    mock_user_repository.get_by_id.return_value = None
    user_service = UserService()
    user_service.user_repository = mock_user_repository

    with pytest.raises(NotFoundException) as excinfo:
        user_service.get_user(1)
    assert str(excinfo.value) == 'User does not exist'

Full Unit test file here :

import pytest
from faker import Faker
from unittest.mock import MagicMock

from Exceptions.not_found_exception import NotFoundException
from database.models.user_model import UserModel
from modules.User.services.user_service import UserService

faker = Faker()


@pytest.fixture
def user_service():
    mock_user_repository = MagicMock()
    mock_user_repository.get_by_id.return_value = UserModel(
        name=faker.name(),
        username=faker.user_name()
    )

    user_service = UserService()
    user_service.user_repository = mock_user_repository

    return user_service


def test_get_user_success(user_service):
    response = user_service.get_user(1)
    assert isinstance(response, UserModel)


def test_get_user_not_found():
    mock_user_repository = MagicMock()
    mock_user_repository.get_by_id.return_value = None
    user_service = UserService()
    user_service.user_repository = mock_user_repository

    with pytest.raises(NotFoundException) as excinfo:
        user_service.get_user(1)
    assert str(excinfo.value) == 'User does not exist'

Additional Part :

How to make sure that have a good coverage of my code , for this i will run :
pytest --cov that gives my a coverage report for my code

Conclusion

Writing unit and integration test is essential as writing the code it self . ignoring tests means that am ignoring the quality of the code am writing . The repo contains more unit and integration tests you can check.
Have a nice testing :D

DEV Community