Pytest Advanced

Advanced Pytest Patterns

Advanced pytest features for robust, maintainable test suites.

When to Use This Skill

Installation

# Core + common plugins
uv add --dev pytest pytest-cov pytest-asyncio pytest-xdist pytest-mock

Configuration (pyproject.toml)

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = [
    "-v",
    "--strict-markers",
    "--tb=short",
    "-ra",
    "--cov=src",
    "--cov-report=term-missing",
    "--cov-fail-under=80",
]
markers = [
    "slow: marks tests as slow (deselect with '-m \"not slow\"')",
    "integration: marks tests as integration tests",
]
asyncio_mode = "auto"

[tool.coverage.run]
branch = true
source = ["src"]

[tool.coverage.report]
exclude_lines = ["pragma: no cover", "if TYPE_CHECKING:", "@abstractmethod"]

Fixtures

Scopes and Lifecycle

import pytest
from typing import Generator

# function (default) - fresh per test
@pytest.fixture
def db() -> Generator[Database, None, None]:
    database = Database(":memory:")
    database.create_tables()
    yield database
    database.close()

# session - shared across all tests
@pytest.fixture(scope="session")
def app():
    return create_app("testing")

# autouse - applies automatically
@pytest.fixture(autouse=True)
def reset_state():
    clear_cache()
    yield

Parametrized Fixtures

@pytest.fixture(params=["sqlite", "postgres", "mysql"])
def database_backend(request) -> str:
    return request.param  # Test runs 3 times

# Indirect parametrization
@pytest.fixture
def user(request) -> User:
    return User(**request.param)

@pytest.mark.parametrize("user", [
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25},
], indirect=True)
def test_user_validation(user: User):
    assert user.name

Factory Pattern

@pytest.fixture
def user_factory() -> Callable[[str], User]:
    created: list[User] = []
    def _create(name: str, **kwargs) -> User:
        user = User(name=name, **kwargs)
        created.append(user)
        return user
    yield _create
    for u in created:
        u.delete()

Markers

# Built-in markers
@pytest.mark.skip(reason="Not implemented")
@pytest.mark.skipif(sys.version_info < (3, 12), reason="Requires 3.12+")
@pytest.mark.xfail(reason="Known bug #123")
@pytest.mark.timeout(10)

# Parametrize
@pytest.mark.parametrize("input,expected", [
    pytest.param(2, 4, id="two"),
    pytest.param(3, 9, id="three"),
    pytest.param(-2, 4, id="negative"),
])
def test_square(input: int, expected: int):
    assert input ** 2 == expected

# Run by marker
pytest -m unit                    # Only unit tests
pytest -m "not slow"              # Skip slow tests
pytest -m "integration and not slow"  # Combine markers

Key Plugins

Async Testing (pytest-asyncio)

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

@pytest.fixture
async def async_client() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient() as client:
        yield client

Mocking (pytest-mock)

def test_with_mock(mocker):
    mock_api = mocker.patch("myapp.external.api_call")
    mock_api.return_value = {"data": "test"}
    result = my_function()
    assert result["data"] == "test"
    mock_api.assert_called_once()

Running Tests

# Execution
pytest                           # All tests
pytest tests/test_models.py::test_user  # Specific test
pytest -k "user and not slow"    # Pattern matching

# Parallel
pytest -n auto                   # All CPUs
pytest -n 4                      # 4 workers

# Coverage
pytest --cov=src --cov-report=html --cov-report=term-missing

# Failed tests
pytest --lf                      # Last failed only
pytest --ff                      # Failed first
pytest -x                        # Stop on first failure
pytest --maxfail=3               # Stop after 3

# Debugging
pytest -x --pdb                  # Debug on failure
pytest -s                        # Show print output
pytest --collect-only            # Dry run

CI Integration

# .github/workflows/test.yml
- name: Run tests
  run: |
    uv run pytest \
      --cov=src \
      --cov-report=xml \
      --cov-report=term-missing \
      --junitxml=test-results.xml

Agentic Optimizations

For detailed patterns on conftest.py hierarchy, async testing, test organization, and common patterns, see REFERENCE.md.

pytest Advanced Reference

Detailed patterns for conftest.py organization, async testing, test structure, and common testing recipes.

conftest.py Patterns

Project Structure

tests/
├── conftest.py              # Root (session-level fixtures)
├── unit/
│   ├── conftest.py          # Unit test fixtures
│   └── test_models.py
├── integration/
│   ├── conftest.py          # Integration test fixtures
│   └── test_api.py
└── e2e/
    ├── conftest.py          # E2E test fixtures
    └── test_workflows.py

Root conftest.py

import pytest
from myapp import create_app

@pytest.fixture(scope="session")
def app():
    return create_app("testing")

@pytest.fixture(scope="session")
def db_engine():
    engine = create_test_engine()
    yield engine
    engine.dispose()

def pytest_configure(config):
    config.addinivalue_line("markers", "slow: slow tests")
    config.addinivalue_line("markers", "integration: integration tests")

def pytest_collection_modifyitems(config, items):
    """Automatically mark tests based on path."""
    for item in items:
        if "integration" in item.nodeid:
            item.add_marker(pytest.mark.integration)

@pytest.fixture(autouse=True)
def reset_database(db_engine):
    clear_tables(db_engine)
    yield
    rollback_transaction(db_engine)

Domain-Specific conftest.py

# tests/integration/conftest.py
@pytest.fixture
def authenticated_client(app) -> Client:
    client = app.test_client()
    client.login("test@example.com", "password")
    return client

@pytest.fixture
def sample_data(db):
    db.load_fixtures("integration_data.json")
    yield
    db.clear_fixtures()

Async Testing Patterns

Basic Async Tests

import pytest
import asyncio

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

@pytest.mark.asyncio
async def test_concurrent_operations():
    results = await asyncio.gather(
        fetch_user(1), fetch_user(2), fetch_user(3)
    )
    assert len(results) == 3

Async Fixtures

from typing import AsyncGenerator

@pytest.fixture
async def async_client() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient() as client:
        yield client

@pytest.fixture(scope="module")
async def async_db() -> AsyncGenerator[AsyncDatabase, None]:
    db = AsyncDatabase("test.db")
    await db.connect()
    yield db
    await db.close()

Event Loop Management

@pytest.fixture(scope="session")
def event_loop():
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    loop.close()

Testing Async Context Managers

@pytest.mark.asyncio
async def test_async_context_manager():
    async with AsyncResource() as resource:
        result = await resource.process()
        assert result is not None
    assert resource.is_closed()

Testing Async Generators

@pytest.mark.asyncio
async def test_async_generator():
    results = []
    async for item in async_generator():
        results.append(item)
    assert len(results) == expected_count

Test Organization Best Practices

AAA Pattern

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

    # Act
    user = create_user(user_data)

    # Assert
    assert user.name == "Alice"
    assert user.id is not None

Naming Conventions

# Pattern: test_<function>_<scenario>_<expected_result>
def test_divide_by_zero_raises_error(): ...
def test_user_login_with_valid_credentials_succeeds(): ...
def test_api_call_with_invalid_token_returns_401(): ...

Test Grouping with Classes

class TestUserAuthentication:
    def test_login_success(self, user):
        assert login(user.email, user.password).success

    def test_login_wrong_password(self, user):
        assert not login(user.email, "wrong").success

    def test_logout(self, authenticated_user):
        logout(authenticated_user)
        assert not is_authenticated(authenticated_user)

Complex Parametrization

@pytest.mark.parametrize("user_data,should_succeed", [
    ({"name": "Alice", "email": "alice@example.com"}, True),
    ({"name": "", "email": "alice@example.com"}, False),
    ({"name": "Bob", "email": "invalid"}, False),
    ({"name": "Charlie"}, False),
])
def test_user_validation(user_data: dict, should_succeed: bool):
    if should_succeed:
        assert create_user(user_data) is not None
    else:
        with pytest.raises(ValidationError):
            create_user(user_data)

Test Data Management

@pytest.fixture
def users_data():
    with open("tests/fixtures/users.json") as f:
        return json.load(f)["users"]

@pytest.fixture
def sample_users(db, users_data):
    users = [db.create_user(data) for data in users_data]
    yield users
    for user in users:
        db.delete_user(user.id)

Common Patterns

Testing Exceptions

# Assert exception is raised
def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

# Assert exception message
def test_invalid_input():
    with pytest.raises(ValueError, match="must be positive"):
        process(-1)

# Capture for inspection
def test_custom_exception():
    with pytest.raises(CustomError) as exc_info:
        trigger_error()
    assert exc_info.value.code == 42

Testing Warnings

def test_deprecated_function():
    with pytest.warns(DeprecationWarning, match="deprecated"):
        deprecated_function()

Mocking and Patching

def test_external_api_call(mocker):
    mock_response = Mock()
    mock_response.json.return_value = {"data": "test"}
    mocker.patch("requests.get", return_value=mock_response)
    result = fetch_data_from_api()
    assert result["data"] == "test"

def test_database_interaction(mocker):
    mock_db = mocker.patch("myapp.database.Database")
    mock_db.return_value.query.return_value = [{"id": 1}]
    result = get_users()
    assert len(result) == 1
    mock_db.return_value.query.assert_called_once()

Temporary Files

def test_file_processing(tmp_path: Path):
    test_file = tmp_path / "test.txt"
    test_file.write_text("test content")
    result = process_file(test_file)
    assert result.success

def test_config_file(tmp_path: Path):
    config_file = tmp_path / "config.yaml"
    config_file.write_text("setting: value")
    app = create_app(config_file)
    assert app.config["setting"] == "value"

Plugin Details

pytest-cov (Coverage)

pytest --cov=src --cov-report=html --cov-report=term-missing
pytest --cov=src --cov-fail-under=80  # Fail if below threshold

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

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

pytest-xdist (Parallel)

pytest -n auto          # All CPUs
pytest -n 4             # 4 workers
pytest --dist loadfile  # Distribute by file
pytest --dist loadscope # Distribute by scope

pytest-timeout

@pytest.mark.timeout(10)
def test_fast(): ...

@pytest.mark.timeout(0)
def test_no_timeout(): ...

[tool.pytest.ini_options]
timeout = 300
timeout_method = "thread"

pytest-benchmark

def test_performance(benchmark):
    result = benchmark(function_to_test, arg1, arg2)
    assert result == expected

pytest --benchmark-compare=0001  # Compare to baseline
pytest --benchmark-save=baseline  # Save as baseline