visit
Tests are generally considered to be code that tests other code. Typically, tests are divided into two groups: unit tests and integration tests.
from fastapi import FastAPI, Body, HTTPException
from core.db import queries
from core import schemas
app = FastAPI()
@app.get(
"/items",
summary="Get items",
status_code=200,
response_model=list[schemas.ItemSchema],
)
def get_items() -> list[schemas.ItemSchema]:
items = queries.get_items()
return items
@app.post(
"/items",
summary="Add items",
status_code=200,
response_model=list[schemas.ItemSchema],
)
def add_items(
items: list[schemas.ItemBaseSchema] = Body(
...,
embed=True,
)
) -> list[schemas.ItemSchema]:
added_items = queries.add_items(
items=items
)
return added_items
@app.patch(
"/items/{item_id}",
summary="Update an item",
status_code=200,
response_model=schemas.ItemSchema,
)
def update_item(
item_id: int,
update_data: schemas.ItemBaseSchema = Body(
...,
embed=True,
)
) -> None:
if queries.get_item(
item_id=item_id
) is None:
raise HTTPException(status_code=404, detail="Item not found")
item = queries.update_item(
item_id=item_id,
update_data=update_data,
)
return item
@app.delete(
"/items/{item_id}",
summary="Delete an item",
status_code=204,
)
def delete_item(
item_id: int
) -> None:
if queries.get_item(
item_id=item_id
) is None:
raise HTTPException(status_code=404, detail="Item not found")
queries.delete_item(
item_id=item_id
)
import json
import pytest
from fastapi import status
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from core.db.models import Item
@pytest.fixture(scope='session')
def db_item(test_db_url, setup_db, setup_db_tables):
engine = create_engine(
test_db_url,
echo=False,
echo_pool=False,
)
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with session() as session:
item = Item(
name='name',
number=1,
is_valid=True,
)
session.add(
item
)
session.commit()
session.refresh(item)
return item.as_dict()
def test_get_items(
fastapi_test_client,
db_item,
):
response = fastapi_test_client.get(
'/items',
)
assert response.status_code == status.HTTP_200_OK
def test_post_items(
fastapi_test_client,
):
item_to_add = {
'name': 'name',
'number': 1,
'is_valid': False,
}
response = fastapi_test_client.post(
'/items',
data=json.dumps(
{
'items': [
item_to_add
],
},
default=str,
),
)
assert response.status_code == status.HTTP_200_OK
def test_update_item(
fastapi_test_client,
db_item,
):
update_data = {
'name': 'new_name',
'number': 2,
'is_valid': True,
}
response = fastapi_test_client.patch(
f'/items/{db_item["id"]}',
data=json.dumps(
{
'update_data': update_data,
},
default=str,
),
)
assert response.status_code == status.HTTP_200_OK
def test_delete_item(
fastapi_test_client,
db_item,
):
response = fastapi_test_client.delete(
f'/items/{db_item["id"]}',
)
assert response.status_code == status.HTTP_204_NO_CONTENT
It’s important to note that setup_db
, setup_db_tables
, and db_item
have a session
scope,
meaning these fixtures are destroyed only at the end of the test session—after all the tests have been completed.
For instance, if we add a new test to fetch data from the database and it runs after the DELETE
API test,
it could potentially fail because there would be no test data left in the database.
Even though there's a POST
method that runs before the DELETE
one, there's a risk that it might be moved or deleted, leading to test failures.
The second flaw is that both the test object in the db_item
fixture and the data added in the test_post_items
test are hardcoded. This approach works until there's a conflict in the database.
Currently, there are no constraints
, except for Item.id
(the primary key), set in the database.
The first key to making tests reliable is ensuring they are independent whenever possible.
In our example, it can achieved by changing the scope of the fixture setup_db_tables
from session
to function
. In result, the fixture will look like as:
@pytest.fixture(scope="function")
def setup_db_tables(setup_db, test_db_url):
create_db_engine = create_engine(test_db_url)
BaseModel.metadata.create_all(bind=create_db_engine)
yield
BaseModel.metadata.drop_all(bind=create_db_engine)
The Factory boy package is useful for generating unique data on the fly and offers the option to create objects directly in the database.
With a slight modification to its default Factory
class, we can create a custom factory that adds objects to the database:
import factory
from tests import conftest
class CustomSQLAlchemyModelFactory(factory.Factory):
class Meta:
abstract = True
@classmethod
def _create(cls, model_class, *args, **kwargs):
with conftest.db_test_session() as session:
session.expire_on_commit = False
obj = model_class(*args, **kwargs)
session.add(obj)
session.commit()
session.expunge_all()
return obj
class ItemModelFactory(CustomSQLAlchemyModelFactory):
class Meta:
model = models.Item
name = factory.Faker("word")
number = factory.Faker("pyint")
is_valid = factory.Faker("boolean")
import json
from unittest.mock import ANY
import pytest
from fastapi import status
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from core.db.models import Item
from tests import factories
def test_get_items(
fastapi_test_client
):
expected_items = factories.models_factory.ItemModelFactory.create_batch(
size=5
)
response = fastapi_test_client.get(
'/items',
)
assert response.status_code == status.HTTP_200_OK
response_data = response.json()
assert response_data == [
{
'id': item.id,
'name': item.name,
'number': item.number,
'is_valid': item.is_valid,
} for item in expected_items
]
def test_post_items(
fastapi_test_client,
test_db_session,
):
assert test_db_session.query(Item).first() is None
item_to_add = factories.schemas_factory.ItemBaseSchemaFactory.create()
response = fastapi_test_client.post(
'/items',
data=json.dumps(
{
'items': [
item_to_add.dict()
],
},
default=str,
),
)
assert response.status_code == status.HTTP_200_OK
response_data = response.json()
assert response_data == [
{
'id': ANY,
'name': item_to_add.name,
'number': item_to_add.number,
'is_valid': item_to_add.is_valid,
},
]
assert test_db_session.query(Item).filter(
Item.name == item_to_add.name,
Item.number == item_to_add.number,
Item.is_valid == item_to_add.is_valid
).first()
def test_update_item(
fastapi_test_client,
test_db_session,
):
item = factories.models_factory.ItemModelFactory.create()
update_data = factories.schemas_factory.ItemBaseSchemaFactory.create()
response = fastapi_test_client.patch(
f'/items/{item.id}',
data=json.dumps(
{
'update_data': update_data.dict(),
},
default=str,
),
)
assert response.status_code == status.HTTP_200_OK
response_data = response.json()
assert response_data == {
'id': ANY,
'name': update_data.name,
'number': update_data.number,
'is_valid': update_data.is_valid,
}
assert test_db_session.query(Item).filter(
Item.name == update_data.name,
Item.number == update_data.number,
Item.is_valid == update_data.is_valid
).first()
def test_delete_item(
fastapi_test_client,
test_db_session,
):
item = factories.models_factory.ItemModelFactory.create()
response = fastapi_test_client.delete(
f'/items/{item.id}',
)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert test_db_session.query(Item).first() is None
In this article, we explored common issues with testing practices in software development, particularly focusing on a FastAPI
application example.
We identified several flaws in existing tests, including dependency between tests, reliance on hardcoded values, and the lack of verification of method functionality. To address these issues, we discussed best practices such as using fixtures with function scope to ensure data isolation, utilizing tools like Factory Boy
to generate unique test data, and verify both the results and any internal changes made by the methods being tested.
I hope you found the article useful. You can clone the repository with the example provided ().