Testing
This document explains how to test the application.
Technologies
How to test a feature
In the current architecture, each feature is divided into three parts: a route, a service, and a repository.
A route is the endpoint of the feature defined using FastAPI. A service is a class that implements the business logic of the feature. A repository is a class that interacts with the database or any other external service, such as cache, message queue, or a third-party API.
In order to test all three parts of a feature, we need to create integrations tests that test the interaction between the route, service, and repository. Also we need to create tests for the service and repository separately.
Example
Let's test a feature called users that has a route, a service, and a
repository.
# app/routers/v1/routes.py
class DeleteUserResponse(pydantic.BaseModel):
message: str
@ROUTER.delete("/users/me", response_model=DeleteUserResponse)
async def delete_current_user(
current_user: Annotated[User, Depends(get_current_logged_user)],
user_service: Annotated[UserService, Depends(get_user_service)],
) -> DeleteUserResponse:
result = await user_service.delete(current_user.id)
if result.is_err():
error = result.err()
match error:
case UserNotFoundError():
msg = f"User with id {current_user.id} not found"
raise UserNotFoundException(msg)
case _:
raise InternalException(str(error))
return DeleteUserResponse(message="User deleted successfully")
# app/features/users/service.py
class UserService:
async def get_by_id(self, id: int) -> Result[User, UserError]:
res = await self._repository.get_by_id(id)
if res.is_err():
err = res.err()
assert err is not None
return Err(err)
user = res.unwrap()
if user is None:
return Err(UserNotFoundError())
return Ok(user)
async def delete(self, id: int) -> Result[None, UserError]:
user_result = await self.get_by_id(id)
if user_result.is_err():
err = user_result.err()
assert err is not None
return Err(err)
user = user_result.unwrap()
return await self._repository.delete(user.id)
# app/features/users/repository.py
class UserRepository:
async def delete(self, user_id: int) -> Result[None, UserError]:
try:
stmt = delete(User).where(User.id == user_id)
await self.session.execute(stmt)
await self.session.commit()
return Ok(None)
except Exception as e:
await self.session.rollback()
return Err(UserError(str(e)))
Now we can test everything:
# tests/features/users/test_users_integration.py
@pytest.mark.asyncio
async def test_delete_current_user_happy_path(
api_client: httpx.AsyncClient,
user_service: UserService,
mock_user: User,
) -> None:
login = await api_client.post(
"/api/v1/sign-in",
json={"cpf": mock_user.cpf, "password": DEFAULT_PASSWORD},
)
assert login.status_code == 200
token = login.json()["token"]
user = (await user_service.get_by_cpf("12345678901")).unwrap()
user_id = user.id
delete_response = await api_client.delete(
"/api/v1/users/me",
headers={"Authorization": f"Bearer {token}"},
)
assert delete_response.status_code == 200
assert (await user_service.get_by_id(user_id)).err_is(UserNotFoundError)
# tests/features/users/test_users_service.py
@pytest.mark.asyncio
async def test_delete_user_happy_path(
user_service: UserService,
mock_user: User,
) -> None:
delete_result = await user_service.delete(mock_user.id)
assert delete_result.is_ok()
get_result = await user_service.get_by_id(mock_user.id)
assert get_result.err_is(UserNotFoundError)
# tests/features/users/test_users_repository.py
@pytest.mark.asyncio
async def test_delete_user_happy_path(
session: AsyncSession,
user_repository: UserRepository,
mock_user: User,
) -> None:
user_id = mock_user.id
result = await user_repository.delete(user_id)
assert result.is_ok()
deleted_user = await session.get(User, user_id)
assert deleted_user is None
You see those parameters in the tests: api_client, user_service,
mock_user, session, user_repository.
They are all fixtures that are defined in conftest.py. The conftest.py file
is the entry point for all tests and it looks like something like this:
# tests/conftest.py
# ==== SERVICES ====
@pytest_asyncio.fixture
async def user_service(user_repository: UserRepository) -> UserService:
return UserService(user_repository)
@pytest_asyncio.fixture
async def real_estate_service(
real_estate_repository: RealEstateRepository,
) -> RealEstateService:
return RealEstateService(real_estate_repository)
# ==== REPOSITORIES ====
@pytest_asyncio.fixture
async def user_repository(session: AsyncSession) -> UserRepository:
return UserRepository(session)
@pytest_asyncio.fixture
async def real_estate_repository(
session: AsyncSession,
) -> RealEstateRepository:
return RealEstateRepository(session)
# ==== API ====
@pytest_asyncio.fixture()
async def api_client(session: AsyncSession) -> AsyncGenerator[
httpx.AsyncClient,
None,
]:
def get_session_override() -> AsyncSession:
return session
@asynccontextmanager
async def _override_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
yield
APP.router.lifespan_context = _override_lifespan
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=APP),
base_url="http://test",
) as client:
APP.dependency_overrides[get_session] = get_session_override
yield client
APP.dependency_overrides.clear()
# ==== DATABASE ====
@pytest.fixture(scope="session")
def postgres_container() -> Generator[PostgresContainer, None, None]:
with PostgresContainer("postgres:17", driver="asyncpg") as container:
yield container
@pytest_asyncio.fixture
async def engine(postgres_container: PostgresContainer) -> AsyncEngine:
return get_async_engine(postgres_container.get_connection_url())
@pytest_asyncio.fixture
async def sessionmaker(engine: AsyncEngine) -> AsyncSessionMaker:
return get_async_sessionmaker(engine)
@pytest_asyncio.fixture
async def session(
sessionmaker: AsyncSessionMaker,
) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker() as session:
async with session.bind.begin() as conn:
assert isinstance(conn, AsyncConnection)
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
yield session
# ==== OTHERS ====
@pytest_asyncio.fixture
async def signed_user_token(
api_client: httpx.AsyncClient,
mock_user: User,
user_service: UserService,
) -> str:
cpf = "12345678901"
name = "Test User"
res = await user_service.create(cpf, name, DEFAULT_PASSWORD)
assert res.is_ok()
response = await api_client.post(
"api/v1/sign-in",
json={
"cpf": mock_user.cpf,
"password": DEFAULT_PASSWORD,
},
)
assert response.status_code == 200
output = response.json()
assert "token" in output
token: str = output["token"]
return token
@pytest_asyncio.fixture
async def mock_user(user_service: UserService) -> User:
cpf = "12345678901"
name = "Test User"
res = await user_service.create(name, cpf, DEFAULT_PASSWORD)
assert res.is_ok()
res = await user_service.get_by_cpf(cpf)
assert res.is_ok()
res = res.unwrap()
assert res is not None
user: User = res
return user
Fixtures are really useful to avoid code duplication and to make our tests easier to read and maintain.