Testing Microservices in Practice with FastAPI and PostgreSQL

Testing microservices presents unique challenges compared to monolithic applications. With multiple services communicating over networks, databases, message queues, and external APIs, a comprehensive testing strategy becomes critical. This guide covers practical approaches to testing FastAPI microservices with PostgreSQL.

Table of Contents

Testing Philosophy for Microservices

Test Environment Setup

Unit Testing

Integration Testing

Contract Testing

End-to-End Testing

Database Testing Strategies

Testing in CI/CD

Common Pitfalls and Solutions

Testing Philosophy for Microservices {#testing-philosophy}

The testing pyramid for microservices differs from traditional applications:

        /\
       /E2E\        ← Few (5-10%)
      /------\
     /Contract\     ← Some (15-20%)
    /----------\
   /Integration\    ← More (25-30%)
  /--------------\
 /  Unit Tests   \  ← Most (40-50%)
/------------------\

Key Principles:

Test independently: Each service should be testable in isolation

Mock external dependencies: Use test doubles for other services

Database per test: Ensure test isolation with separate DB instances

Fast feedback loops: Unit tests should run in milliseconds

Test the contract: Verify API contracts between services

Test Environment Setup {#test-environment-setup}

Project Structure

my-microservice/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models/
│   ├── services/
│   ├── repositories/
│   └── api/
├── tests/
│   ├── __init__.py
│   ├── conftest.py          # Shared fixtures
│   ├── unit/
│   │   ├── test_services.py
│   │   └── test_repositories.py
│   ├── integration/
│   │   ├── test_api.py
│   │   └── test_database.py
│   ├── contract/
│   │   └── test_user_service_contract.py
│   └── e2e/
│       └── test_user_flow.py
├── requirements.txt
├── requirements-dev.txt
├── docker-compose.test.yml
└── pytest.ini

Dependencies Installation

requirements-dev.txt:

# Testing
pytest==7.4.3
pytest-asyncio==0.21.1
pytest-cov==4.1.0
pytest-mock==3.12.0
httpx==0.25.2
faker==20.1.0

# Test Database
pytest-postgresql==5.0.0
sqlalchemy-utils==0.41.1

# Contract Testing
pact-python==2.0.1

# Mocking
responses==0.24.1
freezegun==1.4.0

Pytest Configuration

pytest.ini:

[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
asyncio_mode = auto
addopts = 
    -v
    --strict-markers
    --cov=app
    --cov-report=term-missing
    --cov-report=html
    --cov-fail-under=80
markers =
    unit: Unit tests
    integration: Integration tests
    contract: Contract tests
    e2e: End-to-end tests
    slow: Slow running tests

Test Configuration

tests/conftest.py:

import asyncio
import pytest
from typing import AsyncGenerator, Generator
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool
from httpx import AsyncClient
from app.main import app
from app.database import Base, get_db
from app.config import get_settings

# Test database URL
TEST_DATABASE_URL = "postgresql+asyncpg://test_user:test_pass@localhost:5433/test_db"

@pytest.fixture(scope="session")
def event_loop() -> Generator:
    """Create event loop for async tests."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def test_engine():
    """Create test database engine."""
    engine = create_async_engine(
        TEST_DATABASE_URL,
        poolclass=NullPool,
        echo=False,
    )
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    yield engine
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    
    await engine.dispose()

@pytest.fixture
async def db_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
    """Create a fresh database session for each test."""
    async_session = async_sessionmaker(
        test_engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )
    
    async with async_session() as session:
        async with session.begin():
            yield session
            await session.rollback()

@pytest.fixture
async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    """Create test client with overridden dependencies."""
    async def override_get_db():
        yield db_session
    
    app.dependency_overrides[get_db] = override_get_db
    
    async with AsyncClient(app=app, base_url="http://test") as test_client:
        yield test_client
    
    app.dependency_overrides.clear()

@pytest.fixture
def mock_settings():
    """Mock application settings."""
    settings = get_settings()
    settings.database_url = TEST_DATABASE_URL
    settings.environment = "test"
    return settings

Unit Testing {#unit-testing}

Unit tests focus on individual functions and classes in isolation.

Testing Services

app/services/user_service.py:

from typing import Optional
from app.models.user import User
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate, UserUpdate
from app.utils.password import hash_password, verify_password
from app.exceptions import UserAlreadyExistsError, InvalidCredentialsError

class UserService:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    async def create_user(self, user_data: UserCreate) -> User:
        """Create a new user."""
        # Check if user exists
        existing_user = await self.user_repository.get_by_email(user_data.email)
        if existing_user:
            raise UserAlreadyExistsError(f"User with email {user_data.email} already exists")
        
        # Hash password
        hashed_password = hash_password(user_data.password)
        
        # Create user
        user = User(
            email=user_data.email,
            username=user_data.username,
            hashed_password=hashed_password,
        )
        
        return await self.user_repository.create(user)
    
    async def authenticate(self, email: str, password: str) -> Optional[User]:
        """Authenticate user by email and password."""
        user = await self.user_repository.get_by_email(email)
        
        if not user or not verify_password(password, user.hashed_password):
            raise InvalidCredentialsError("Invalid email or password")
        
        return user
    
    async def update_user(self, user_id: int, user_data: UserUpdate) -> User:
        """Update user information."""
        user = await self.user_repository.get_by_id(user_id)
        
        if not user:
            raise ValueError(f"User {user_id} not found")
        
        if user_data.email:
            user.email = user_data.email
        if user_data.username:
            user.username = user_data.username
        
        return await self.user_repository.update(user)

tests/unit/test_user_service.py:

import pytest
from unittest.mock import AsyncMock, Mock, patch
from app.services.user_service import UserService
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
from app.exceptions import UserAlreadyExistsError, InvalidCredentialsError

@pytest.fixture
def mock_user_repository():
    """Mock user repository."""
    return AsyncMock()

@pytest.fixture
def user_service(mock_user_repository):
    """Create user service with mocked repository."""
    return UserService(mock_user_repository)

@pytest.mark.unit
class TestUserService:
    
    @pytest.mark.asyncio
    async def test_create_user_success(self, user_service, mock_user_repository):
        """Test successful user creation."""
        # Arrange
        user_data = UserCreate(
            email="test@example.com",
            username="testuser",
            password="SecurePass123!"
        )
        
        mock_user_repository.get_by_email.return_value = None
        mock_user_repository.create.return_value = User(
            id=1,
            email=user_data.email,
            username=user_data.username,
            hashed_password="hashed_password"
        )
        
        # Act
        with patch('app.services.user_service.hash_password') as mock_hash:
            mock_hash.return_value = "hashed_password"
            result = await user_service.create_user(user_data)
        
        # Assert
        assert result.email == user_data.email
        assert result.username == user_data.username
        mock_user_repository.get_by_email.assert_called_once_with(user_data.email)
        mock_user_repository.create.assert_called_once()
    
    @pytest.mark.asyncio
    async def test_create_user_already_exists(self, user_service, mock_user_repository):
        """Test user creation when email already exists."""
        # Arrange
        user_data = UserCreate(
            email="existing@example.com",
            username="testuser",
            password="SecurePass123!"
        )
        
        mock_user_repository.get_by_email.return_value = User(
            id=1,
            email=user_data.email,
            username="existinguser",
            hashed_password="hashed"
        )
        
        # Act & Assert
        with pytest.raises(UserAlreadyExistsError):
            await user_service.create_user(user_data)
        
        mock_user_repository.create.assert_not_called()
    
    @pytest.mark.asyncio
    async def test_authenticate_success(self, user_service, mock_user_repository):
        """Test successful authentication."""
        # Arrange
        email = "test@example.com"
        password = "SecurePass123!"
        
        mock_user = User(
            id=1,
            email=email,
            username="testuser",
            hashed_password="hashed_password"
        )
        
        mock_user_repository.get_by_email.return_value = mock_user
        
        # Act
        with patch('app.services.user_service.verify_password') as mock_verify:
            mock_verify.return_value = True
            result = await user_service.authenticate(email, password)
        
        # Assert
        assert result == mock_user
        mock_user_repository.get_by_email.assert_called_once_with(email)
    
    @pytest.mark.asyncio
    async def test_authenticate_invalid_credentials(self, user_service, mock_user_repository):
        """Test authentication with invalid credentials."""
        # Arrange
        mock_user_repository.get_by_email.return_value = User(
            id=1,
            email="test@example.com",
            username="testuser",
            hashed_password="hashed_password"
        )
        
        # Act & Assert
        with patch('app.services.user_service.verify_password') as mock_verify:
            mock_verify.return_value = False
            with pytest.raises(InvalidCredentialsError):
                await user_service.authenticate("test@example.com", "wrong_password")

Testing Repositories

tests/unit/test_user_repository.py:

import pytest
from app.repositories.user_repository import UserRepository
from app.models.user import User

@pytest.mark.unit
class TestUserRepository:
    
    @pytest.mark.asyncio
    async def test_create_user(self, db_session):
        """Test creating a user in the database."""
        # Arrange
        repository = UserRepository(db_session)
        user = User(
            email="test@example.com",
            username="testuser",
            hashed_password="hashed_password"
        )
        
        # Act
        result = await repository.create(user)
        await db_session.commit()
        
        # Assert
        assert result.id is not None
        assert result.email == "test@example.com"
        assert result.username == "testuser"
    
    @pytest.mark.asyncio
    async def test_get_by_email(self, db_session):
        """Test retrieving user by email."""
        # Arrange
        repository = UserRepository(db_session)
        user = User(
            email="find@example.com",
            username="finduser",
            hashed_password="hashed"
        )
        await repository.create(user)
        await db_session.commit()
        
        # Act
        found_user = await repository.get_by_email("find@example.com")
        
        # Assert
        assert found_user is not None
        assert found_user.email == "find@example.com"
    
    @pytest.mark.asyncio
    async def test_get_by_email_not_found(self, db_session):
        """Test retrieving non-existent user."""
        # Arrange
        repository = UserRepository(db_session)
        
        # Act
        result = await repository.get_by_email("nonexistent@example.com")
        
        # Assert
        assert result is None

Integration Testing {#integration-testing}

Integration tests verify that different components work together correctly.

Testing API Endpoints

tests/integration/test_user_api.py:

import pytest
from httpx import AsyncClient
from app.models.user import User

@pytest.mark.integration
class TestUserAPI:
    
    @pytest.mark.asyncio
    async def test_create_user_endpoint(self, client: AsyncClient):
        """Test POST /users endpoint."""
        # Arrange
        user_data = {
            "email": "newuser@example.com",
            "username": "newuser",
            "password": "SecurePass123!"
        }
        
        # Act
        response = await client.post("/api/v1/users", json=user_data)
        
        # Assert
        assert response.status_code == 201
        data = response.json()
        assert data["email"] == user_data["email"]
        assert data["username"] == user_data["username"]
        assert "id" in data
        assert "password" not in data
        assert "hashed_password" not in data
    
    @pytest.mark.asyncio
    async def test_create_user_duplicate_email(self, client: AsyncClient, db_session):
        """Test creating user with duplicate email."""
        # Arrange
        existing_user = User(
            email="duplicate@example.com",
            username="existing",
            hashed_password="hashed"
        )
        db_session.add(existing_user)
        await db_session.commit()
        
        user_data = {
            "email": "duplicate@example.com",
            "username": "newuser",
            "password": "SecurePass123!"
        }
        
        # Act
        response = await client.post("/api/v1/users", json=user_data)
        
        # Assert
        assert response.status_code == 409
        assert "already exists" in response.json()["detail"]
    
    @pytest.mark.asyncio
    async def test_get_user_by_id(self, client: AsyncClient, db_session):
        """Test GET /users/{id} endpoint."""
        # Arrange
        user = User(
            email="getuser@example.com",
            username="getuser",
            hashed_password="hashed"
        )
        db_session.add(user)
        await db_session.commit()
        await db_session.refresh(user)
        
        # Act
        response = await client.get(f"/api/v1/users/{user.id}")
        
        # Assert
        assert response.status_code == 200
        data = response.json()
        assert data["id"] == user.id
        assert data["email"] == user.email
    
    @pytest.mark.asyncio
    async def test_login_success(self, client: AsyncClient, db_session):
        """Test POST /auth/login endpoint."""
        # Arrange
        from app.utils.password import hash_password
        
        user = User(
            email="login@example.com",
            username="loginuser",
            hashed_password=hash_password("SecurePass123!")
        )
        db_session.add(user)
        await db_session.commit()
        
        login_data = {
            "email": "login@example.com",
            "password": "SecurePass123!"
        }
        
        # Act
        response = await client.post("/api/v1/auth/login", json=login_data)
        
        # Assert
        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert "refresh_token" in data
        assert data["token_type"] == "bearer"
    
    @pytest.mark.asyncio
    async def test_protected_endpoint_requires_auth(self, client: AsyncClient):
        """Test that protected endpoints require authentication."""
        # Act
        response = await client.get("/api/v1/users/me")
        
        # Assert
        assert response.status_code == 401
    
    @pytest.mark.asyncio
    async def test_protected_endpoint_with_auth(self, client: AsyncClient, db_session):
        """Test accessing protected endpoint with valid token."""
        # Arrange
        from app.utils.password import hash_password
        from app.utils.jwt import create_access_token
        
        user = User(
            email="protected@example.com",
            username="protecteduser",
            hashed_password=hash_password("SecurePass123!")
        )
        db_session.add(user)
        await db_session.commit()
        await db_session.refresh(user)
        
        token = create_access_token({"sub": str(user.id)})
        
        # Act
        response = await client.get(
            "/api/v1/users/me",
            headers={"Authorization": f"Bearer {token}"}
        )
        
        # Assert
        assert response.status_code == 200
        data = response.json()
        assert data["email"] == user.email

Contract Testing {#contract-testing}

Contract tests ensure that services communicate correctly with each other.

tests/contract/test_user_service_contract.py:

import pytest
from pact import Consumer, Provider, Like, EachLike
import asyncio

@pytest.fixture(scope="module")
def pact():
    """Setup Pact consumer."""
    pact = Consumer("OrderService").has_pact_with(Provider("UserService"))
    pact.start_service()
    yield pact
    pact.stop_service()

@pytest.mark.contract
class TestUserServiceContract:
    
    def test_get_user_by_id_contract(self, pact):
        """Verify contract for GET /users/{id}."""
        expected = {
            "id": Like(1),
            "email": Like("user@example.com"),
            "username": Like("username"),
            "created_at": Like("2024-01-01T00:00:00Z"),
        }
        
        (pact
         .given("user with id 1 exists")
         .upon_receiving("a request for user with id 1")
         .with_request("GET", "/api/v1/users/1")
         .will_respond_with(200, body=expected))
        
        with pact:
            # Make actual request to mock server
            import httpx
            response = httpx.get(f"{pact.uri}/api/v1/users/1")
            
            assert response.status_code == 200
            assert "id" in response.json()
            assert "email" in response.json()
    
    def test_create_user_contract(self, pact):
        """Verify contract for POST /users."""
        request_body = {
            "email": "newuser@example.com",
            "username": "newuser",
            "password": "SecurePass123!"
        }
        
        expected_response = {
            "id": Like(1),
            "email": Like("newuser@example.com"),
            "username": Like("newuser"),
            "created_at": Like("2024-01-01T00:00:00Z"),
        }
        
        (pact
         .given("user service is available")
         .upon_receiving("a request to create a user")
         .with_request("POST", "/api/v1/users", body=request_body)
         .will_respond_with(201, body=expected_response))
        
        with pact:
            import httpx
            response = httpx.post(
                f"{pact.uri}/api/v1/users",
                json=request_body
            )
            
            assert response.status_code == 201
            assert response.json()["email"] == request_body["email"]

End-to-End Testing {#end-to-end-testing}

E2E tests verify complete user flows across multiple services.

tests/e2e/test_user_flow.py:

import pytest
from httpx import AsyncClient

@pytest.mark.e2e
@pytest.mark.slow
class TestUserFlow:
    
    @pytest.mark.asyncio
    async def test_complete_user_registration_and_login_flow(self, client: AsyncClient):
        """Test complete user flow: register -> verify email -> login -> access protected resource."""
        
        # Step 1: Register new user
        user_data = {
            "email": "e2e@example.com",
            "username": "e2euser",
            "password": "SecurePass123!"
        }
        
        register_response = await client.post("/api/v1/users", json=user_data)
        assert register_response.status_code == 201
        user_id = register_response.json()["id"]
        
        # Step 2: Login
        login_data = {
            "email": user_data["email"],
            "password": user_data["password"]
        }
        
        login_response = await client.post("/api/v1/auth/login", json=login_data)
        assert login_response.status_code == 200
        tokens = login_response.json()
        access_token = tokens["access_token"]
        
        # Step 3: Access protected resource
        headers = {"Authorization": f"Bearer {access_token}"}
        profile_response = await client.get("/api/v1/users/me", headers=headers)
        assert profile_response.status_code == 200
        profile = profile_response.json()
        assert profile["id"] == user_id
        assert profile["email"] == user_data["email"]
        
        # Step 4: Update profile
        update_data = {"username": "updated_username"}
        update_response = await client.patch(
            f"/api/v1/users/{user_id}",
            json=update_data,
            headers=headers
        )
        assert update_response.status_code == 200
        assert update_response.json()["username"] == "updated_username"
        
        # Step 5: Logout (invalidate token)
        logout_response = await client.post("/api/v1/auth/logout", headers=headers)
        assert logout_response.status_code == 200
        
        # Step 6: Verify token is invalid
        invalid_response = await client.get("/api/v1/users/me", headers=headers)
        assert invalid_response.status_code == 401

Database Testing Strategies {#database-testing-strategies}

Using Test Containers

tests/conftest.py (enhanced):

import pytest
import asyncio
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres_container():
    """Start PostgreSQL container for tests."""
    with PostgresContainer("postgres:15-alpine") as postgres:
        yield postgres

@pytest.fixture(scope="session")
async def test_engine(postgres_container):
    """Create test database engine with containerized PostgreSQL."""
    database_url = postgres_container.get_connection_url().replace(
        "psycopg2", "asyncpg"
    )
    
    engine = create_async_engine(
        database_url,
        poolclass=NullPool,
        echo=False,
    )
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    yield engine
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    
    await engine.dispose()

Database Fixtures and Factories

tests/factories.py:

from faker import Faker
from app.models.user import User
from app.models.post import Post

fake = Faker()

class UserFactory:
    """Factory for creating test users."""
    
    @staticmethod
    def build(**kwargs) -> User:
        """Build a user instance without saving."""
        defaults = {
            "email": fake.email(),
            "username": fake.user_name(),
            "hashed_password": "hashed_password",
            "is_active": True,
            "is_verified": False,
        }
        defaults.update(kwargs)
        return User(**defaults)
    
    @staticmethod
    async def create(db_session, **kwargs) -> User:
        """Create and save a user to the database."""
        user = UserFactory.build(**kwargs)
        db_session.add(user)
        await db_session.commit()
        await db_session.refresh(user)
        return user

class PostFactory:
    """Factory for creating test posts."""
    
    @staticmethod
    def build(user_id: int, **kwargs) -> Post:
        """Build a post instance without saving."""
        defaults = {
            "title": fake.sentence(),
            "content": fake.text(),
            "user_id": user_id,
            "published": False,
        }
        defaults.update(kwargs)
        return Post(**defaults)
    
    @staticmethod
    async def create(db_session, user_id: int, **kwargs) -> Post:
        """Create and save a post to the database."""
        post = PostFactory.build(user_id, **kwargs)
        db_session.add(post)
        await db_session.commit()
        await db_session.refresh(post)
        return post

Usage in tests:

@pytest.mark.asyncio
async def test_user_can_create_post(db_session):
    """Test user can create a post."""
    from tests.factories import UserFactory, PostFactory
    
    # Arrange
    user = await UserFactory.create(db_session)
    
    # Act
    post = await PostFactory.create(
        db_session,
        user_id=user.id,
        title="My Test Post"
    )
    
    # Assert
    assert post.id is not None
    assert post.user_id == user.id
    assert post.title == "My Test Post"

Testing Database Migrations

tests/integration/test_migrations.py:

import pytest
from alembic import command
from alembic.config import Config

@pytest.mark.integration
def test_migrations_can_run_forward_and_backward():
    """Test that all migrations can be applied and reverted."""
    alembic_cfg = Config("alembic.ini")
    
    # Downgrade to base
    command.downgrade(alembic_cfg, "base")
    
    # Upgrade to head
    command.upgrade(alembic_cfg, "head")
    
    # Downgrade one version
    command.downgrade(alembic_cfg, "-1")
    
    # Upgrade back to head
    command.upgrade(alembic_cfg, "head")

Testing in CI/CD {#testing-in-cicd}

GitHub Actions Workflow

.github/workflows/test.yml:

name: Test Suite

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main, develop]

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:15-alpine
        env:
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_pass
          POSTGRES_DB: test_db
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      
      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
          cache: 'pip'
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-dev.txt
      
      - name: Run linting
        run: |
          ruff check .
          black --check .
          mypy app/
      
      - name: Run unit tests
        run: pytest tests/unit -v --cov=app --cov-report=xml
        env:
          DATABASE_URL: postgresql+asyncpg://test_user:test_pass@localhost:5432/test_db
          REDIS_URL: redis://localhost:6379/0
      
      - name: Run integration tests
        run: pytest tests/integration -v
        env:
          DATABASE_URL: postgresql+asyncpg://test_user:test_pass@localhost:5432/test_db
          REDIS_URL: redis://localhost:6379/0
      
      - name: Run contract tests
        run: pytest tests/contract -v
      
      - name: Run E2E tests
        run: pytest tests/e2e -v --maxfail=1
        env:
          DATABASE_URL: postgresql+asyncpg://test_user:test_pass@localhost:5432/test_db
          REDIS_URL: redis://localhost:6379/0
      
      - name: Upload coverage reports
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml
          flags: unittests
          name: codecov-umbrella
      
      - name: Check coverage threshold
        run: |
          coverage report --fail-under=80

Docker Compose for Local Testing

docker-compose.test.yml:

version: '3.8'

services:
  test-db:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: test_user
      POSTGRES_PASSWORD: test_pass
      POSTGRES_DB: test_db
    ports:
      - "5433:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U test_user"]
      interval: 5s
      timeout: 5s
      retries: 5
  
  test-redis:
    image: redis:7-alpine
    ports:
      - "6380:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5
  
  test-runner:
    build:
      context: .
      dockerfile: Dockerfile.test
    depends_on:
      test-db:
        condition: service_healthy
      test-redis:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql+asyncpg://test_user:test_pass@test-db:5432/test_db
      REDIS_URL: redis://test-redis:6379/0
    volumes:
      - .:/app
      - /app/.pytest_cache
    command: pytest tests/ -v --cov=app --cov-report=html

Dockerfile.test:

FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    postgresql-client \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt requirements-dev.txt ./
RUN pip install --no-cache-dir -r requirements.txt -r requirements-dev.txt

# Copy application code
COPY . .

# Run tests by default
CMD ["pytest", "tests/", "-v"]

Run tests locally:

# Start test services and run tests
docker-compose -f docker-compose.test.yml up --abort-on-container-exit

# Run specific test category
docker-compose -f docker-compose.test.yml run test-runner pytest tests/unit -v

# Clean up
docker-compose -f docker-compose.test.yml down -v

Common Pitfalls and Solutions {#common-pitfalls}

1. Test Isolation Issues

Problem: Tests fail when run together but pass individually.

Solution:

@pytest.fixture(autouse=True)
async def clean_database(db_session):
    """Clean database before each test."""
    yield
    # Rollback any uncommitted changes
    await db_session.rollback()
    
    # Clear all tables
    for table in reversed(Base.metadata.sorted_tables):
        await db_session.execute(table.delete())
    await db_session.commit()

2. Async Test Deadlocks

Problem: Async tests hang or timeout.

Solution:

# Use pytest-timeout
@pytest.mark.timeout(5)
@pytest.mark.asyncio
async def test_with_timeout():
    await some_async_operation()

# Or configure globally in pytest.ini
[pytest]
timeout = 10

3. Flaky Tests Due to Timing

Problem: Tests intermittently fail due to race conditions.

Solution:

import asyncio
from tenacity import retry, stop_after_delay, wait_fixed

@retry(stop=stop_after_delay(5), wait=wait_fixed(0.1))
async def wait_for_condition(check_func):
    """Wait for a condition to be true."""
    result = await check_func()
    if not result:
        raise Exception("Condition not met")
    return result

# Usage
async def test_eventual_consistency(client):
    # Create resource
    response = await client.post("/api/v1/resources", json=data)
    resource_id = response.json()["id"]
    
    # Wait for async processing
    async def check_processed():
        resp = await client.get(f"/api/v1/resources/{resource_id}")
        return resp.json()["status"] == "processed"
    
    await wait_for_condition(check_processed)

4. Mocking External Services

Problem: Tests depend on external APIs.

Solution:

import responses
import pytest

@pytest.fixture
def mock_external_api():
    """Mock external API calls."""
    with responses.RequestsMock() as rsps:
        rsps.add(
            responses.GET,
            "https://api.external.com/users/1",
            json={"id": 1, "name": "Test User"},
            status=200
        )
        yield rsps

@pytest.mark.asyncio
async def test_with_mocked_external_api(mock_external_api):
    """Test that uses mocked external API."""
    result = await fetch_external_user(1)
    assert result["name"] == "Test User"

5. Testing Background Tasks

Problem: Background tasks don't complete during tests.

Solution:

from fastapi import BackgroundTasks
import asyncio

@pytest.mark.asyncio
async def test_background_task(client):
    """Test endpoint with background task."""
    # Use a flag to track task completion
    task_completed = asyncio.Event()
    
    async def mock_background_task(*args, **kwargs):
        # Perform task
        await asyncio.sleep(0.1)
        task_completed.set()
    
    # Replace background task with mock
    with patch('app.tasks.send_email', side_effect=mock_background_task):
        response = await client.post("/api/v1/send-notification")
        assert response.status_code == 200
        
        # Wait for background task
        await asyncio.wait_for(task_completed.wait(), timeout=1.0)

6. Database Connection Pool Exhaustion

Problem: Tests fail with "connection pool exhausted" errors.

Solution:

from sqlalchemy.pool import NullPool

# Use NullPool for tests
test_engine = create_async_engine(
    TEST_DATABASE_URL,
    poolclass=NullPool,  # Don't pool connections in tests
    echo=False,
)

# Or configure pool size
test_engine = create_async_engine(
    TEST_DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
)

Best Practices Summary

Write tests first - Use TDD for critical business logic

Keep tests fast - Unit tests should run in milliseconds

Isolate tests - Each test should be independent

Use factories - Generate test data with factories, not fixtures

Test behavior, not implementation - Focus on what, not how

Mock external dependencies - Never call real external APIs in tests

Use test containers - Run real databases in containers for integration tests

Measure coverage - Aim for 80%+ coverage, but don't obsess over 100%

Run tests in CI - Automate testing on every commit

Keep tests maintainable - Refactor tests like production code

Conclusion

Testing microservices requires a multi-layered approach. By combining unit tests for business logic, integration tests for component interaction, contract tests for service boundaries, and E2E tests for critical flows, you can build confidence in your FastAPI microservices.

Remember:

Fast feedback is crucial - optimize test execution time

Test isolation prevents flaky tests and debugging nightmares

Realistic test data makes tests more valuable

Continuous testing in CI/CD catches issues early

Start with unit tests for your core business logic, add integration tests for database operations and API endpoints, implement contract tests for service boundaries, and reserve E2E tests for critical user journeys.

With these practices in place, you'll ship more reliable microservices with confidence.

---

Additional Resources:

FastAPI Testing Documentation

pytest Documentation

Pact Contract Testing

TestContainers Python