| name | pytest-patterns |
| description | Python testing with pytest covering fixtures, parametrization, mocking, and test organization for reliable test suites |
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
assertstatements 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_*.pyor*_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_*.pyor*_test.py - Check function naming:
test_* - Verify
__init__.pyfiles exist in test directories - Run with
-vflag to see discovery process
Fixtures Not Found
- Check fixture is in
conftest.pyor same file - Verify fixture scope is appropriate
- Check for typos in fixture name
- Use
--fixturesflag to list available fixtures
Test Failures
- Use
-vfor verbose output - Use
--tb=longfor detailed tracebacks - Use
--pdbto drop into debugger on failure - Use
-xto stop on first failure - Use
--lfto rerun last failed tests
Import Errors
- Ensure package is installed:
pip install -e . - Check PYTHONPATH is set correctly
- Verify
__init__.pyfiles exist - Use
sys.pathmanipulation 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+