Backend Testing (Python/FastAPI)

Stack Overview

Tool	Purpose
pytest	Test runner and framework
pytest-asyncio	Async test support
pytest-cov	Coverage reporting
pytest-xdist	Parallel test execution
httpx	Async HTTP client for API tests
Testcontainers	Real PostgreSQL for E2E tests
freezegun	Time manipulation for tests

---

Test Organization

tests/
├── conftest.py              # Shared fixtures (db, client, auth)
├── api/
│   ├── conftest.py          # API-specific fixtures
│   ├── test_auth.py         # Authentication endpoints
│   ├── test_users.py        # User endpoints
│   └── dependencies/
│       └── test_auth_dependencies.py
├── services/
│   ├── test_auth_service.py
│   └── test_email_service.py
├── crud/
│   ├── test_base.py         # Generic CRUD operations
│   └── test_user.py
├── models/
│   └── test_user.py
└── e2e/
    ├── conftest.py          # E2E fixtures (Testcontainers)
    └── test_user_workflows.py

---

Fixtures & Factories

Database Fixtures

Use function scope for test isolation — each test gets a fresh database:

# tests/conftest.py
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.models import Base


@pytest_asyncio.fixture(scope="function")
async def async_test_db():
    """Create a fresh in-memory database for each test."""
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    AsyncTestingSessionLocal = sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )

    yield engine, AsyncTestingSessionLocal

    await engine.dispose()

User Fixtures

@pytest_asyncio.fixture
async def test_user(async_test_db):
    """Create a standard test user."""
    _, AsyncTestingSessionLocal = async_test_db

    async with AsyncTestingSessionLocal() as session:
        user = User(
            email="testuser@example.com",
            password_hash=get_password_hash("TestPassword123!"),
            is_active=True,
        )
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user


@pytest_asyncio.fixture
async def test_superuser(async_test_db):
    """Create an admin test user."""
    _, AsyncTestingSessionLocal = async_test_db

    async with AsyncTestingSessionLocal() as session:
        user = User(
            email="superuser@example.com",
            password_hash=get_password_hash("AdminPassword123!"),
            is_active=True,
            is_superuser=True,
        )
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

Client Fixtures

from httpx import AsyncClient, ASGITransport
from app.main import app
from app.api.dependencies import get_db


@pytest_asyncio.fixture
async def client(async_test_db):
    """Async test client with database override."""
    _, AsyncTestingSessionLocal = async_test_db

    async def override_get_db():
        async with AsyncTestingSessionLocal() as session:
            yield session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        yield client

    app.dependency_overrides.clear()

Token Fixtures

@pytest_asyncio.fixture
async def user_token(client, test_user):
    """Get an access token for the test user."""
    response = await client.post(
        "/api/v1/auth/login",
        json={"email": "testuser@example.com", "password": "TestPassword123!"}
    )
    return response.json()["access_token"]


@pytest_asyncio.fixture
async def superuser_token(client, test_superuser):
    """Get an access token for the superuser."""
    response = await client.post(
        "/api/v1/auth/login",
        json={"email": "superuser@example.com", "password": "AdminPassword123!"}
    )
    return response.json()["access_token"]

---

API Testing Patterns

Basic Endpoint Test

import pytest


class TestUserEndpoints:

    @pytest.mark.asyncio
    async def test_get_current_user_returns_user_data(self, client, user_token):
        response = await client.get(
            "/api/v1/users/me",
            headers={"Authorization": f"Bearer {user_token}"}
        )

        assert response.status_code == 200
        data = response.json()
        assert data["email"] == "testuser@example.com"
        assert "password" not in data  # Sensitive data excluded

    @pytest.mark.asyncio
    async def test_get_current_user_without_token_returns_401(self, client):
        response = await client.get("/api/v1/users/me")

        assert response.status_code == 401

Testing All Response Codes

class TestLoginEndpoint:

    @pytest.mark.asyncio
    async def test_login_with_valid_credentials_returns_tokens(self, client, test_user):
        response = await client.post(
            "/api/v1/auth/login",
            json={"email": "testuser@example.com", "password": "TestPassword123!"}
        )

        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert "refresh_token" in data

    @pytest.mark.asyncio
    async def test_login_with_wrong_password_returns_401(self, client, test_user):
        response = await client.post(
            "/api/v1/auth/login",
            json={"email": "testuser@example.com", "password": "WrongPassword"}
        )

        assert response.status_code == 401
        assert response.json()["detail"] == "Invalid credentials"

    @pytest.mark.asyncio
    async def test_login_with_nonexistent_user_returns_401(self, client):
        response = await client.post(
            "/api/v1/auth/login",
            json={"email": "nobody@example.com", "password": "Password123!"}
        )

        assert response.status_code == 401
        # Same message as wrong password (prevent user enumeration)
        assert response.json()["detail"] == "Invalid credentials"

    @pytest.mark.asyncio
    async def test_login_with_invalid_email_returns_422(self, client):
        response = await client.post(
            "/api/v1/auth/login",
            json={"email": "not-an-email", "password": "Password123!"}
        )

        assert response.status_code == 422  # Pydantic validation error

Class-Based Organization

Group related tests in classes:

class TestListUsers:
    """GET /api/v1/users (admin only)"""

    @pytest.mark.asyncio
    async def test_list_users_as_admin_returns_users(self, client, superuser_token):
        response = await client.get(
            "/api/v1/users",
            headers={"Authorization": f"Bearer {superuser_token}"}
        )
        assert response.status_code == 200
        assert isinstance(response.json(), list)

    @pytest.mark.asyncio
    async def test_list_users_as_regular_user_returns_403(self, client, user_token):
        response = await client.get(
            "/api/v1/users",
            headers={"Authorization": f"Bearer {user_token}"}
        )
        assert response.status_code == 403

    @pytest.mark.asyncio
    async def test_list_users_without_auth_returns_401(self, client):
        response = await client.get("/api/v1/users")
        assert response.status_code == 401

---

Service Layer Testing

Test business logic directly, mocking only external dependencies:

from unittest.mock import AsyncMock, patch
import pytest

from app.services.email import EmailService


class TestEmailService:

    @pytest.mark.asyncio
    async def test_send_password_reset_sends_email_with_token(self):
        # Arrange
        backend_mock = AsyncMock()
        backend_mock.send_email = AsyncMock(return_value=True)
        service = EmailService(backend=backend_mock)

        # Act
        await service.send_password_reset(
            email="user@example.com",
            token="abc123",
            reset_url="https://app.sartiq.com/reset"
        )

        # Assert
        backend_mock.send_email.assert_called_once()
        call_kwargs = backend_mock.send_email.call_args.kwargs
        assert call_kwargs["to"] == "user@example.com"
        assert "abc123" in call_kwargs["html_content"]

    @pytest.mark.asyncio
    async def test_send_password_reset_handles_backend_failure(self):
        backend_mock = AsyncMock()
        backend_mock.send_email = AsyncMock(return_value=False)
        service = EmailService(backend=backend_mock)

        with pytest.raises(EmailDeliveryError):
            await service.send_password_reset(
                email="user@example.com",
                token="abc123",
                reset_url="https://app.sartiq.com/reset"
            )

---

Database Testing

Using SQLite for Speed

For most tests, SQLite in-memory is fast and sufficient:

@pytest.mark.asyncio
async def test_create_user_persists_to_database(async_test_db):
    _, AsyncTestingSessionLocal = async_test_db

    async with AsyncTestingSessionLocal() as session:
        user = User(email="new@example.com", password_hash="hashed")
        session.add(user)
        await session.commit()

        # Verify persistence
        result = await session.execute(
            select(User).where(User.email == "new@example.com")
        )
        saved_user = result.scalar_one()
        assert saved_user.id is not None

Using PostgreSQL for E2E

For tests that need PostgreSQL-specific features or full fidelity:

# tests/e2e/conftest.py
import pytest
from testcontainers.postgres import PostgresContainer


@pytest.fixture(scope="session")
def postgres_container():
    """Start PostgreSQL container once per test session."""
    with PostgresContainer("postgres:17-alpine") as postgres:
        yield postgres


@pytest_asyncio.fixture(scope="function")
async def e2e_db_session(postgres_container):
    """Fresh schema per test using real PostgreSQL."""
    url = postgres_container.get_connection_url()
    async_url = url.replace("postgresql://", "postgresql+asyncpg://")

    engine = create_async_engine(async_url)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession)

    async with AsyncSessionLocal() as session:
        yield session

    await engine.dispose()

Dependency Override Pattern

Override FastAPI dependencies to inject test databases:

from app.api.dependencies import get_db


async def override_get_db():
    async with AsyncTestingSessionLocal() as session:
        yield session


app.dependency_overrides[get_db] = override_get_db
# ... run tests ...
app.dependency_overrides.clear()  # Clean up after

---

Mocking Strategy

Mock at Boundaries

Mock external services, not internal logic:

# Good: Mock the email backend
with patch("app.services.email.SMTPBackend.send") as mock_send:
    mock_send.return_value = True
    await email_service.send_welcome_email(user)

# Bad: Mock internal method
with patch.object(email_service, "_format_template"):  # Don't do this
    ...

AsyncMock for Async Code

from unittest.mock import AsyncMock, patch


@pytest.mark.asyncio
async def test_payment_service_charges_card():
    with patch("app.services.payment.stripe_client") as mock_stripe:
        mock_stripe.charges.create = AsyncMock(return_value={"id": "ch_123"})

        result = await payment_service.charge(amount=1000, customer_id="cus_456")

        assert result.charge_id == "ch_123"
        mock_stripe.charges.create.assert_called_once_with(
            amount=1000,
            currency="usd",
            customer="cus_456"
        )

Simulating Database Errors

from sqlalchemy.exc import IntegrityError


@pytest.mark.asyncio
async def test_create_user_handles_duplicate_email(async_test_db):
    _, AsyncTestingSessionLocal = async_test_db

    async with AsyncTestingSessionLocal() as session:
        # Create first user
        user1 = User(email="dupe@example.com", password_hash="hash1")
        session.add(user1)
        await session.commit()

        # Attempt duplicate
        user2 = User(email="dupe@example.com", password_hash="hash2")
        session.add(user2)

        with pytest.raises(IntegrityError):
            await session.commit()

---

Error & Edge Case Testing

Testing Exceptions

import pytest
from app.exceptions import NotFoundError


@pytest.mark.asyncio
async def test_get_user_raises_not_found_for_missing_user(user_service):
    with pytest.raises(NotFoundError) as exc_info:
        await user_service.get_user(id="nonexistent-uuid")

    assert "User not found" in str(exc_info.value)

Testing Validation Errors

@pytest.mark.asyncio
async def test_create_user_rejects_weak_password(client):
    response = await client.post(
        "/api/v1/auth/register",
        json={"email": "user@example.com", "password": "123"}
    )

    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any("password" in e["loc"] for e in errors)

Parameterized Tests

Test multiple scenarios efficiently:

import pytest


@pytest.mark.parametrize("email,expected_valid", [
    ("user@example.com", True),
    ("user@sartiq.com", True),
    ("user@subdomain.example.com", True),
    ("invalid", False),
    ("@example.com", False),
    ("user@", False),
    ("", False),
])
def test_email_validation(email, expected_valid):
    result = validate_email(email)
    assert result == expected_valid

---

Running Tests

# Run all tests
pytest

# Run with verbose output
pytest -v

# Stop on first failure
pytest -x

# Run specific file
pytest tests/api/test_auth.py

# Run specific test
pytest tests/api/test_auth.py::TestLogin::test_login_success -v

# Run by marker
pytest -m "not e2e"              # Skip E2E tests
pytest -m e2e                    # Only E2E tests

# Run with coverage
pytest --cov=app --cov-report=term-missing

# Run in parallel (faster)
pytest -n auto

---

Coverage Configuration

In pyproject.toml:

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
addopts = "-v --tb=short --strict-markers"
markers = [
    "e2e: End-to-end tests with real database",
    "slow: Tests that take a long time",
]

[tool.coverage.run]
source = ["app"]
branch = true
omit = [
    "app/alembic/*",
    "app/scripts/*",
    "tests/*",
]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "raise NotImplementedError",
]
fail_under = 80

---

Checklist

Before merging:

• All tests pass locally (pytest)
• New code has test coverage
• Tests are independent (can run in any order)
• No flaky tests introduced
• Edge cases covered (empty, null, invalid, concurrent)
• Error paths tested (401, 403, 404, 422, 500)
• Async code tested with @pytest.mark.asyncio