Python Testing

Quick Reference

When to Use This Skill

Use for testing Python code:

• Writing pytest tests with fixtures
• Mocking external dependencies
• Testing async code
• Setting up code coverage
• Testing FastAPI applications

Related skills:

• For FastAPI: see python-fastapi
• For async patterns: see python-asyncio
• For CI/CD: see python-github-actions

---

Python Testing Best Practices (2025)

Overview

Modern Python testing centers around pytest as the de facto standard, with additional tools for coverage, mocking, and async testing.

Pytest Fundamentals

Installation

# With uv
uv add --dev pytest pytest-cov pytest-asyncio pytest-xdist

# With pip
pip install pytest pytest-cov pytest-asyncio pytest-xdist

Basic Test Structure

# tests/test_calculator.py
import pytest
from mypackage.calculator import add, divide

def test_add_positive_numbers():
    assert add(2, 3) == 5

def test_add_negative_numbers():
    assert add(-2, -3) == -5

def test_add_mixed_numbers():
    assert add(-2, 3) == 1

def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

def test_divide_result():
    result = divide(10, 2)
    assert result == pytest.approx(5.0)

Running Tests

# Run all tests
pytest

# Verbose output
pytest -v

# Run specific file
pytest tests/test_calculator.py

# Run specific test
pytest tests/test_calculator.py::test_add_positive_numbers

# Run tests matching pattern
pytest -k "add"

# Stop on first failure
pytest -x

# Run last failed tests
pytest --lf

# Parallel execution
pytest -n auto

Fixtures

Basic Fixtures

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def sample_user():
    """Provides a sample user dictionary."""
    return {
        "name": "John Doe",
        "email": "john@example.com",
        "age": 30,
    }

@pytest.fixture
def db_session():
    """Provides a database session that rolls back after test."""
    engine = create_engine("sqlite:///:memory:")
    Session = sessionmaker(bind=engine)
    session = Session()

    yield session

    session.rollback()
    session.close()

Fixture Scopes

@pytest.fixture(scope="function")  # Default: new for each test
def per_test_fixture():
    return create_resource()

@pytest.fixture(scope="class")  # Shared within test class
def per_class_fixture():
    return create_expensive_resource()

@pytest.fixture(scope="module")  # Shared within test module
def per_module_fixture():
    return create_very_expensive_resource()

@pytest.fixture(scope="session")  # Shared across entire test session
def per_session_fixture():
    resource = create_global_resource()
    yield resource
    cleanup_global_resource(resource)

Parameterized Fixtures

@pytest.fixture(params=["sqlite", "postgresql", "mysql"])
def database_type(request):
    return request.param

def test_with_multiple_databases(database_type):
    # This test runs 3 times, once for each database type
    db = create_connection(database_type)
    assert db.is_connected()

Parameterized Tests

Basic Parametrize

import pytest

@pytest.mark.parametrize("input,expected", [
    (1, 1),
    (2, 4),
    (3, 9),
    (4, 16),
])
def test_square(input, expected):
    assert input ** 2 == expected

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3),
    (-1, 1, 0),
    (0, 0, 0),
    (100, 200, 300),
])
def test_add(a, b, expected):
    assert add(a, b) == expected

Parametrize with IDs

@pytest.mark.parametrize("input,expected", [
    pytest.param(1, 1, id="one"),
    pytest.param(2, 4, id="two"),
    pytest.param(3, 9, id="three"),
])
def test_square_with_ids(input, expected):
    assert input ** 2 == expected

# Run with: pytest -v
# Shows: test_square_with_ids[one] PASSED

Combining Parametrize

@pytest.mark.parametrize("x", [1, 2])
@pytest.mark.parametrize("y", [3, 4])
def test_combinations(x, y):
    # Runs 4 times: (1,3), (1,4), (2,3), (2,4)
    assert x + y in [4, 5, 6]

Mocking

Using pytest-mock

from unittest.mock import MagicMock, patch

def test_with_mock(mocker):
    # Mock a method
    mock_api = mocker.patch("mypackage.api.fetch_data")
    mock_api.return_value = {"status": "success"}

    result = process_data()
    assert result["status"] == "success"
    mock_api.assert_called_once()

def test_mock_return_values(mocker):
    # Different return values for successive calls
    mock_func = mocker.patch("mypackage.service.get_item")
    mock_func.side_effect = [
        {"id": 1},
        {"id": 2},
        ValueError("Not found"),
    ]

    assert get_item(1) == {"id": 1}
    assert get_item(2) == {"id": 2}

    with pytest.raises(ValueError):
        get_item(3)

Context Manager Mocking

def test_mock_context_manager(mocker):
    mock_open = mocker.patch("builtins.open", mocker.mock_open(read_data="file content"))

    with open("test.txt") as f:
        content = f.read()

    assert content == "file content"
    mock_open.assert_called_once_with("test.txt")

Mocking Classes

def test_mock_class(mocker):
    MockUser = mocker.patch("mypackage.models.User")
    mock_instance = MockUser.return_value
    mock_instance.name = "Test User"
    mock_instance.save.return_value = True

    user = create_user("Test User")
    assert user.name == "Test User"
    mock_instance.save.assert_called_once()

Async Testing

pytest-asyncio

import pytest
import asyncio
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_function():
    result = await async_fetch_data()
    assert result is not None

@pytest.mark.asyncio
async def test_async_with_fixture(async_client: AsyncClient):
    response = await async_client.get("/api/users")
    assert response.status_code == 200

# Fixture for async client
@pytest.fixture
async def async_client():
    async with AsyncClient(base_url="http://test") as client:
        yield client

Async Fixtures

@pytest.fixture
async def async_db_session():
    engine = create_async_engine("postgresql+asyncpg://...")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async_session = sessionmaker(engine, class_=AsyncSession)

    async with async_session() as session:
        yield session
        await session.rollback()

Testing FastAPI

import pytest
from httpx import AsyncClient, ASGITransport
from myapp.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        yield client

@pytest.mark.asyncio
async def test_read_main(client: AsyncClient):
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

@pytest.mark.asyncio
async def test_create_item(client: AsyncClient):
    response = await client.post(
        "/items/",
        json={"name": "Test Item", "price": 10.0},
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Test Item"

Coverage

Configuration

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = [
    "-ra",
    "-q",
    "--strict-markers",
    "--cov=src",
    "--cov-report=term-missing",
    "--cov-report=html",
    "--cov-fail-under=80",
]

[tool.coverage.run]
branch = true
source = ["src"]
omit = ["*/tests/*", "*/__init__.py"]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise NotImplementedError",
    "if TYPE_CHECKING:",
    "if __name__ == .__main__.:",
]

Running with Coverage

# Generate coverage report
pytest --cov=src --cov-report=html

# Check coverage threshold
pytest --cov=src --cov-fail-under=80

# Coverage for specific paths
pytest --cov=src/mypackage --cov-report=term-missing

Test Organization

Recommended Structure

tests/
├── conftest.py          # Shared fixtures
├── unit/                # Unit tests
│   ├── conftest.py      # Unit test fixtures
│   ├── test_models.py
│   └── test_utils.py
├── integration/         # Integration tests
│   ├── conftest.py
│   └── test_api.py
├── e2e/                 # End-to-end tests
│   └── test_workflows.py
└── fixtures/            # Test data
    ├── users.json
    └── products.json

Conftest Hierarchy

# tests/conftest.py - Shared across all tests
@pytest.fixture
def app_config():
    return {"debug": True, "testing": True}

# tests/unit/conftest.py - Unit test specific
@pytest.fixture
def mock_database(mocker):
    return mocker.MagicMock()

# tests/integration/conftest.py - Integration test specific
@pytest.fixture
def real_database():
    db = create_test_database()
    yield db
    db.cleanup()

Markers

Built-in Markers

import pytest

@pytest.mark.skip(reason="Not implemented yet")
def test_future_feature():
    pass

@pytest.mark.skipif(sys.platform == "win32", reason="Unix only")
def test_unix_specific():
    pass

@pytest.mark.xfail(reason="Known bug, see issue #123")
def test_known_failure():
    assert False

@pytest.mark.slow
def test_slow_operation():
    time.sleep(10)

Custom Markers

# pytest.ini or pyproject.toml
[tool.pytest.ini_options]
markers = [
    "slow: marks tests as slow",
    "integration: marks tests as integration tests",
    "requires_db: marks tests that need database",
]

# Run only slow tests
pytest -m slow

# Skip slow tests
pytest -m "not slow"

# Combine markers
pytest -m "integration and not slow"

Property-Based Testing

Hypothesis

from hypothesis import given, strategies as st

@given(st.integers(), st.integers())
def test_addition_commutative(a, b):
    assert add(a, b) == add(b, a)

@given(st.lists(st.integers()))
def test_sort_idempotent(items):
    sorted_once = sorted(items)
    sorted_twice = sorted(sorted_once)
    assert sorted_once == sorted_twice

@given(st.text(min_size=1))
def test_string_reverse_twice(s):
    assert s[::-1][::-1] == s

# Custom strategy
@given(st.builds(
    User,
    name=st.text(min_size=1, max_size=100),
    email=st.emails(),
    age=st.integers(min_value=0, max_value=150),
))
def test_user_validation(user):
    assert user.is_valid()

Snapshot Testing

syrupy

from syrupy.assertion import SnapshotAssertion

def test_api_response(snapshot: SnapshotAssertion):
    response = get_api_response()
    assert response == snapshot

def test_html_output(snapshot: SnapshotAssertion):
    html = render_template("user.html", user=sample_user)
    assert html == snapshot(extension_class=HTMLSnapshotExtension)

Best Practices

1. Test Naming

# Clear, descriptive names
def test_user_creation_with_valid_email_succeeds():
    ...

def test_user_creation_with_invalid_email_raises_validation_error():
    ...

def test_empty_cart_returns_zero_total():
    ...

2. Arrange-Act-Assert

def test_order_total():
    # Arrange
    order = Order()
    order.add_item(Item(price=10.00))
    order.add_item(Item(price=20.00))

    # Act
    total = order.calculate_total()

    # Assert
    assert total == 30.00

3. One Assertion Per Test (Generally)

# Prefer multiple focused tests
def test_user_has_correct_name():
    user = create_user("John")
    assert user.name == "John"

def test_user_has_default_role():
    user = create_user("John")
    assert user.role == "member"

# Over one test with many assertions
def test_user_creation():  # Less focused
    user = create_user("John")
    assert user.name == "John"
    assert user.role == "member"
    assert user.is_active
    assert user.created_at is not None

4. Use Fixtures for Setup

# Good: Reusable fixture
@pytest.fixture
def authenticated_user(db_session):
    user = User(name="Test", email="test@example.com")
    db_session.add(user)
    db_session.commit()
    return user

def test_user_can_post(authenticated_user):
    post = authenticated_user.create_post("Hello")
    assert post.author == authenticated_user

# Avoid: Repeated setup in each test
def test_user_can_post_bad():
    user = User(name="Test", email="test@example.com")
    db_session.add(user)  # Repeated in every test
    db_session.commit()
    post = user.create_post("Hello")
    assert post.author == user

Additional References

For production-ready fixture patterns beyond this guide, see:

• [Pytest Fixtures Cookbook](references/pytest-fixtures-cookbook.md) - SQLAlchemy async sessions, PostgreSQL with Docker, user factories, generic model factories, authentication fixtures, external API mocks, time freezing, environment fixtures, file fixtures, parametrized fixtures, cleanup patterns, Redis fixtures

Pytest Fixtures Cookbook

Production-ready pytest fixtures for common testing scenarios.

Database Fixtures

SQLAlchemy Async Session

# tests/conftest.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.pool import StaticPool

from app.database import Base
from app.main import app
from app.dependencies import get_db


@pytest.fixture(scope="session")
def event_loop():
    """Create event loop for session scope."""
    import asyncio
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session")
async def engine():
    """Create test database engine."""
    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)

    yield engine

    await engine.dispose()


@pytest.fixture
async def db_session(engine):
    """Database session with automatic rollback."""
    async_session = async_sessionmaker(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):
    """FastAPI test client with database override."""
    from httpx import AsyncClient, ASGITransport

    async def override_get_db():
        yield db_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()

PostgreSQL with Docker

import pytest
import asyncpg
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 pg_pool(postgres_container):
    """PostgreSQL connection pool."""
    pool = await asyncpg.create_pool(
        postgres_container.get_connection_url().replace("postgresql://", "postgresql://")
    )
    yield pool
    await pool.close()


@pytest.fixture
async def pg_connection(pg_pool):
    """PostgreSQL connection with transaction rollback."""
    async with pg_pool.acquire() as conn:
        tr = conn.transaction()
        await tr.start()
        yield conn
        await tr.rollback()

Factory Fixtures

User Factory

from dataclasses import dataclass, field
from typing import Any
from faker import Faker

fake = Faker()

@dataclass
class UserFactory:
    """Factory for creating test users."""
    db_session: AsyncSession
    defaults: dict[str, Any] = field(default_factory=dict)

    async def create(self, **overrides) -> User:
        data = {
            "email": fake.email(),
            "full_name": fake.name(),
            "hashed_password": "hashed_test_password",
            "is_active": True,
            **self.defaults,
            **overrides,
        }
        user = User(**data)
        self.db_session.add(user)
        await self.db_session.flush()
        return user

    async def create_batch(self, count: int, **overrides) -> list[User]:
        return [await self.create(**overrides) for _ in range(count)]


@pytest.fixture
def user_factory(db_session):
    """User factory fixture."""
    return UserFactory(db_session)


# Usage
async def test_list_users(client, user_factory):
    await user_factory.create_batch(5)

    response = await client.get("/users")

    assert response.status_code == 200
    assert len(response.json()) == 5

Generic Factory

from typing import TypeVar, Generic, Type

T = TypeVar("T")

class ModelFactory(Generic[T]):
    """Generic factory for SQLAlchemy models."""

    def __init__(self, session: AsyncSession, model: Type[T], defaults: dict = None):
        self.session = session
        self.model = model
        self.defaults = defaults or {}

    async def create(self, **overrides) -> T:
        data = {**self.defaults, **overrides}
        instance = self.model(**data)
        self.session.add(instance)
        await self.session.flush()
        return instance


@pytest.fixture
def model_factory(db_session):
    """Generic model factory."""
    def factory(model: Type[T], defaults: dict = None) -> ModelFactory[T]:
        return ModelFactory(db_session, model, defaults)
    return factory


# Usage
async def test_orders(client, model_factory):
    user_factory = model_factory(User, {"email": "test@test.com"})
    order_factory = model_factory(Order, {"status": "pending"})

    user = await user_factory.create()
    order = await order_factory.create(user_id=user.id)

Authentication Fixtures

import jwt
from datetime import datetime, timedelta

@pytest.fixture
def auth_token():
    """Generate authentication token."""
    def _generate(user_id: int, **extra) -> str:
        payload = {
            "sub": str(user_id),
            "exp": datetime.utcnow() + timedelta(hours=1),
            **extra,
        }
        return jwt.encode(payload, "test-secret", algorithm="HS256")
    return _generate


@pytest.fixture
async def authenticated_user(user_factory, auth_token):
    """Create user and return with auth token."""
    user = await user_factory.create()
    token = auth_token(user.id)
    return user, token


@pytest.fixture
async def auth_client(client, authenticated_user):
    """Client with authentication headers."""
    user, token = authenticated_user
    client.headers["Authorization"] = f"Bearer {token}"
    return client, user


# Usage
async def test_protected_endpoint(auth_client):
    client, user = auth_client
    response = await client.get("/users/me")
    assert response.status_code == 200
    assert response.json()["id"] == user.id

Mock Fixtures

External API Mock

import pytest
from unittest.mock import AsyncMock, MagicMock
from httpx import Response

@pytest.fixture
def mock_external_api(mocker):
    """Mock external API calls."""
    mock = AsyncMock()

    def _setup(responses: dict[str, dict]):
        async def mock_get(url, **kwargs):
            for pattern, data in responses.items():
                if pattern in url:
                    return Response(
                        status_code=data.get("status", 200),
                        json=data.get("json", {}),
                    )
            raise ValueError(f"Unexpected URL: {url}")

        mock.get = mock_get
        return mock

    return _setup


# Usage
async def test_fetch_user_data(mock_external_api):
    mock = mock_external_api({
        "/users/123": {"json": {"name": "John", "email": "john@example.com"}},
        "/users/456": {"status": 404, "json": {"error": "Not found"}},
    })

    result = await fetch_user(123, client=mock)
    assert result["name"] == "John"

Time Freezing

from freezegun import freeze_time
from datetime import datetime

@pytest.fixture
def frozen_time():
    """Freeze time for tests."""
    with freeze_time("2024-01-15 12:00:00") as frozen:
        yield frozen


# Usage
def test_created_at_timestamp(frozen_time, db_session):
    user = User(email="test@test.com")
    db_session.add(user)
    db_session.commit()

    assert user.created_at == datetime(2024, 1, 15, 12, 0, 0)

    frozen_time.tick(timedelta(hours=1))
    user.update()
    assert user.updated_at == datetime(2024, 1, 15, 13, 0, 0)

Environment Fixtures

import os
import pytest

@pytest.fixture
def env_vars(monkeypatch):
    """Set environment variables for test."""
    def _set(**vars):
        for key, value in vars.items():
            monkeypatch.setenv(key, value)
    return _set


@pytest.fixture
def temp_settings(env_vars):
    """Override settings for test."""
    def _override(**settings):
        env_vars(**{k.upper(): str(v) for k, v in settings.items()})
        # Reload settings
        from app.config import settings
        settings.__init__()
    return _override


# Usage
def test_debug_mode(temp_settings):
    temp_settings(debug=True, log_level="DEBUG")

    from app.config import settings
    assert settings.debug is True

File Fixtures

import pytest
from pathlib import Path
import tempfile

@pytest.fixture
def temp_dir():
    """Temporary directory for test files."""
    with tempfile.TemporaryDirectory() as tmpdir:
        yield Path(tmpdir)


@pytest.fixture
def sample_file(temp_dir):
    """Create sample file."""
    def _create(name: str, content: str) -> Path:
        path = temp_dir / name
        path.write_text(content)
        return path
    return _create


@pytest.fixture
def sample_csv(sample_file):
    """Create sample CSV file."""
    content = """id,name,email
1,John,john@example.com
2,Jane,jane@example.com
3,Bob,bob@example.com"""
    return sample_file("data.csv", content)


# Usage
async def test_import_csv(client, sample_csv):
    with open(sample_csv, "rb") as f:
        response = await client.post(
            "/import",
            files={"file": ("data.csv", f, "text/csv")}
        )
    assert response.status_code == 200

Parametrized Fixtures

@pytest.fixture(params=["sqlite", "postgresql"])
def database_url(request):
    """Test with multiple databases."""
    if request.param == "sqlite":
        return "sqlite+aiosqlite:///:memory:"
    elif request.param == "postgresql":
        # Would need a real PostgreSQL for this
        pytest.skip("PostgreSQL not available")
    return None


@pytest.fixture(params=[
    ("admin", True),
    ("user", False),
    ("guest", False),
])
def user_with_role(request, user_factory):
    """Test with different user roles."""
    role, is_admin = request.param
    return user_factory.create(role=role, is_admin=is_admin)

Cleanup Fixtures

@pytest.fixture
def cleanup():
    """Register cleanup functions."""
    cleanups = []

    def _register(func):
        cleanups.append(func)

    yield _register

    for func in reversed(cleanups):
        func()


# Usage
async def test_create_resource(cleanup, client):
    response = await client.post("/resources", json={"name": "test"})
    resource_id = response.json()["id"]

    # Register cleanup
    cleanup(lambda: asyncio.run(client.delete(f"/resources/{resource_id}")))

    # Test continues...
    assert response.status_code == 201

Redis Fixtures

import pytest
import redis.asyncio as redis

@pytest.fixture(scope="session")
async def redis_client():
    """Redis client for tests."""
    client = redis.from_url("redis://localhost:6379/15")  # Use test database
    await client.flushdb()  # Clean before tests
    yield client
    await client.flushdb()  # Clean after tests
    await client.close()


@pytest.fixture
async def clean_redis(redis_client):
    """Redis with cleanup after each test."""
    yield redis_client
    await redis_client.flushdb()


# Usage
async def test_caching(clean_redis):
    await clean_redis.set("key", "value")
    result = await clean_redis.get("key")
    assert result == b"value"