Pytest Patterns

Pytest Patterns - Comprehensive Testing Guide

A comprehensive skill for mastering Python testing with pytest. This skill covers everything from basic test structure to advanced patterns including fixtures, parametrization, mocking, test organization, coverage analysis, and CI/CD integration.

When to Use This Skill

Use this skill when:

• Writing tests for Python applications (web apps, APIs, CLI tools, libraries)
• Setting up test infrastructure for a new Python project
• Refactoring existing tests to be more maintainable and efficient
• Implementing test-driven development (TDD) workflows
• Creating fixture patterns for database, API, or external service testing
• Organizing large test suites with hundreds or thousands of tests
• Debugging failing tests or improving test reliability
• Setting up continuous integration testing pipelines
• Measuring and improving code coverage
• Writing integration, unit, or end-to-end tests
• Testing async Python code
• Mocking external dependencies and services

Core Concepts

What is pytest?

pytest is a mature, full-featured Python testing framework that makes it easy to write simple tests, yet scales to support complex functional testing. It provides:

• Simple syntax: Use plain assert statements instead of special assertion methods
• Powerful fixtures: Modular, composable test setup and teardown
• Parametrization: Run the same test with different inputs
• Plugin ecosystem: Hundreds of plugins for extended functionality
• Detailed reporting: Clear failure messages and debugging information
• Test discovery: Automatic test collection following naming conventions

pytest vs unittest

# unittest (traditional)
import unittest

class TestMath(unittest.TestCase):
    def test_addition(self):
        self.assertEqual(2 + 2, 4)

# pytest (simpler)
def test_addition():
    assert 2 + 2 == 4

Test Discovery Rules

pytest automatically discovers tests by following these conventions:

1. Test files: test_*.py or *_test.py 2. Test functions: Functions prefixed with test_ 3. Test classes: Classes prefixed with Test (no __init__ method) 4. Test methods: Methods prefixed with test_ inside Test classes

Fixtures - The Heart of pytest

What are Fixtures?

Fixtures provide a fixed baseline for tests to run reliably and repeatably. They handle setup, provide test data, and perform cleanup.

Basic Fixture Pattern

import pytest

@pytest.fixture
def sample_data():
    """Provides sample data for testing."""
    return {"name": "Alice", "age": 30}

def test_data_access(sample_data):
    assert sample_data["name"] == "Alice"
    assert sample_data["age"] == 30

Fixture Scopes

Fixtures can have different scopes controlling how often they're created:

• function (default): Created for each test function
• class: Created once per test class
• module: Created once per test module
• package: Created once per test package
• session: Created once per test session

@pytest.fixture(scope="session")
def database_connection():
    """Database connection created once for entire test session."""
    conn = create_db_connection()
    yield conn
    conn.close()  # Cleanup after all tests

@pytest.fixture(scope="module")
def api_client():
    """API client created once per test module."""
    client = APIClient()
    client.authenticate()
    yield client
    client.logout()

@pytest.fixture  # scope="function" is default
def temp_file():
    """Temporary file created for each test."""
    import tempfile
    f = tempfile.NamedTemporaryFile(mode='w', delete=False)
    yield f.name
    os.unlink(f.name)

Fixture Dependencies

Fixtures can depend on other fixtures, creating a dependency graph:

@pytest.fixture
def database():
    db = Database()
    db.connect()
    yield db
    db.disconnect()

@pytest.fixture
def user_repository(database):
    """Depends on database fixture."""
    return UserRepository(database)

@pytest.fixture
def sample_user(user_repository):
    """Depends on user_repository, which depends on database."""
    user = user_repository.create(name="Test User")
    yield user
    user_repository.delete(user.id)

def test_user_operations(sample_user):
    """Uses sample_user fixture (which uses user_repository and database)."""
    assert sample_user.name == "Test User"

Autouse Fixtures

Fixtures that run automatically without being explicitly requested:

@pytest.fixture(autouse=True)
def reset_database():
    """Runs before every test automatically."""
    clear_database()
    seed_test_data()

@pytest.fixture(autouse=True, scope="session")
def configure_logging():
    """Configure logging once for entire test session."""
    import logging
    logging.basicConfig(level=logging.DEBUG)

Fixture Factories

Fixtures that return functions for creating test data:

@pytest.fixture
def make_user():
    """Factory fixture for creating users."""
    users = []

    def _make_user(name, email=None):
        user = User(name=name, email=email or f"{name}@example.com")
        users.append(user)
        return user

    yield _make_user

    # Cleanup all created users
    for user in users:
        user.delete()

def test_multiple_users(make_user):
    user1 = make_user("Alice")
    user2 = make_user("Bob", email="bob@test.com")
    assert user1.name == "Alice"
    assert user2.email == "bob@test.com"

Parametrization - Testing Multiple Cases

Basic Parametrization

Run the same test with different inputs:

import pytest

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

Multiple Parameters

@pytest.mark.parametrize("x", [0, 1])
@pytest.mark.parametrize("y", [2, 3])
def test_combinations(x, y):
    """Runs 4 times: (0,2), (0,3), (1,2), (1,3)."""
    assert x < y

Parametrizing with IDs

Make test output more readable:

@pytest.mark.parametrize("test_input,expected", [
    pytest.param("3+5", 8, id="addition"),
    pytest.param("2*4", 8, id="multiplication"),
    pytest.param("10-2", 8, id="subtraction"),
])
def test_eval(test_input, expected):
    assert eval(test_input) == expected

# Output:
# test_eval[addition] PASSED
# test_eval[multiplication] PASSED
# test_eval[subtraction] PASSED

Parametrizing Fixtures

Create fixture instances with different values:

@pytest.fixture(params=["mysql", "postgresql", "sqlite"])
def database_type(request):
    """Test runs three times, once for each database."""
    return request.param

def test_database_connection(database_type):
    conn = connect_to_database(database_type)
    assert conn.is_connected()

Combining Parametrization and Marks

@pytest.mark.parametrize("test_input,expected", [
    ("valid@email.com", True),
    ("invalid-email", False),
    pytest.param("edge@case", True, marks=pytest.mark.xfail),
    pytest.param("slow@test.com", True, marks=pytest.mark.slow),
])
def test_email_validation(test_input, expected):
    assert is_valid_email(test_input) == expected

Indirect Parametrization

Pass parameters through fixtures:

@pytest.fixture
def database(request):
    """Create database based on parameter."""
    db_type = request.param
    db = Database(db_type)
    db.connect()
    yield db
    db.close()

@pytest.mark.parametrize("database", ["mysql", "postgres"], indirect=True)
def test_database_operations(database):
    """database fixture receives the parameter value."""
    assert database.is_connected()
    database.execute("SELECT 1")

Mocking and Monkeypatching

Using pytest's monkeypatch

The monkeypatch fixture provides safe patching that's automatically undone:

def test_get_user_env(monkeypatch):
    """Test environment variable access."""
    monkeypatch.setenv("USER", "testuser")
    assert os.getenv("USER") == "testuser"

def test_remove_env(monkeypatch):
    """Test with missing environment variable."""
    monkeypatch.delenv("PATH", raising=False)
    assert os.getenv("PATH") is None

def test_modify_path(monkeypatch):
    """Test sys.path modification."""
    monkeypatch.syspath_prepend("/custom/path")
    assert "/custom/path" in sys.path

Mocking Functions and Methods

import requests

def get_user_data(user_id):
    response = requests.get(f"https://api.example.com/users/{user_id}")
    return response.json()

def test_get_user_data(monkeypatch):
    """Mock external API call."""
    class MockResponse:
        @staticmethod
        def json():
            return {"id": 1, "name": "Test User"}

    def mock_get(*args, **kwargs):
        return MockResponse()

    monkeypatch.setattr(requests, "get", mock_get)

    result = get_user_data(1)
    assert result["name"] == "Test User"

Using unittest.mock

from unittest.mock import Mock, MagicMock, patch, call

def test_with_mock():
    """Basic mock usage."""
    mock_db = Mock()
    mock_db.get_user.return_value = {"id": 1, "name": "Alice"}

    user = mock_db.get_user(1)
    assert user["name"] == "Alice"
    mock_db.get_user.assert_called_once_with(1)

def test_with_patch():
    """Patch during test execution."""
    with patch('mymodule.database.get_connection') as mock_conn:
        mock_conn.return_value = Mock()
        # Test code that uses database.get_connection()
        assert mock_conn.called

@patch('mymodule.send_email')
def test_notification(mock_email):
    """Patch as decorator."""
    send_notification("test@example.com", "Hello")
    mock_email.assert_called_once()

Mock Return Values and Side Effects

def test_mock_return_values():
    """Different return values for sequential calls."""
    mock_api = Mock()
    mock_api.fetch.side_effect = [
        {"status": "pending"},
        {"status": "processing"},
        {"status": "complete"}
    ]

    assert mock_api.fetch()["status"] == "pending"
    assert mock_api.fetch()["status"] == "processing"
    assert mock_api.fetch()["status"] == "complete"

def test_mock_exception():
    """Mock raising exceptions."""
    mock_service = Mock()
    mock_service.connect.side_effect = ConnectionError("Failed to connect")

    with pytest.raises(ConnectionError):
        mock_service.connect()

Spy Pattern - Partial Mocking

def test_spy_pattern(monkeypatch):
    """Spy on a function while preserving original behavior."""
    original_function = mymodule.process_data
    call_count = 0

    def spy_function(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        return original_function(*args, **kwargs)

    monkeypatch.setattr(mymodule, "process_data", spy_function)

    result = mymodule.process_data([1, 2, 3])
    assert call_count == 1
    assert result is not None  # Original function executed

Test Organization

Directory Structure

project/
├── src/
│   └── mypackage/
│       ├── __init__.py
│       ├── models.py
│       ├── services.py
│       └── utils.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py          # Shared fixtures
│   ├── unit/
│   │   ├── __init__.py
│   │   ├── test_models.py
│   │   └── test_utils.py
│   ├── integration/
│   │   ├── __init__.py
│   │   ├── conftest.py      # Integration-specific fixtures
│   │   └── test_services.py
│   └── e2e/
│       └── test_workflows.py
├── pytest.ini               # pytest configuration
└── setup.py

conftest.py - Sharing Fixtures

The conftest.py file makes fixtures available to all tests in its directory and subdirectories:

# tests/conftest.py
import pytest

@pytest.fixture(scope="session")
def database():
    """Database connection available to all tests."""
    db = Database()
    db.connect()
    yield db
    db.disconnect()

@pytest.fixture
def clean_database(database):
    """Reset database before each test."""
    database.clear_all_tables()
    return database

def pytest_configure(config):
    """Register custom markers."""
    config.addinivalue_line(
        "markers", "slow: marks tests as slow (deselect with '-m \"not slow\"')"
    )
    config.addinivalue_line(
        "markers", "integration: marks tests as integration tests"
    )

Using Markers

Markers allow categorizing and selecting tests:

import pytest

@pytest.mark.slow
def test_slow_operation():
    """Marked as slow test."""
    time.sleep(5)
    assert True

@pytest.mark.integration
def test_api_integration():
    """Marked as integration test."""
    response = requests.get("https://api.example.com")
    assert response.status_code == 200

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

@pytest.mark.skipif(sys.version_info < (3, 8), reason="Requires Python 3.8+")
def test_python38_feature():
    """Conditionally skipped."""
    pass

@pytest.mark.xfail(reason="Known bug in dependency")
def test_known_failure():
    """Expected to fail."""
    assert False

@pytest.mark.parametrize("env", ["dev", "staging", "prod"])
@pytest.mark.integration
def test_environments(env):
    """Multiple markers on one test."""
    assert environment_exists(env)

Running tests with markers:

pytest -m slow                    # Run only slow tests
pytest -m "not slow"              # Skip slow tests
pytest -m "integration and not slow"  # Integration tests that aren't slow
pytest --markers                  # List all available markers

Test Classes for Organization

class TestUserAuthentication:
    """Group related authentication tests."""

    @pytest.fixture(autouse=True)
    def setup(self):
        """Setup for all tests in this class."""
        self.user_service = UserService()

    def test_login_success(self):
        result = self.user_service.login("user", "password")
        assert result.success

    def test_login_failure(self):
        result = self.user_service.login("user", "wrong")
        assert not result.success

    def test_logout(self):
        self.user_service.login("user", "password")
        assert self.user_service.logout()

class TestUserRegistration:
    """Group related registration tests."""

    def test_register_new_user(self):
        pass

    def test_register_duplicate_email(self):
        pass

Coverage Analysis

Installing Coverage Tools

pip install pytest-cov

Running Coverage

# Basic coverage report
pytest --cov=mypackage tests/

# Coverage with HTML report
pytest --cov=mypackage --cov-report=html tests/
# Opens htmlcov/index.html

# Coverage with terminal report
pytest --cov=mypackage --cov-report=term-missing tests/

# Coverage with multiple formats
pytest --cov=mypackage --cov-report=html --cov-report=term tests/

# Fail if coverage below threshold
pytest --cov=mypackage --cov-fail-under=80 tests/

Coverage Configuration

# pytest.ini or setup.cfg
[tool:pytest]
addopts =
    --cov=mypackage
    --cov-report=html
    --cov-report=term-missing
    --cov-fail-under=80

[coverage:run]
source = mypackage
omit =
    */tests/*
    */venv/*
    */__pycache__/*

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

Coverage in Code

def critical_function():  # pragma: no cover
    """Excluded from coverage."""
    pass

if sys.platform == 'win32':  # pragma: no cover
    # Platform-specific code excluded
    pass

pytest Configuration

pytest.ini

[pytest]
# Test discovery
testpaths = tests
python_files = test_*.py *_test.py
python_classes = Test*
python_functions = test_*

# Output options
addopts =
    -ra
    --strict-markers
    --strict-config
    --showlocals
    --tb=short
    --cov=mypackage
    --cov-report=html
    --cov-report=term-missing

# Markers
markers =
    slow: marks tests as slow (deselect with '-m "not slow"')
    integration: marks tests as integration tests
    unit: marks tests as unit tests
    smoke: marks tests as smoke tests
    regression: marks tests as regression tests

# Timeout for tests
timeout = 300

# Minimum Python version
minversion = 7.0

# Directories to ignore
norecursedirs = .git .tox dist build *.egg venv

# Warning filters
filterwarnings =
    error
    ignore::DeprecationWarning

pyproject.toml Configuration

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
addopts = [
    "-ra",
    "--strict-markers",
    "--cov=mypackage",
    "--cov-report=html",
    "--cov-report=term-missing",
]
markers = [
    "slow: marks tests as slow",
    "integration: marks tests as integration tests",
]

[tool.coverage.run]
source = ["mypackage"]
omit = ["*/tests/*", "*/venv/*"]

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

CI/CD Integration

GitHub Actions

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ${{ matrix.os }}
    strategy:
      matrix:
        os: [ubuntu-latest, windows-latest, macos-latest]
        python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']

    steps:
    - uses: actions/checkout@v3

    - name: Set up Python ${{ matrix.python-version }}
      uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -e .[dev]
        pip install pytest pytest-cov pytest-xdist

    - name: Run tests
      run: |
        pytest --cov=mypackage --cov-report=xml --cov-report=term-missing -n auto

    - name: Upload coverage to Codecov
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml
        fail_ci_if_error: true

GitLab CI

# .gitlab-ci.yml
image: python:3.11

stages:
  - test
  - coverage

variables:
  PIP_CACHE_DIR: "$CI_PROJECT_DIR/.cache/pip"

cache:
  paths:
    - .cache/pip
    - venv/

before_script:
  - python -m venv venv
  - source venv/bin/activate
  - pip install -e .[dev]
  - pip install pytest pytest-cov

test:
  stage: test
  script:
    - pytest --junitxml=report.xml --cov=mypackage --cov-report=xml
  artifacts:
    when: always
    reports:
      junit: report.xml
      coverage_report:
        coverage_format: cobertura
        path: coverage.xml

coverage:
  stage: coverage
  script:
    - pytest --cov=mypackage --cov-report=html --cov-fail-under=80
  coverage: '/(?i)total.*? (100(?:\.0+)?\%|[1-9]?\d(?:\.\d+)?\%)$/'
  artifacts:
    paths:
      - htmlcov/

Jenkins Pipeline

// Jenkinsfile
pipeline {
    agent any

    stages {
        stage('Setup') {
            steps {
                sh 'python -m venv venv'
                sh '. venv/bin/activate && pip install -e .[dev]'
                sh '. venv/bin/activate && pip install pytest pytest-cov pytest-html'
            }
        }

        stage('Test') {
            steps {
                sh '. venv/bin/activate && pytest --junitxml=results.xml --html=report.html --cov=mypackage'
            }
            post {
                always {
                    junit 'results.xml'
                    publishHTML([
                        allowMissing: false,
                        alwaysLinkToLastBuild: true,
                        keepAll: true,
                        reportDir: 'htmlcov',
                        reportFiles: 'index.html',
                        reportName: 'Coverage Report'
                    ])
                }
            }
        }
    }
}

Advanced Patterns

Testing Async Code

import pytest
import asyncio

@pytest.fixture
def event_loop():
    """Create event loop for async tests."""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

@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_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_async_operation(), timeout=1.0)

# Using pytest-asyncio plugin
# pip install pytest-asyncio

Testing Database Operations

@pytest.fixture(scope="session")
def database_engine():
    """Create database engine for test session."""
    engine = create_engine("postgresql://test:test@localhost/testdb")
    Base.metadata.create_all(engine)
    yield engine
    Base.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture
def db_session(database_engine):
    """Create new database session for each test."""
    connection = database_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()

def test_user_creation(db_session):
    user = User(name="Test User", email="test@example.com")
    db_session.add(user)
    db_session.commit()

    assert user.id is not None
    assert db_session.query(User).count() == 1

Testing with Temporary Files

@pytest.fixture
def temp_directory(tmp_path):
    """Create temporary directory with sample files."""
    data_dir = tmp_path / "data"
    data_dir.mkdir()

    (data_dir / "config.json").write_text('{"debug": true}')
    (data_dir / "data.csv").write_text("name,value\ntest,42")

    return data_dir

def test_file_processing(temp_directory):
    config = load_config(temp_directory / "config.json")
    assert config["debug"] is True

    data = load_csv(temp_directory / "data.csv")
    assert len(data) == 1

Caplog - Capturing Log Output

import logging

def test_logging_output(caplog):
    """Test that function logs correctly."""
    with caplog.at_level(logging.INFO):
        process_data()

    assert "Processing started" in caplog.text
    assert "Processing completed" in caplog.text
    assert len(caplog.records) == 2

def test_warning_logged(caplog):
    """Test warning is logged."""
    caplog.set_level(logging.WARNING)
    risky_operation()

    assert any(record.levelname == "WARNING" for record in caplog.records)

Capsys - Capturing stdout/stderr

def test_print_output(capsys):
    """Test console output."""
    print("Hello, World!")
    print("Error message", file=sys.stderr)

    captured = capsys.readouterr()
    assert "Hello, World!" in captured.out
    assert "Error message" in captured.err

def test_progressive_output(capsys):
    """Test multiple output captures."""
    print("First")
    captured = capsys.readouterr()
    assert captured.out == "First\n"

    print("Second")
    captured = capsys.readouterr()
    assert captured.out == "Second\n"

Test Examples

Example 1: Basic Unit Test

# test_calculator.py
import pytest
from calculator import add, subtract, multiply, divide

def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0

def test_subtract():
    assert subtract(5, 3) == 2
    assert subtract(0, 5) == -5

def test_multiply():
    assert multiply(3, 4) == 12
    assert multiply(-2, 3) == -6

def test_divide():
    assert divide(10, 2) == 5
    assert divide(7, 2) == 3.5

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

Example 2: Parametrized String Validation

# test_validators.py
import pytest
from validators import is_valid_email, is_valid_phone, is_valid_url

@pytest.mark.parametrize("email,expected", [
    ("user@example.com", True),
    ("user.name+tag@example.co.uk", True),
    ("invalid.email", False),
    ("@example.com", False),
    ("user@", False),
    ("", False),
])
def test_email_validation(email, expected):
    assert is_valid_email(email) == expected

@pytest.mark.parametrize("phone,expected", [
    ("+1-234-567-8900", True),
    ("(555) 123-4567", True),
    ("1234567890", True),
    ("123", False),
    ("abc-def-ghij", False),
])
def test_phone_validation(phone, expected):
    assert is_valid_phone(phone) == expected

@pytest.mark.parametrize("url,expected", [
    ("https://www.example.com", True),
    ("http://example.com/path?query=1", True),
    ("ftp://files.example.com", True),
    ("not a url", False),
    ("http://", False),
])
def test_url_validation(url, expected):
    assert is_valid_url(url) == expected

Example 3: API Testing with Fixtures

# test_api.py
import pytest
import requests
from api_client import APIClient

@pytest.fixture(scope="module")
def api_client():
    """Create API client for test module."""
    client = APIClient(base_url="https://api.example.com")
    client.authenticate(api_key="test-key")
    yield client
    client.close()

@pytest.fixture
def sample_user(api_client):
    """Create sample user for testing."""
    user = api_client.create_user({
        "name": "Test User",
        "email": "test@example.com"
    })
    yield user
    api_client.delete_user(user["id"])

def test_get_user(api_client, sample_user):
    user = api_client.get_user(sample_user["id"])
    assert user["name"] == "Test User"
    assert user["email"] == "test@example.com"

def test_update_user(api_client, sample_user):
    updated = api_client.update_user(sample_user["id"], {
        "name": "Updated Name"
    })
    assert updated["name"] == "Updated Name"

def test_list_users(api_client):
    users = api_client.list_users()
    assert isinstance(users, list)
    assert len(users) > 0

def test_user_not_found(api_client):
    with pytest.raises(requests.HTTPError) as exc:
        api_client.get_user("nonexistent-id")
    assert exc.value.response.status_code == 404

Example 4: Database Testing

# test_models.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from models import Base, User, Post

@pytest.fixture(scope="function")
def db_session():
    """Create clean database session for each test."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    session = Session(engine)

    yield session

    session.close()

@pytest.fixture
def sample_user(db_session):
    """Create sample user."""
    user = User(username="testuser", email="test@example.com")
    db_session.add(user)
    db_session.commit()
    return user

def test_user_creation(db_session):
    user = User(username="newuser", email="new@example.com")
    db_session.add(user)
    db_session.commit()

    assert user.id is not None
    assert db_session.query(User).count() == 1

def test_user_posts(db_session, sample_user):
    post1 = Post(title="First Post", content="Content 1", user=sample_user)
    post2 = Post(title="Second Post", content="Content 2", user=sample_user)
    db_session.add_all([post1, post2])
    db_session.commit()

    assert len(sample_user.posts) == 2
    assert sample_user.posts[0].title == "First Post"

def test_user_deletion_cascades(db_session, sample_user):
    post = Post(title="Post", content="Content", user=sample_user)
    db_session.add(post)
    db_session.commit()

    db_session.delete(sample_user)
    db_session.commit()

    assert db_session.query(Post).count() == 0

Example 5: Mocking External Services

# test_notification_service.py
import pytest
from unittest.mock import Mock, patch
from notification_service import NotificationService, EmailProvider, SMSProvider

@pytest.fixture
def mock_email_provider():
    provider = Mock(spec=EmailProvider)
    provider.send.return_value = {"status": "sent", "id": "email-123"}
    return provider

@pytest.fixture
def mock_sms_provider():
    provider = Mock(spec=SMSProvider)
    provider.send.return_value = {"status": "sent", "id": "sms-456"}
    return provider

@pytest.fixture
def notification_service(mock_email_provider, mock_sms_provider):
    return NotificationService(
        email_provider=mock_email_provider,
        sms_provider=mock_sms_provider
    )

def test_send_email_notification(notification_service, mock_email_provider):
    result = notification_service.send_email(
        to="user@example.com",
        subject="Test",
        body="Test message"
    )

    assert result["status"] == "sent"
    mock_email_provider.send.assert_called_once()
    call_args = mock_email_provider.send.call_args
    assert call_args[1]["to"] == "user@example.com"

def test_send_sms_notification(notification_service, mock_sms_provider):
    result = notification_service.send_sms(
        to="+1234567890",
        message="Test SMS"
    )

    assert result["status"] == "sent"
    mock_sms_provider.send.assert_called_once_with(
        to="+1234567890",
        message="Test SMS"
    )

def test_notification_retry_on_failure(notification_service, mock_email_provider):
    mock_email_provider.send.side_effect = [
        Exception("Network error"),
        Exception("Network error"),
        {"status": "sent", "id": "email-123"}
    ]

    result = notification_service.send_email_with_retry(
        to="user@example.com",
        subject="Test",
        body="Test message",
        max_retries=3
    )

    assert result["status"] == "sent"
    assert mock_email_provider.send.call_count == 3

Example 6: Testing File Operations

# test_file_processor.py
import pytest
from pathlib import Path
from file_processor import process_csv, process_json, FileProcessor

@pytest.fixture
def csv_file(tmp_path):
    """Create temporary CSV file."""
    csv_path = tmp_path / "data.csv"
    csv_path.write_text(
        "name,age,city\n"
        "Alice,30,New York\n"
        "Bob,25,Los Angeles\n"
        "Charlie,35,Chicago\n"
    )
    return csv_path

@pytest.fixture
def json_file(tmp_path):
    """Create temporary JSON file."""
    import json
    json_path = tmp_path / "data.json"
    data = {
        "users": [
            {"name": "Alice", "age": 30},
            {"name": "Bob", "age": 25}
        ]
    }
    json_path.write_text(json.dumps(data))
    return json_path

def test_process_csv(csv_file):
    data = process_csv(csv_file)
    assert len(data) == 3
    assert data[0]["name"] == "Alice"
    assert data[1]["age"] == "25"

def test_process_json(json_file):
    data = process_json(json_file)
    assert len(data["users"]) == 2
    assert data["users"][0]["name"] == "Alice"

def test_file_not_found():
    with pytest.raises(FileNotFoundError):
        process_csv("nonexistent.csv")

def test_file_processor_creates_backup(tmp_path):
    processor = FileProcessor(tmp_path)
    source = tmp_path / "original.txt"
    source.write_text("original content")

    processor.process_with_backup(source)

    backup = tmp_path / "original.txt.bak"
    assert backup.exists()
    assert backup.read_text() == "original content"

Example 7: Testing Classes and Methods

# test_shopping_cart.py
import pytest
from shopping_cart import ShoppingCart, Product

@pytest.fixture
def cart():
    """Create empty shopping cart."""
    return ShoppingCart()

@pytest.fixture
def products():
    """Create sample products."""
    return [
        Product(id=1, name="Book", price=10.99),
        Product(id=2, name="Pen", price=2.50),
        Product(id=3, name="Notebook", price=5.99),
    ]

def test_add_product(cart, products):
    cart.add_product(products[0], quantity=2)
    assert cart.total_items() == 2
    assert cart.subtotal() == 21.98

def test_remove_product(cart, products):
    cart.add_product(products[0], quantity=2)
    cart.remove_product(products[0].id, quantity=1)
    assert cart.total_items() == 1

def test_clear_cart(cart, products):
    cart.add_product(products[0])
    cart.add_product(products[1])
    cart.clear()
    assert cart.total_items() == 0

def test_apply_discount(cart, products):
    cart.add_product(products[0], quantity=2)
    cart.apply_discount(0.10)  # 10% discount
    assert cart.total() == pytest.approx(19.78, rel=0.01)

def test_cannot_add_negative_quantity(cart, products):
    with pytest.raises(ValueError, match="Quantity must be positive"):
        cart.add_product(products[0], quantity=-1)

class TestShoppingCartDiscounts:
    """Test various discount scenarios."""

    @pytest.fixture
    def cart_with_items(self, cart, products):
        cart.add_product(products[0], quantity=2)
        cart.add_product(products[1], quantity=3)
        return cart

    def test_percentage_discount(self, cart_with_items):
        original = cart_with_items.total()
        cart_with_items.apply_discount(0.20)
        assert cart_with_items.total() == original * 0.80

    def test_fixed_discount(self, cart_with_items):
        original = cart_with_items.total()
        cart_with_items.apply_fixed_discount(5.00)
        assert cart_with_items.total() == original - 5.00

    def test_cannot_apply_negative_discount(self, cart_with_items):
        with pytest.raises(ValueError):
            cart_with_items.apply_discount(-0.10)

Example 8: Testing Command-Line Interface

# test_cli.py
import pytest
from click.testing import CliRunner
from myapp.cli import cli

@pytest.fixture
def runner():
    """Create CLI test runner."""
    return CliRunner()

def test_cli_help(runner):
    result = runner.invoke(cli, ['--help'])
    assert result.exit_code == 0
    assert 'Usage:' in result.output

def test_cli_version(runner):
    result = runner.invoke(cli, ['--version'])
    assert result.exit_code == 0
    assert '1.0.0' in result.output

def test_cli_process_file(runner, tmp_path):
    input_file = tmp_path / "input.txt"
    input_file.write_text("test data")

    result = runner.invoke(cli, ['process', str(input_file)])
    assert result.exit_code == 0
    assert 'Processing complete' in result.output

def test_cli_invalid_option(runner):
    result = runner.invoke(cli, ['--invalid-option'])
    assert result.exit_code != 0
    assert 'Error' in result.output

Example 9: Testing Async Functions

# test_async_operations.py
import pytest
import asyncio
from async_service import fetch_data, process_batch, AsyncWorker

@pytest.mark.asyncio
async def test_fetch_data():
    data = await fetch_data("https://api.example.com/data")
    assert data is not None
    assert 'results' in data

@pytest.mark.asyncio
async def test_process_batch():
    items = [1, 2, 3, 4, 5]
    results = await process_batch(items)
    assert len(results) == 5

@pytest.mark.asyncio
async def test_async_worker():
    worker = AsyncWorker()
    await worker.start()

    result = await worker.submit_task("process", data={"key": "value"})
    assert result["status"] == "completed"

    await worker.stop()

@pytest.mark.asyncio
async def test_concurrent_requests():
    async with AsyncWorker() as worker:
        tasks = [
            worker.submit_task("task1"),
            worker.submit_task("task2"),
            worker.submit_task("task3"),
        ]
        results = await asyncio.gather(*tasks)
        assert len(results) == 3

Example 10: Fixture Parametrization

# test_database_backends.py
import pytest
from database import DatabaseConnection

@pytest.fixture(params=['sqlite', 'postgresql', 'mysql'])
def db_connection(request):
    """Test runs three times, once for each database."""
    db = DatabaseConnection(request.param)
    db.connect()
    yield db
    db.disconnect()

def test_database_insert(db_connection):
    """Test insert operation on each database."""
    db_connection.execute("INSERT INTO users (name) VALUES ('test')")
    result = db_connection.execute("SELECT COUNT(*) FROM users")
    assert result[0][0] == 1

def test_database_transaction(db_connection):
    """Test transaction support on each database."""
    with db_connection.transaction():
        db_connection.execute("INSERT INTO users (name) VALUES ('test')")
        db_connection.rollback()

    result = db_connection.execute("SELECT COUNT(*) FROM users")
    assert result[0][0] == 0

Example 11: Testing Exceptions

# test_error_handling.py
import pytest
from custom_errors import ValidationError, AuthenticationError
from validator import validate_user_input
from auth import authenticate_user

def test_validation_error_message():
    with pytest.raises(ValidationError) as exc_info:
        validate_user_input({"email": "invalid"})

    assert "Invalid email format" in str(exc_info.value)
    assert exc_info.value.field == "email"

def test_multiple_validation_errors():
    with pytest.raises(ValidationError) as exc_info:
        validate_user_input({
            "email": "invalid",
            "age": -5
        })

    assert len(exc_info.value.errors) == 2

def test_authentication_error():
    with pytest.raises(AuthenticationError, match="Invalid credentials"):
        authenticate_user("user", "wrong_password")

@pytest.mark.parametrize("input_data,error_type", [
    ({"email": ""}, ValidationError),
    ({"email": None}, ValidationError),
    ({}, ValidationError),
])
def test_various_validation_errors(input_data, error_type):
    with pytest.raises(error_type):
        validate_user_input(input_data)

Example 12: Testing with Fixtures and Mocks

# test_payment_service.py
import pytest
from unittest.mock import Mock, patch
from payment_service import PaymentService, PaymentGateway
from models import Order, PaymentStatus

@pytest.fixture
def mock_gateway():
    gateway = Mock(spec=PaymentGateway)
    gateway.process_payment.return_value = {
        "transaction_id": "tx-12345",
        "status": "success"
    }
    return gateway

@pytest.fixture
def payment_service(mock_gateway):
    return PaymentService(gateway=mock_gateway)

@pytest.fixture
def sample_order():
    return Order(
        id="order-123",
        amount=99.99,
        currency="USD",
        customer_id="cust-456"
    )

def test_successful_payment(payment_service, mock_gateway, sample_order):
    result = payment_service.process_order(sample_order)

    assert result.status == PaymentStatus.SUCCESS
    assert result.transaction_id == "tx-12345"
    mock_gateway.process_payment.assert_called_once()

def test_payment_failure(payment_service, mock_gateway, sample_order):
    mock_gateway.process_payment.return_value = {
        "status": "failed",
        "error": "Insufficient funds"
    }

    result = payment_service.process_order(sample_order)

    assert result.status == PaymentStatus.FAILED
    assert "Insufficient funds" in result.error_message

def test_payment_retry_logic(payment_service, mock_gateway, sample_order):
    mock_gateway.process_payment.side_effect = [
        {"status": "error", "error": "Network timeout"},
        {"status": "error", "error": "Network timeout"},
        {"transaction_id": "tx-12345", "status": "success"}
    ]

    result = payment_service.process_order_with_retry(sample_order, max_retries=3)

    assert result.status == PaymentStatus.SUCCESS
    assert mock_gateway.process_payment.call_count == 3

Example 13: Integration Test Example

# test_integration_workflow.py
import pytest
from app import create_app
from database import db, User, Order

@pytest.fixture(scope="module")
def app():
    """Create application for testing."""
    app = create_app('testing')
    return app

@pytest.fixture(scope="module")
def client(app):
    """Create test client."""
    return app.test_client()

@pytest.fixture(scope="function")
def clean_db(app):
    """Clean database before each test."""
    with app.app_context():
        db.drop_all()
        db.create_all()
        yield db
        db.session.remove()

@pytest.fixture
def authenticated_user(client, clean_db):
    """Create and authenticate user."""
    user = User(username="testuser", email="test@example.com")
    user.set_password("password123")
    clean_db.session.add(user)
    clean_db.session.commit()

    # Login
    response = client.post('/api/auth/login', json={
        'username': 'testuser',
        'password': 'password123'
    })
    token = response.json['access_token']

    return {'user': user, 'token': token}

def test_create_order_workflow(client, authenticated_user):
    """Test complete order creation workflow."""
    headers = {'Authorization': f'Bearer {authenticated_user["token"]}'}

    # Create order
    response = client.post('/api/orders',
        headers=headers,
        json={
            'items': [
                {'product_id': 1, 'quantity': 2},
                {'product_id': 2, 'quantity': 1}
            ]
        }
    )
    assert response.status_code == 201
    order_id = response.json['order_id']

    # Verify order was created
    response = client.get(f'/api/orders/{order_id}', headers=headers)
    assert response.status_code == 200
    assert len(response.json['items']) == 2

    # Update order status
    response = client.patch(f'/api/orders/{order_id}',
        headers=headers,
        json={'status': 'processing'}
    )
    assert response.status_code == 200
    assert response.json['status'] == 'processing'

Example 14: Property-Based Testing

# test_property_based.py
import pytest
from hypothesis import given, strategies as st
from string_utils import reverse_string, is_palindrome

@given(st.text())
def test_reverse_string_twice(s):
    """Reversing twice should return original string."""
    assert reverse_string(reverse_string(s)) == s

@given(st.lists(st.integers()))
def test_sort_idempotent(lst):
    """Sorting twice should be same as sorting once."""
    sorted_once = sorted(lst)
    sorted_twice = sorted(sorted_once)
    assert sorted_once == sorted_twice

@given(st.text(alphabet=st.characters(whitelist_categories=('Lu', 'Ll'))))
def test_palindrome_reverse(s):
    """If a string is a palindrome, its reverse is too."""
    if is_palindrome(s):
        assert is_palindrome(reverse_string(s))

@given(st.integers(min_value=1, max_value=1000))
def test_factorial_positive(n):
    """Factorial should always be positive."""
    from math import factorial
    assert factorial(n) > 0

Example 15: Performance Testing

# test_performance.py
import pytest
import time
from data_processor import process_large_dataset, optimize_query

@pytest.mark.slow
def test_large_dataset_processing_time():
    """Test that large dataset is processed within acceptable time."""
    start = time.time()
    data = list(range(1000000))
    result = process_large_dataset(data)
    duration = time.time() - start

    assert len(result) == 1000000
    assert duration < 5.0  # Should complete in under 5 seconds

@pytest.mark.benchmark
def test_query_optimization(benchmark):
    """Benchmark query performance."""
    result = benchmark(optimize_query, "SELECT * FROM users WHERE active=1")
    assert result is not None

@pytest.mark.parametrize("size", [100, 1000, 10000])
def test_scaling_performance(size):
    """Test performance with different data sizes."""
    data = list(range(size))
    start = time.time()
    result = process_large_dataset(data)
    duration = time.time() - start

    # Should scale linearly
    expected_max_time = size / 100000  # 1 second per 100k items
    assert duration < expected_max_time

Best Practices

Test Organization

1. One test file per source file: mymodule.py → test_mymodule.py 2. Group related tests in classes: Use Test* classes for logical grouping 3. Use descriptive test names: test_user_login_with_invalid_credentials 4. Keep tests independent: Each test should work in isolation 5. Use fixtures for setup: Avoid duplicate setup code

Writing Effective Tests

1. Follow AAA pattern: Arrange, Act, Assert

   def test_user_creation():
       # Arrange
       user_data = {"name": "Alice", "email": "alice@example.com"}

       # Act
       user = create_user(user_data)

       # Assert
       assert user.name == "Alice"

2. Test one thing per test: Each test should verify a single behavior 3. Use descriptive assertions: Make failures easy to understand 4. Avoid test interdependencies: Tests should not depend on execution order 5. Test edge cases: Empty lists, None values, boundary conditions

Fixture Best Practices

1. Use appropriate scope: Minimize fixture creation cost 2. Keep fixtures small: Each fixture should have a single responsibility 3. Use fixture factories: For creating multiple test objects 4. Clean up resources: Use yield for teardown 5. Share fixtures via conftest.py: Make common fixtures available

Coverage Guidelines

1. Aim for high coverage: 80%+ is a good target 2. Focus on critical paths: Prioritize important business logic 3. Don't chase 100%: Some code doesn't need tests (getters, setters) 4. Use coverage to find gaps: Not as a quality metric 5. Exclude generated code: Mark with # pragma: no cover

CI/CD Integration

1. Run tests on every commit: Catch issues early 2. Test on multiple Python versions: Ensure compatibility 3. Generate coverage reports: Track coverage trends 4. Fail on low coverage: Maintain coverage standards 5. Run tests in parallel: Speed up CI pipeline

Useful Plugins

• pytest-cov: Coverage reporting
• pytest-xdist: Parallel test execution
• pytest-asyncio: Async/await support
• pytest-mock: Enhanced mocking
• pytest-timeout: Test timeouts
• pytest-randomly: Randomize test order
• pytest-html: HTML test reports
• pytest-benchmark: Performance benchmarking
• hypothesis: Property-based testing
• pytest-django: Django testing support
• pytest-flask: Flask testing support

Troubleshooting

Tests Not Discovered

• Check file naming: test_*.py or *_test.py
• Check function naming: test_*
• Verify __init__.py files exist in test directories
• Run with -v flag to see discovery process

Fixtures Not Found

• Check fixture is in conftest.py or same file
• Verify fixture scope is appropriate
• Check for typos in fixture name
• Use --fixtures flag to list available fixtures

Test Failures

• Use -v for verbose output
• Use --tb=long for detailed tracebacks
• Use --pdb to drop into debugger on failure
• Use -x to stop on first failure
• Use --lf to rerun last failed tests

Import Errors

• Ensure package is installed: pip install -e .
• Check PYTHONPATH is set correctly
• Verify __init__.py files exist
• Use sys.path manipulation if needed

Resources

• pytest Documentation: https://docs.pytest.org/
• pytest GitHub: https://github.com/pytest-dev/pytest
• pytest Plugins: https://docs.pytest.org/en/latest/reference/plugin_list.html
• Real Python pytest Guide: https://realpython.com/pytest-python-testing/
• Test-Driven Development with Python: https://www.obeythetestinggoat.com/

---

Skill Version: 1.0.0 Last Updated: October 2025 Skill Category: Testing, Python, Quality Assurance, Test Automation Compatible With: pytest 7.0+, Python 3.8+

pytest Fixture Examples

Comprehensive examples demonstrating pytest fixture patterns, from basic to advanced usage.

Table of Contents

1. Basic Fixtures 2. Fixture Scopes 3. Fixture Dependencies 4. Fixture Factories 5. Parametrized Fixtures 6. Autouse Fixtures 7. Fixture Finalization 8. Database Fixtures 9. API Testing Fixtures 10. File and Directory Fixtures 11. Mocking Fixtures 12. Complex Fixture Patterns

Basic Fixtures

Example 1: Simple Data Fixture

import pytest

@pytest.fixture
def sample_user():
    """Provide sample user data."""
    return {
        "id": 1,
        "name": "Alice Smith",
        "email": "alice@example.com",
        "age": 30
    }

def test_user_name(sample_user):
    assert sample_user["name"] == "Alice Smith"

def test_user_email(sample_user):
    assert "@" in sample_user["email"]

def test_user_age(sample_user):
    assert sample_user["age"] >= 18

Example 2: Object Instance Fixture

import pytest
from models import User

@pytest.fixture
def user_instance():
    """Create a User instance for testing."""
    return User(
        username="testuser",
        email="test@example.com",
        first_name="Test",
        last_name="User"
    )

def test_user_full_name(user_instance):
    assert user_instance.full_name() == "Test User"

def test_user_is_active(user_instance):
    assert user_instance.is_active is True

def test_user_string_representation(user_instance):
    assert str(user_instance) == "testuser"

Example 3: List Fixture

import pytest

@pytest.fixture
def number_list():
    """Provide a list of numbers for testing."""
    return [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

def test_list_length(number_list):
    assert len(number_list) == 10

def test_list_sum(number_list):
    assert sum(number_list) == 55

def test_list_contains(number_list):
    assert 5 in number_list
    assert 11 not in number_list

Fixture Scopes

Example 4: Function Scope (Default)

import pytest

@pytest.fixture  # scope="function" is default
def counter():
    """Create new counter for each test."""
    return {"count": 0}

def test_increment_1(counter):
    counter["count"] += 1
    assert counter["count"] == 1

def test_increment_2(counter):
    # New counter created, starts at 0
    counter["count"] += 1
    assert counter["count"] == 1

Example 5: Class Scope

import pytest

@pytest.fixture(scope="class")
def shared_resource():
    """Shared across all tests in a class."""
    print("\nSetup shared resource")
    resource = {"data": []}
    yield resource
    print("\nTeardown shared resource")

class TestResourceSharing:
    def test_add_item_1(self, shared_resource):
        shared_resource["data"].append(1)
        assert len(shared_resource["data"]) == 1

    def test_add_item_2(self, shared_resource):
        # Same resource from previous test
        shared_resource["data"].append(2)
        assert len(shared_resource["data"]) == 2

class TestOtherClass:
    def test_fresh_resource(self, shared_resource):
        # New resource for new class
        assert len(shared_resource["data"]) == 0

Example 6: Module Scope

import pytest
from database import Database

@pytest.fixture(scope="module")
def database_connection():
    """Create database connection once per module."""
    print("\nConnecting to database...")
    db = Database()
    db.connect()
    yield db
    print("\nDisconnecting from database...")
    db.disconnect()

def test_query_1(database_connection):
    result = database_connection.query("SELECT 1")
    assert result is not None

def test_query_2(database_connection):
    # Same connection used
    result = database_connection.query("SELECT 2")
    assert result is not None

Example 7: Session Scope

import pytest
import tempfile
import shutil

@pytest.fixture(scope="session")
def test_data_directory():
    """Create test directory once for entire session."""
    temp_dir = tempfile.mkdtemp(prefix="test_data_")
    print(f"\nCreated test directory: {temp_dir}")

    # Populate with test data
    with open(f"{temp_dir}/config.json", "w") as f:
        f.write('{"debug": true}')

    yield temp_dir

    # Cleanup after all tests
    print(f"\nRemoving test directory: {temp_dir}")
    shutil.rmtree(temp_dir)

def test_config_exists(test_data_directory):
    import os
    assert os.path.exists(f"{test_data_directory}/config.json")

def test_config_content(test_data_directory):
    import json
    with open(f"{test_data_directory}/config.json") as f:
        config = json.load(f)
    assert config["debug"] is True

Fixture Dependencies

Example 8: Simple Dependency Chain

import pytest

@pytest.fixture
def database():
    """Database connection."""
    return {"connected": True, "data": {}}

@pytest.fixture
def user_repository(database):
    """User repository depends on database."""
    class UserRepository:
        def __init__(self, db):
            self.db = db

        def create_user(self, name):
            user_id = len(self.db["data"]) + 1
            self.db["data"][user_id] = {"id": user_id, "name": name}
            return self.db["data"][user_id]

    return UserRepository(database)

@pytest.fixture
def sample_user(user_repository):
    """Sample user depends on user_repository."""
    return user_repository.create_user("Alice")

def test_user_creation(sample_user):
    assert sample_user["name"] == "Alice"
    assert sample_user["id"] == 1

def test_repository_create(user_repository):
    user = user_repository.create_user("Bob")
    assert user["name"] == "Bob"

Example 9: Complex Dependency Graph

import pytest

@pytest.fixture
def config():
    """Application configuration."""
    return {
        "db_host": "localhost",
        "db_port": 5432,
        "api_key": "test-key"
    }

@pytest.fixture
def database(config):
    """Database connection using config."""
    class DB:
        def __init__(self, host, port):
            self.host = host
            self.port = port
            self.connected = True

    return DB(config["db_host"], config["db_port"])

@pytest.fixture
def cache(config):
    """Cache service using config."""
    class Cache:
        def __init__(self):
            self.data = {}

        def get(self, key):
            return self.data.get(key)

        def set(self, key, value):
            self.data[key] = value

    return Cache()

@pytest.fixture
def service(database, cache, config):
    """Service depends on database, cache, and config."""
    class Service:
        def __init__(self, db, cache, api_key):
            self.db = db
            self.cache = cache
            self.api_key = api_key

        def get_data(self, key):
            # Try cache first
            cached = self.cache.get(key)
            if cached:
                return cached

            # Otherwise fetch from database
            data = f"data_for_{key}"
            self.cache.set(key, data)
            return data

    return Service(database, cache, config["api_key"])

def test_service_caching(service):
    # First call - cache miss
    data1 = service.get_data("test")
    assert data1 == "data_for_test"

    # Second call - cache hit
    data2 = service.get_data("test")
    assert data2 == data1

def test_service_has_dependencies(service):
    assert service.db.connected
    assert service.cache is not None
    assert service.api_key == "test-key"

Example 10: Optional Fixture Dependencies

import pytest

@pytest.fixture
def base_config():
    """Base configuration always available."""
    return {"env": "test"}

@pytest.fixture
def extended_config(base_config, request):
    """Extended config with optional additions."""
    config = base_config.copy()

    # Add optional features if markers present
    if "feature_x" in request.keywords:
        config["feature_x"] = True

    if "feature_y" in request.keywords:
        config["feature_y"] = True

    return config

def test_basic_config(extended_config):
    assert extended_config["env"] == "test"
    assert "feature_x" not in extended_config

@pytest.mark.feature_x
def test_with_feature_x(extended_config):
    assert extended_config["feature_x"] is True

@pytest.mark.feature_x
@pytest.mark.feature_y
def test_with_both_features(extended_config):
    assert extended_config["feature_x"] is True
    assert extended_config["feature_y"] is True

Fixture Factories

Example 11: User Factory

import pytest
from models import User

@pytest.fixture
def make_user():
    """Factory for creating users."""
    created_users = []

    def _make_user(username, email=None, **kwargs):
        if email is None:
            email = f"{username}@example.com"

        user = User(username=username, email=email, **kwargs)
        created_users.append(user)
        return user

    yield _make_user

    # Cleanup all created users
    for user in created_users:
        user.delete()

def test_create_single_user(make_user):
    user = make_user("alice")
    assert user.username == "alice"
    assert user.email == "alice@example.com"

def test_create_multiple_users(make_user):
    alice = make_user("alice", is_admin=True)
    bob = make_user("bob", email="bob@test.com")
    charlie = make_user("charlie", age=25)

    assert alice.is_admin is True
    assert bob.email == "bob@test.com"
    assert charlie.age == 25

Example 12: Object Factory with Counter

import pytest

@pytest.fixture
def make_product():
    """Factory for creating unique products."""
    counter = {"count": 0}

    def _make_product(name=None, price=9.99, **kwargs):
        counter["count"] += 1
        if name is None:
            name = f"Product {counter['count']}"

        return {
            "id": counter["count"],
            "name": name,
            "price": price,
            **kwargs
        }

    return _make_product

def test_unique_products(make_product):
    product1 = make_product()
    product2 = make_product()
    product3 = make_product(name="Custom")

    assert product1["id"] == 1
    assert product2["id"] == 2
    assert product3["id"] == 3

    assert product1["name"] == "Product 1"
    assert product2["name"] == "Product 2"
    assert product3["name"] == "Custom"

Example 13: API Request Factory

import pytest
from unittest.mock import Mock

@pytest.fixture
def make_api_request():
    """Factory for creating mock API requests."""
    def _make_request(method="GET", path="/", status=200, data=None):
        request = Mock()
        request.method = method
        request.path = path
        request.status_code = status
        request.json.return_value = data or {}
        request.text = str(data)
        return request

    return _make_request

def test_successful_request(make_api_request):
    request = make_api_request(
        method="POST",
        path="/api/users",
        status=201,
        data={"id": 1, "name": "Alice"}
    )

    assert request.method == "POST"
    assert request.status_code == 201
    assert request.json()["name"] == "Alice"

def test_error_request(make_api_request):
    request = make_api_request(
        status=404,
        data={"error": "Not found"}
    )

    assert request.status_code == 404
    assert "error" in request.json()

Parametrized Fixtures

Example 14: Database Type Parametrization

import pytest
from database import SQLiteDB, PostgresDB, MySQLDB

@pytest.fixture(params=[
    "sqlite",
    "postgresql",
    "mysql"
])
def database(request):
    """Test runs three times with different databases."""
    db_type = request.param

    if db_type == "sqlite":
        db = SQLiteDB(":memory:")
    elif db_type == "postgresql":
        db = PostgresDB("localhost", "testdb")
    elif db_type == "mysql":
        db = MySQLDB("localhost", "testdb")

    db.connect()
    yield db
    db.disconnect()

def test_database_insert(database):
    """Runs 3 times: sqlite, postgresql, mysql."""
    database.execute("INSERT INTO users (name) VALUES ('test')")
    result = database.execute("SELECT COUNT(*) FROM users")
    assert result[0][0] == 1

def test_database_transaction(database):
    """Also runs 3 times with each database."""
    with database.transaction():
        database.execute("INSERT INTO users (name) VALUES ('test')")

Example 15: Parametrized Fixtures with IDs

import pytest

@pytest.fixture(params=[
    pytest.param("dev", id="development"),
    pytest.param("staging", id="staging"),
    pytest.param("prod", id="production"),
])
def environment(request):
    """Test different environments."""
    return {
        "name": request.param,
        "debug": request.param == "dev",
        "api_url": f"https://api.{request.param}.example.com"
    }

def test_environment_config(environment):
    """Runs as: test_environment_config[development], etc."""
    assert environment["name"] in ["dev", "staging", "prod"]
    assert environment["api_url"].startswith("https://")

# Output:
# test_environment_config[development] PASSED
# test_environment_config[staging] PASSED
# test_environment_config[production] PASSED

Example 16: Combining Parametrized Fixtures

import pytest

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

@pytest.fixture(params=[10, 100, 1000])
def record_count(request):
    return request.param

def test_database_performance(database_type, record_count):
    """Runs 6 times: 2 databases × 3 record counts."""
    # Simulate database operation
    import time
    start = time.time()

    # Simulate operation
    for i in range(record_count):
        pass  # Insert record

    duration = time.time() - start
    print(f"\n{database_type} with {record_count} records: {duration:.4f}s")
    assert duration < 1.0  # Performance requirement

Autouse Fixtures

Example 17: Automatic Database Reset

import pytest
from database import db

@pytest.fixture(autouse=True)
def reset_database():
    """Automatically run before each test."""
    db.clear_all_tables()
    db.seed_test_data()
    yield
    # Optional cleanup after test

def test_user_count():
    # Database automatically reset before this test
    assert db.users.count() == 0

def test_create_user():
    # Database automatically reset before this test
    db.users.create(name="Alice")
    assert db.users.count() == 1

Example 18: Automatic Logging Configuration

import pytest
import logging

@pytest.fixture(autouse=True, scope="session")
def configure_logging():
    """Configure logging once for entire test session."""
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger()
    logger.info("Test session started")
    yield
    logger.info("Test session completed")

def test_logging_works():
    logger = logging.getLogger(__name__)
    logger.info("Test is running")
    assert True

Example 19: Automatic Test Timing

import pytest
import time

@pytest.fixture(autouse=True)
def measure_test_duration(request):
    """Measure and report test duration."""
    start = time.time()
    yield
    duration = time.time() - start
    print(f"\n{request.node.name} took {duration:.4f} seconds")

def test_quick_operation():
    time.sleep(0.1)
    assert True

def test_slow_operation():
    time.sleep(0.5)
    assert True

Example 20: Autouse with Class Scope

import pytest

@pytest.fixture(autouse=True, scope="class")
def class_setup(request):
    """Setup once for entire test class."""
    print(f"\nSetting up for {request.cls.__name__}")
    request.cls.shared_data = []
    yield
    print(f"\nTearing down for {request.cls.__name__}")

class TestDataOperations:
    def test_append(self):
        # shared_data available without requesting fixture
        self.shared_data.append(1)
        assert len(self.shared_data) == 1

    def test_extend(self):
        # Same shared_data instance
        self.shared_data.extend([2, 3])
        assert len(self.shared_data) == 3

Fixture Finalization

Example 21: Cleanup with Yield

import pytest
import tempfile
import os

@pytest.fixture
def temp_file():
    """Create and cleanup temporary file."""
    # Setup
    fd, path = tempfile.mkstemp()
    os.write(fd, b"test data")
    os.close(fd)

    yield path

    # Cleanup
    if os.path.exists(path):
        os.unlink(path)

def test_file_exists(temp_file):
    assert os.path.exists(temp_file)
    with open(temp_file, 'rb') as f:
        assert f.read() == b"test data"
# File automatically deleted after test

Example 22: Multiple Cleanup Actions

import pytest

@pytest.fixture
def complex_resource():
    """Resource with multiple cleanup steps."""
    resource = {
        "connection": None,
        "cache": None,
        "temp_files": []
    }

    # Setup
    resource["connection"] = open_connection()
    resource["cache"] = create_cache()

    yield resource

    # Cleanup in reverse order
    for temp_file in resource["temp_files"]:
        os.unlink(temp_file)

    if resource["cache"]:
        resource["cache"].clear()

    if resource["connection"]:
        resource["connection"].close()

Example 23: Conditional Cleanup

import pytest

@pytest.fixture
def database_transaction(request):
    """Database transaction with conditional rollback."""
    db = Database()
    db.begin_transaction()

    yield db

    # Only rollback if test failed
    if request.node.rep_call.failed:
        print("\nTest failed - rolling back transaction")
        db.rollback()
    else:
        print("\nTest passed - committing transaction")
        db.commit()

    db.close()

Database Fixtures

Example 24: SQLAlchemy Session

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from models import Base

@pytest.fixture(scope="session")
def engine():
    """Create database engine once per session."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()

@pytest.fixture(scope="function")
def db_session(engine):
    """Create new database session for each test."""
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()

def test_create_user(db_session):
    from models import User
    user = User(name="Alice", email="alice@example.com")
    db_session.add(user)
    db_session.commit()

    assert user.id is not None
    assert db_session.query(User).count() == 1

Example 25: Database with Sample Data

import pytest
from models import User, Post, Comment

@pytest.fixture
def db_with_users(db_session):
    """Database with sample users."""
    users = [
        User(name="Alice", email="alice@example.com"),
        User(name="Bob", email="bob@example.com"),
        User(name="Charlie", email="charlie@example.com"),
    ]
    db_session.add_all(users)
    db_session.commit()
    return db_session

@pytest.fixture
def db_with_posts(db_with_users):
    """Database with users and posts."""
    users = db_with_users.query(User).all()
    posts = [
        Post(title="First Post", user=users[0]),
        Post(title="Second Post", user=users[1]),
        Post(title="Third Post", user=users[0]),
    ]
    db_with_users.add_all(posts)
    db_with_users.commit()
    return db_with_users

def test_user_posts(db_with_posts):
    alice = db_with_posts.query(User).filter_by(name="Alice").first()
    assert len(alice.posts) == 2

Example 26: MongoDB Fixture

import pytest
from pymongo import MongoClient

@pytest.fixture(scope="session")
def mongo_client():
    """Create MongoDB client for test session."""
    client = MongoClient("localhost", 27017)
    yield client
    client.close()

@pytest.fixture
def mongo_db(mongo_client):
    """Provide clean database for each test."""
    db = mongo_client.test_database
    yield db
    # Cleanup: drop all collections
    for collection_name in db.list_collection_names():
        db.drop_collection(collection_name)

@pytest.fixture
def users_collection(mongo_db):
    """Provide users collection with sample data."""
    collection = mongo_db.users
    collection.insert_many([
        {"name": "Alice", "age": 30},
        {"name": "Bob", "age": 25},
        {"name": "Charlie", "age": 35},
    ])
    return collection

def test_find_users(users_collection):
    users = list(users_collection.find({"age": {"$gte": 30}}))
    assert len(users) == 2
    assert users[0]["name"] in ["Alice", "Charlie"]

API Testing Fixtures

Example 27: REST API Client

import pytest
import requests

@pytest.fixture(scope="session")
def api_base_url():
    """Base URL for API."""
    return "https://api.example.com"

@pytest.fixture
def api_client(api_base_url):
    """HTTP client for API testing."""
    class APIClient:
        def __init__(self, base_url):
            self.base_url = base_url
            self.session = requests.Session()
            self.token = None

        def authenticate(self, username, password):
            response = self.session.post(
                f"{self.base_url}/auth/login",
                json={"username": username, "password": password}
            )
            self.token = response.json()["access_token"]
            self.session.headers["Authorization"] = f"Bearer {self.token}"

        def get(self, path, **kwargs):
            return self.session.get(f"{self.base_url}{path}", **kwargs)

        def post(self, path, **kwargs):
            return self.session.post(f"{self.base_url}{path}", **kwargs)

        def close(self):
            self.session.close()

    client = APIClient(api_base_url)
    yield client
    client.close()

def test_api_get_users(api_client):
    api_client.authenticate("test_user", "test_pass")
    response = api_client.get("/users")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

Example 28: Mock API Responses

import pytest
from unittest.mock import Mock, patch

@pytest.fixture
def mock_api_success():
    """Mock successful API response."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {
        "id": 1,
        "name": "Test User",
        "email": "test@example.com"
    }
    return mock_response

@pytest.fixture
def mock_api_error():
    """Mock API error response."""
    mock_response = Mock()
    mock_response.status_code = 404
    mock_response.json.return_value = {
        "error": "Not found"
    }
    mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found")
    return mock_response

def test_successful_api_call(mock_api_success):
    with patch('requests.get', return_value=mock_api_success):
        response = requests.get("https://api.example.com/users/1")
        assert response.status_code == 200
        assert response.json()["name"] == "Test User"

def test_api_error_handling(mock_api_error):
    with patch('requests.get', return_value=mock_api_error):
        response = requests.get("https://api.example.com/users/999")
        assert response.status_code == 404
        assert "error" in response.json()

Example 29: GraphQL Client Fixture

import pytest
from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport

@pytest.fixture(scope="session")
def graphql_client():
    """GraphQL client for testing."""
    transport = RequestsHTTPTransport(
        url="https://api.example.com/graphql",
        headers={"Authorization": "Bearer test-token"}
    )
    client = Client(transport=transport, fetch_schema_from_transport=True)
    yield client

@pytest.fixture
def sample_graphql_query():
    """Sample GraphQL query."""
    return gql("""
        query GetUser($id: ID!) {
            user(id: $id) {
                id
                name
                email
            }
        }
    """)

def test_graphql_query(graphql_client, sample_graphql_query):
    result = graphql_client.execute(
        sample_graphql_query,
        variable_values={"id": "1"}
    )
    assert result["user"]["name"] is not None

File and Directory Fixtures

Example 30: Temporary Directory with Files

import pytest
from pathlib import Path

@pytest.fixture
def temp_workspace(tmp_path):
    """Create temporary workspace with directory structure."""
    # Create directory structure
    (tmp_path / "src").mkdir()
    (tmp_path / "tests").mkdir()
    (tmp_path / "docs").mkdir()

    # Create some files
    (tmp_path / "README.md").write_text("# Test Project")
    (tmp_path / "src" / "main.py").write_text("print('Hello')")
    (tmp_path / "tests" / "test_main.py").write_text("def test(): pass")

    return tmp_path

def test_workspace_structure(temp_workspace):
    assert (temp_workspace / "src").is_dir()
    assert (temp_workspace / "tests").is_dir()
    assert (temp_workspace / "README.md").is_file()

def test_can_create_new_files(temp_workspace):
    new_file = temp_workspace / "config.json"
    new_file.write_text('{"debug": true}')
    assert new_file.exists()

Example 31: Sample Data Files

import pytest
import json
import csv

@pytest.fixture
def json_data_file(tmp_path):
    """Create JSON data file."""
    data_file = tmp_path / "data.json"
    data = {
        "users": [
            {"id": 1, "name": "Alice", "age": 30},
            {"id": 2, "name": "Bob", "age": 25},
        ],
        "posts": [
            {"id": 1, "user_id": 1, "title": "First Post"},
        ]
    }
    data_file.write_text(json.dumps(data, indent=2))
    return data_file

@pytest.fixture
def csv_data_file(tmp_path):
    """Create CSV data file."""
    csv_file = tmp_path / "data.csv"
    with csv_file.open('w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(["id", "name", "age"])
        writer.writerow([1, "Alice", 30])
        writer.writerow([2, "Bob", 25])
        writer.writerow([3, "Charlie", 35])
    return csv_file

def test_read_json(json_data_file):
    with json_data_file.open() as f:
        data = json.load(f)
    assert len(data["users"]) == 2
    assert data["users"][0]["name"] == "Alice"

def test_read_csv(csv_data_file):
    with csv_data_file.open() as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    assert len(rows) == 3
    assert rows[0]["name"] == "Alice"

Example 32: Configuration File Fixture

import pytest
import configparser

@pytest.fixture
def config_file(tmp_path):
    """Create configuration file."""
    config = configparser.ConfigParser()
    config['database'] = {
        'host': 'localhost',
        'port': '5432',
        'name': 'testdb'
    }
    config['api'] = {
        'url': 'https://api.example.com',
        'timeout': '30'
    }

    config_path = tmp_path / "config.ini"
    with config_path.open('w') as f:
        config.write(f)

    return config_path

def test_read_config(config_file):
    config = configparser.ConfigParser()
    config.read(config_file)

    assert config['database']['host'] == 'localhost'
    assert config['api']['timeout'] == '30'

Mocking Fixtures

Example 33: Mock External Service

import pytest
from unittest.mock import Mock, patch

@pytest.fixture
def mock_email_service():
    """Mock email service."""
    service = Mock()
    service.send_email.return_value = {
        "status": "sent",
        "message_id": "msg-12345"
    }
    return service

@pytest.fixture
def mock_payment_gateway():
    """Mock payment gateway."""
    gateway = Mock()
    gateway.process_payment.return_value = {
        "success": True,
        "transaction_id": "tx-67890",
        "amount": 99.99
    }
    return gateway

def test_send_email(mock_email_service):
    result = mock_email_service.send_email(
        to="user@example.com",
        subject="Test",
        body="Test message"
    )

    assert result["status"] == "sent"
    mock_email_service.send_email.assert_called_once()

def test_process_payment(mock_payment_gateway):
    result = mock_payment_gateway.process_payment(
        amount=99.99,
        currency="USD",
        card_number="4111111111111111"
    )

    assert result["success"] is True
    assert result["transaction_id"] is not None

Example 34: Mock with Side Effects

import pytest
from unittest.mock import Mock

@pytest.fixture
def mock_api_with_retries():
    """Mock API that fails then succeeds."""
    api = Mock()

    # First two calls fail, third succeeds
    api.fetch_data.side_effect = [
        ConnectionError("Network timeout"),
        ConnectionError("Network timeout"),
        {"status": "success", "data": [1, 2, 3]}
    ]

    return api

def test_api_retry_logic(mock_api_with_retries):
    # Implement retry logic
    max_retries = 3
    for attempt in range(max_retries):
        try:
            result = mock_api_with_retries.fetch_data()
            break
        except ConnectionError:
            if attempt == max_retries - 1:
                raise

    assert result["status"] == "success"
    assert mock_api_with_retries.fetch_data.call_count == 3

Example 35: Monkeypatch Fixture

import pytest
import os

@pytest.fixture
def mock_environment(monkeypatch):
    """Set up mock environment variables."""
    monkeypatch.setenv("DATABASE_URL", "postgresql://test:test@localhost/testdb")
    monkeypatch.setenv("API_KEY", "test-api-key-12345")
    monkeypatch.setenv("DEBUG", "true")

    return monkeypatch

def test_environment_config(mock_environment):
    assert os.getenv("DATABASE_URL").startswith("postgresql://")
    assert os.getenv("API_KEY") == "test-api-key-12345"
    assert os.getenv("DEBUG") == "true"

@pytest.fixture
def mock_datetime(monkeypatch):
    """Mock datetime.now()."""
    from datetime import datetime

    class MockDatetime:
        @staticmethod
        def now():
            return datetime(2025, 1, 1, 12, 0, 0)

    monkeypatch.setattr("datetime.datetime", MockDatetime)

def test_with_fixed_datetime(mock_datetime):
    from datetime import datetime
    now = datetime.now()
    assert now.year == 2025
    assert now.month == 1
    assert now.day == 1

Complex Fixture Patterns

Example 36: Fixture Composition

import pytest

@pytest.fixture
def smtp_config():
    return {"host": "smtp.example.com", "port": 587}

@pytest.fixture
def email_templates():
    return {
        "welcome": "Welcome {name}!",
        "reset": "Reset your password: {link}"
    }

@pytest.fixture
def email_service(smtp_config, email_templates):
    """Compose email service from multiple fixtures."""
    class EmailService:
        def __init__(self, config, templates):
            self.config = config
            self.templates = templates

        def send_welcome(self, name):
            return self.templates["welcome"].format(name=name)

        def send_reset(self, link):
            return self.templates["reset"].format(link=link)

    return EmailService(smtp_config, email_templates)

def test_email_service(email_service):
    welcome = email_service.send_welcome("Alice")
    assert "Alice" in welcome

    reset = email_service.send_reset("https://example.com/reset")
    assert "https://example.com/reset" in reset

Example 37: Context Manager Fixture

import pytest
from contextlib import contextmanager

@pytest.fixture
def transaction_manager():
    """Fixture that returns a context manager."""
    @contextmanager
    def transaction():
        print("\nBegin transaction")
        try:
            yield
            print("\nCommit transaction")
        except Exception:
            print("\nRollback transaction")
            raise

    return transaction

def test_successful_transaction(transaction_manager):
    with transaction_manager():
        # Perform operations
        print("\nExecuting operations...")
        assert True
    # Transaction committed

def test_failed_transaction(transaction_manager):
    with pytest.raises(ValueError):
        with transaction_manager():
            print("\nExecuting operations...")
            raise ValueError("Something went wrong")
    # Transaction rolled back

Example 38: Dynamic Fixture Selection

import pytest

@pytest.fixture
def get_storage(request):
    """Return different storage based on marker."""
    if "s3" in request.keywords:
        return {"type": "s3", "bucket": "test-bucket"}
    elif "local" in request.keywords:
        return {"type": "local", "path": "/tmp/storage"}
    else:
        return {"type": "memory", "data": {}}

@pytest.mark.s3
def test_s3_storage(get_storage):
    assert get_storage["type"] == "s3"
    assert "bucket" in get_storage

@pytest.mark.local
def test_local_storage(get_storage):
    assert get_storage["type"] == "local"
    assert "path" in get_storage

def test_default_storage(get_storage):
    assert get_storage["type"] == "memory"

Example 39: Fixture with Request Parameter

import pytest

@pytest.fixture
def user(request):
    """Flexible user fixture based on test needs."""
    # Get parameters from test marker if available
    marker = request.node.get_closest_marker("user_config")
    if marker:
        config = marker.kwargs
    else:
        config = {}

    return {
        "name": config.get("name", "Default User"),
        "role": config.get("role", "user"),
        "permissions": config.get("permissions", [])
    }

def test_default_user(user):
    assert user["name"] == "Default User"
    assert user["role"] == "user"

@pytest.mark.user_config(name="Admin", role="admin", permissions=["read", "write", "delete"])
def test_admin_user(user):
    assert user["name"] == "Admin"
    assert user["role"] == "admin"
    assert "delete" in user["permissions"]

Example 40: Nested Fixtures with Cleanup

import pytest

@pytest.fixture
def outer_resource():
    """Outer fixture with cleanup."""
    print("\nSetup outer resource")
    resource = {"outer": True, "data": []}

    yield resource

    print("\nCleanup outer resource")
    resource["data"].clear()

@pytest.fixture
def middle_resource(outer_resource):
    """Middle fixture depending on outer."""
    print("\nSetup middle resource")
    outer_resource["middle"] = True

    yield outer_resource

    print("\nCleanup middle resource")
    del outer_resource["middle"]

@pytest.fixture
def inner_resource(middle_resource):
    """Inner fixture depending on middle."""
    print("\nSetup inner resource")
    middle_resource["inner"] = True

    yield middle_resource

    print("\nCleanup inner resource")
    del middle_resource["inner"]

def test_with_nested_fixtures(inner_resource):
    """Test showing fixture setup/cleanup order."""
    assert inner_resource["outer"] is True
    assert inner_resource["middle"] is True
    assert inner_resource["inner"] is True

# Execution order:
# Setup outer -> Setup middle -> Setup inner
# Test runs
# Cleanup inner -> Cleanup middle -> Cleanup outer

---

Version: 1.0.0 Last Updated: October 2025

These examples demonstrate the full power and flexibility of pytest fixtures. Use them as templates for your own test suites.

pytest-patterns

A comprehensive guide to Python testing with pytest, covering everything from basic testing to advanced patterns and CI/CD integration.

Overview

pytest is the de facto standard for testing Python applications. This skill provides comprehensive patterns, examples, and best practices for writing effective tests using pytest.

What You'll Learn

• Fixtures: Modular test setup and teardown patterns
• Parametrization: Testing multiple inputs efficiently
• Mocking: Isolating code from external dependencies
• Test Organization: Structuring large test suites
• Coverage: Measuring and improving test coverage
• CI/CD: Integrating tests into continuous integration pipelines

Quick Start

Installation

# Basic installation
pip install pytest

# With coverage support
pip install pytest pytest-cov

# With async support
pip install pytest pytest-asyncio

# Full test environment
pip install pytest pytest-cov pytest-xdist pytest-mock pytest-timeout

Your First Test

# test_example.py
def test_simple_addition():
    assert 2 + 2 == 4

def test_string_operations():
    text = "hello world"
    assert text.upper() == "HELLO WORLD"
    assert "hello" in text
    assert len(text) == 11

Run it:

pytest test_example.py

Core Features

Simple and Powerful Syntax

pytest uses plain Python assert statements - no special assertion methods needed:

# Clear and readable
assert result == expected_value
assert user.is_active
assert len(items) > 0
assert "error" not in response

Automatic Test Discovery

pytest automatically finds your tests:

project/
├── src/
│   └── mypackage/
│       └── calculator.py
└── tests/
    ├── test_calculator.py      # ✓ Found
    ├── test_utils.py           # ✓ Found
    └── calculator_test.py      # ✓ Found

Detailed Failure Messages

When tests fail, pytest shows you exactly what went wrong:

def test_user_age():
    user = User(name="Alice", age=25)
    assert user.age == 30

# Output:
# AssertionError: assert 25 == 30
#  +  where 25 = User(name='Alice', age=25).age

Fixtures - Reusable Test Setup

Fixtures are pytest's most powerful feature for test setup:

import pytest

@pytest.fixture
def user():
    """Create a test user."""
    return User(name="Test User", email="test@example.com")

@pytest.fixture
def database():
    """Create and teardown test database."""
    db = Database()
    db.connect()
    yield db
    db.disconnect()

def test_user_creation(user, database):
    """Use both fixtures in a test."""
    database.save(user)
    assert database.count() == 1

Fixture Scopes

Control how often fixtures are created:

@pytest.fixture(scope="session")    # Once per test session
def database_engine():
    return create_engine("postgresql://test")

@pytest.fixture(scope="module")     # Once per test module
def api_client():
    return APIClient()

@pytest.fixture(scope="function")   # Once per test (default)
def temp_file():
    return create_temp_file()

Parametrization - Test Multiple Cases

Run the same test with different inputs:

import pytest

@pytest.mark.parametrize("input_value,expected", [
    ("hello", "HELLO"),
    ("world", "WORLD"),
    ("python", "PYTHON"),
])
def test_uppercase(input_value, expected):
    assert input_value.upper() == expected

# Runs 3 tests:
# test_uppercase[hello-HELLO]
# test_uppercase[world-WORLD]
# test_uppercase[python-PYTHON]

Multiple Parameters

@pytest.mark.parametrize("x", [1, 2])
@pytest.mark.parametrize("y", [10, 20])
def test_addition(x, y):
    assert x + y > 0

# Runs 4 tests: (1,10), (1,20), (2,10), (2,20)

Mocking - Isolate Your Tests

Use monkeypatch for safe, automatic cleanup:

def test_environment_variable(monkeypatch):
    monkeypatch.setenv("API_KEY", "test-key")
    assert os.getenv("API_KEY") == "test-key"
    # Automatically restored after test

def test_api_call(monkeypatch):
    def mock_get(*args, **kwargs):
        return MockResponse({"status": "ok"})

    monkeypatch.setattr(requests, "get", mock_get)
    result = fetch_data()
    assert result["status"] == "ok"

Test Organization

Recommended Structure

project/
├── src/
│   └── mypackage/
│       ├── __init__.py
│       ├── models.py
│       └── services.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py           # Shared fixtures
│   ├── unit/
│   │   ├── __init__.py
│   │   ├── test_models.py
│   │   └── test_services.py
│   ├── integration/
│   │   └── test_api.py
│   └── e2e/
│       └── test_workflows.py
├── pytest.ini                # Configuration
└── requirements-dev.txt

Using conftest.py

Share fixtures across tests:

# tests/conftest.py
import pytest

@pytest.fixture(scope="session")
def database():
    """Available to all tests."""
    db = Database()
    db.connect()
    yield db
    db.disconnect()

@pytest.fixture
def clean_database(database):
    """Reset database before each test."""
    database.clear()
    return database

Markers - Organize and Filter Tests

Mark tests for selective execution:

import pytest

@pytest.mark.slow
def test_long_running_operation():
    time.sleep(5)
    assert True

@pytest.mark.integration
def test_external_api():
    response = requests.get("https://api.example.com")
    assert response.status_code == 200

@pytest.mark.skip(reason="Feature not implemented")
def test_future_feature():
    pass

@pytest.mark.xfail(reason="Known bug")
def test_with_known_issue():
    assert buggy_function() == expected_value

Run specific tests:

pytest -m slow                    # Run only slow tests
pytest -m "not slow"              # Skip slow tests
pytest -m "integration and not slow"  # Complex filtering

Coverage - Measure Test Effectiveness

Generate Coverage Reports

# Terminal report
pytest --cov=mypackage tests/

# HTML report
pytest --cov=mypackage --cov-report=html tests/

# Fail if coverage below threshold
pytest --cov=mypackage --cov-fail-under=80 tests/

Coverage Output

---------- coverage: platform darwin, python 3.11.0 -----------
Name                      Stmts   Miss  Cover
---------------------------------------------
mypackage/__init__.py         4      0   100%
mypackage/models.py          42      2    95%
mypackage/services.py        38      5    87%
mypackage/utils.py           15      1    93%
---------------------------------------------
TOTAL                        99      8    92%

Configuration

# pytest.ini
[pytest]
addopts =
    --cov=mypackage
    --cov-report=html
    --cov-report=term-missing
    --cov-fail-under=80

Common Testing Patterns

Testing Exceptions

import pytest

def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        result = 10 / 0

def test_validation_error():
    with pytest.raises(ValueError, match="Invalid email"):
        validate_email("not-an-email")

Testing Logs

def test_logging(caplog):
    import logging
    logger = logging.getLogger("myapp")

    logger.info("Processing started")
    logger.warning("High memory usage")

    assert "Processing started" in caplog.text
    assert any(record.levelname == "WARNING" for record in caplog.records)

Testing Output

def test_print_output(capsys):
    print("Hello, World!")
    captured = capsys.readouterr()
    assert "Hello, World!" in captured.out

Testing with Temporary Files

def test_file_processing(tmp_path):
    # tmp_path is a pytest fixture
    test_file = tmp_path / "data.txt"
    test_file.write_text("test content")

    result = process_file(test_file)
    assert result.success

Running Tests

Basic Commands

# Run all tests
pytest

# Run specific file
pytest tests/test_models.py

# Run specific test
pytest tests/test_models.py::test_user_creation

# Run tests matching pattern
pytest -k "user and not delete"

# Stop on first failure
pytest -x

# Show local variables on failure
pytest -l

# Verbose output
pytest -v

# Quiet output
pytest -q

Advanced Options

# Run in parallel (requires pytest-xdist)
pytest -n auto

# Rerun failed tests
pytest --lf

# Run tests that failed in last run
pytest --ff

# Show slowest tests
pytest --durations=10

# Drop into debugger on failure
pytest --pdb

# Show print statements
pytest -s

CI/CD Integration

GitHub Actions

name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ['3.8', '3.9', '3.10', '3.11']

    steps:
    - uses: actions/checkout@v3
    - uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}
    - run: pip install -e .[dev]
    - run: pytest --cov --cov-report=xml
    - uses: codecov/codecov-action@v3

GitLab CI

test:
  image: python:3.11
  script:
    - pip install -e .[dev]
    - pytest --cov --cov-report=xml
  coverage: '/(?i)total.*? (100(?:\.0+)?\%|[1-9]?\d(?:\.\d+)?\%)$/'

Configuration

pytest.ini

[pytest]
# Test discovery
testpaths = tests
python_files = test_*.py *_test.py
python_classes = Test*
python_functions = test_*

# Output
addopts =
    -ra
    --strict-markers
    --strict-config
    --showlocals

# Markers
markers =
    slow: marks tests as slow
    integration: marks tests as integration tests
    unit: marks tests as unit tests

pyproject.toml

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = ["-ra", "--strict-markers", "--cov=mypackage"]
markers = [
    "slow: marks tests as slow",
    "integration: marks tests as integration tests",
]

Useful Plugins

Essential pytest plugins:

• pytest-cov: Coverage reporting
• pytest-xdist: Parallel test execution
• pytest-asyncio: Async/await support
• pytest-mock: Enhanced mocking
• pytest-timeout: Test timeouts
• pytest-randomly: Randomize test order
• pytest-html: HTML test reports

Install them:

pip install pytest-cov pytest-xdist pytest-asyncio pytest-mock

Best Practices

1. Write Isolated Tests

Each test should be independent and not rely on other tests:

# Bad - tests depend on order
def test_create_user():
    global user
    user = create_user("Alice")

def test_update_user():
    user.update(age=30)  # Depends on previous test

# Good - each test is independent
def test_create_user():
    user = create_user("Alice")
    assert user.name == "Alice"

def test_update_user():
    user = create_user("Bob")
    user.update(age=30)
    assert user.age == 30

2. Use Descriptive Names

Make test names explain what they test:

# Bad
def test_user():
    pass

# Good
def test_user_creation_with_valid_email():
    pass

def test_user_login_fails_with_wrong_password():
    pass

3. Follow AAA Pattern

Arrange, Act, Assert:

def test_shopping_cart_total():
    # Arrange
    cart = ShoppingCart()
    cart.add_item(Product(name="Book", price=10.99))
    cart.add_item(Product(name="Pen", price=2.50))

    # Act
    total = cart.get_total()

    # Assert
    assert total == 13.49

4. Don't Test Implementation Details

Test behavior, not implementation:

# Bad - testing internal implementation
def test_user_password_hash():
    user = User(password="secret")
    assert user._password_hash.startswith("$2b$")

# Good - testing behavior
def test_user_can_login_with_correct_password():
    user = User(password="secret")
    assert user.verify_password("secret") is True
    assert user.verify_password("wrong") is False

5. Keep Tests Fast

• Use in-memory databases for tests
• Mock external services
• Use appropriate fixture scopes
• Run slow tests separately with markers

Troubleshooting

Tests Not Found

# See what pytest discovers
pytest --collect-only

# Ensure file naming is correct
test_*.py or *_test.py

# Check function naming
test_*

Import Errors

# Install package in development mode
pip install -e .

# Or add to PYTHONPATH
export PYTHONPATH="${PYTHONPATH}:$(pwd)/src"

Fixture Not Found

# List available fixtures
pytest --fixtures

# Check fixture is in conftest.py or same file
# Verify scope is appropriate

Learning Path

1. Beginner: Basic tests, simple fixtures, parametrization 2. Intermediate: Complex fixtures, mocking, test organization 3. Advanced: Custom plugins, advanced parametrization, property-based testing 4. Expert: Performance optimization, distributed testing, custom reporters

Resources

Official Documentation

• pytest docs: https://docs.pytest.org/
• pytest GitHub: https://github.com/pytest-dev/pytest
• Plugin list: https://docs.pytest.org/en/latest/reference/plugin_list.html

Tutorials

• Real Python pytest guide: https://realpython.com/pytest-python-testing/
• pytest with Eric: https://testandcode.com/
• Full pytest documentation: https://docs.pytest.org/en/stable/contents.html

Books

• Python Testing with pytest by Brian Okken
• Test-Driven Development with Python by Harry Percival

Community

• pytest Discord: https://discord.com/invite/pytest-dev
• Stack Overflow: https://stackoverflow.com/questions/tagged/pytest

Quick Reference

Common Commands

pytest                          # Run all tests
pytest test_file.py            # Run specific file
pytest -k "pattern"            # Run tests matching pattern
pytest -m marker               # Run tests with marker
pytest -x                      # Stop on first failure
pytest --lf                    # Rerun last failed
pytest -v                      # Verbose
pytest -q                      # Quiet
pytest --cov=pkg              # Coverage report
pytest -n auto                # Parallel execution

Common Fixtures

tmp_path        # Temporary directory (pathlib.Path)
tmp_path_factory # Factory for temporary directories
capsys          # Capture stdout/stderr
caplog          # Capture log messages
monkeypatch     # Modify objects/environment
request         # Request object for fixtures

Common Markers

@pytest.mark.parametrize    # Run with multiple inputs
@pytest.mark.skip           # Skip test
@pytest.mark.skipif         # Conditional skip
@pytest.mark.xfail          # Expected to fail
@pytest.mark.slow           # Custom marker

---

Version: 1.0.0 Last Updated: October 2025 License: MIT