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:

• Test files: test_*.py or *_test.py

• Test functions: Functions prefixed with test_

• Test classes: Classes prefixed with Test (no __init__ method)

• 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

• One test file per source file: mymodule.py → test_mymodule.py

• Group related tests in classes: Use Test* classes for logical grouping

• Use descriptive test names: test_user_login_with_invalid_credentials

• Keep tests independent: Each test should work in isolation

• Use fixtures for setup: Avoid duplicate setup code

Writing Effective Tests

-

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"

-

Test one thing per test: Each test should verify a single behavior

-

Use descriptive assertions: Make failures easy to understand

-

Avoid test interdependencies: Tests should not depend on execution order

-

Test edge cases: Empty lists, None values, boundary conditions

Fixture Best Practices

• Use appropriate scope: Minimize fixture creation cost

• Keep fixtures small: Each fixture should have a single responsibility

• Use fixture factories: For creating multiple test objects

• Clean up resources: Use yield for teardown

• Share fixtures via conftest.py: Make common fixtures available

Coverage Guidelines

• Aim for high coverage: 80%+ is a good target

• Focus on critical paths: Prioritize important business logic

• Don't chase 100%: Some code doesn't need tests (getters, setters)

• Use coverage to find gaps: Not as a quality metric

• Exclude generated code: Mark with # pragma: no cover

CI/CD Integration

• Run tests on every commit: Catch issues early

• Test on multiple Python versions: Ensure compatibility

• Generate coverage reports: Track coverage trends

• Fail on low coverage: Maintain coverage standards

• 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+