python-testing-advanced

name	python-testing-advanced
description	Advanced Python testing — async testing with pytest-asyncio, exception/side-effect testing, test organization, common patterns (API, database, class methods), pytest configuration, and CLI reference. Extends python-testing.

Python Testing — Advanced Patterns

This skill extends python-testing with async testing, advanced mocking, test organization, common patterns, and configuration.

When to Activate

• Testing async code (FastAPI, aiohttp, asyncio)
• Structuring large test suites
• Setting up pytest configuration for projects
• Testing database layers, API endpoints, or class methods in depth
• Mocking async functions and async context managers with AsyncMock
• Writing integration tests for FastAPI endpoints using httpx.AsyncClient
• Configuring pytest.ini or pyproject.toml with coverage thresholds and custom markers
• Organizing tests into unit, integration, and e2e directories with appropriate shared conftest fixtures

Testing Async Code

pytest-asyncio Setup

pip install pytest-asyncio

# pytest.ini or pyproject.toml
[pytest]
asyncio_mode = auto  # or "strict" for explicit marking

Basic Async Tests

import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_function():
    """Test async function."""
    result = await async_add(2, 3)
    assert result == 5

@pytest.mark.asyncio
async def test_async_with_timeout():
    """Test with timeout."""
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(
            slow_operation(),
            timeout=0.1
        )

Async Fixtures

@pytest.fixture
async def async_client():
    """Async fixture for HTTP client."""
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client

@pytest.fixture
async def async_db():
    """Async database fixture."""
    async with AsyncSessionLocal() as session:
        yield session
        await session.rollback()

@pytest.mark.asyncio
async def test_async_endpoint(async_client):
    """Test with async client fixture."""
    response = await async_client.get("/users/1")
    assert response.status_code == 200

Mocking Async Functions

from unittest.mock import AsyncMock, patch

@patch("mypackage.fetch_data", new_callable=AsyncMock)
async def test_with_async_mock(mock_fetch):
    """Test with mocked async function."""
    mock_fetch.return_value = {"data": "result"}

    result = await process_data()

    mock_fetch.assert_called_once()
    assert result["data"] == "result"

@pytest.mark.asyncio
async def test_async_context_manager():
    """Test with async context manager mock."""
    mock_session = AsyncMock()
    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
    mock_session.__aexit__ = AsyncMock(return_value=None)
    mock_session.query.return_value = [{"id": 1}]

    with patch("mypackage.get_session", return_value=mock_session):
        result = await fetch_users()
        assert len(result) == 1

Testing FastAPI Endpoints

from fastapi.testclient import TestClient
import pytest

@pytest.fixture
def client():
    """FastAPI test client."""
    return TestClient(app)

def test_create_item(client):
    """Test POST endpoint."""
    response = client.post("/items/", json={"name": "Widget", "price": 9.99})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Widget"

def test_get_item_not_found(client):
    """Test 404 response."""
    response = client.get("/items/99999")
    assert response.status_code == 404

# Async FastAPI with httpx
@pytest.fixture
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

@pytest.mark.asyncio
async def test_async_create_item(async_client):
    response = await async_client.post("/items/", json={"name": "Widget"})
     response.status_code == 

Exception and Side-Effect Testing

Testing Exception Details

def test_exception_message():
    """Test specific exception message."""
    with pytest.raises(ValueError) as exc_info:
        validate_age(-1)

    assert "Age must be positive" in str(exc_info.value)
    assert exc_info.value.args[0] == "Age must be positive, got -1"

def test_exception_type_hierarchy():
    """Test exception type hierarchy."""
    with pytest.raises(DatabaseError) as exc_info:
        query_database("INVALID SQL")

    # Check it's the right subtype
    assert isinstance(exc_info.value, ConnectionError)
    assert exc_info.value.code == 500

Testing Side Effects

def test_function_calls_side_effect(mocker):
    """Test that side effects happen (using pytest-mock)."""
    mock_email = mocker.patch("mypackage.send_email")
    mock_cache = mocker.patch("mypackage.cache.invalidate")

    create_user(email="[e-mail]")

    mock_email.assert_called_once_with(
        to="[e-mail]",
        subject="Welcome"
    )
    mock_cache.assert_called_once_with("users")

def test_database_writes(database):
    """Test database side effects."""
    create_user(name="Alice", db=database)

    user = database.query(User).filter(User.name == "Alice").first()
    assert user is not None
    assert user.name == "Alice"

Testing File I/O Side Effects

def test_writes_output_file(tmp_path):
    """Test file writing."""
    output_file = tmp_path / "output.csv"

    export_to_csv(data=[{"id": 1, "name": "Alice"}], path=output_file)

    assert output_file.exists()
    content = output_file.read_text()
    assert "Alice" in content
    assert "id,name" in content  # headers

def test_reads_config_file(tmp_path):
    """Test file reading."""
    config_file = tmp_path / "config.json"
    config_file.write_text('{"debug": true, "port": 8080}')

    config = load_config(config_file)

    assert config.debug is True
    assert config.port == 8080

Test Organization

Directory Structure

tests/
├── conftest.py              # Shared fixtures for all tests
├── unit/
│   ├── conftest.py          # Unit test fixtures
│   ├── test_models.py
│   ├── test_services.py
│   └── test_utils.py
├── integration/
│   ├── conftest.py          # Integration fixtures (DB, HTTP)
│   ├── test_api.py
│   └── test_database.py
└── e2e/
    ├── conftest.py
    └── test_user_flows.py

Test Classes

class TestUserService:
    """Tests for UserService class."""

    @pytest.fixture(autouse=True)
    def setup(self, database):
        """Run before each test in this class."""
        self.service = UserService(db=database)
        self.db = database

    def test_create_user(self):
        user = self.service.create(name="Alice", email="[e-mail]")
        assert user.id is not None
        assert user.name == "Alice"

    def test_create_user_duplicate_email(self):
        self.service.create(name="Alice", email="[e-mail]")
        with pytest.raises(DuplicateEmailError):
            self.service.create(name="Bob", email="[e-mail]")

    def test_get_user(self):
        created = self.service.create(name="Alice", email="[e-mail]")
        fetched = self.service.get(id=created.id)
        assert fetched.name == "Alice"

     ():
         pytest.raises(UserNotFoundError):
            .service.get(=)

Organizing with Markers

# Mark entire class
@pytest.mark.integration
class TestDatabaseIntegration:
    def test_connection(self, database):
        assert database.is_connected()

# Mark individual tests
@pytest.mark.slow
@pytest.mark.integration
def test_bulk_import():
    """Test importing 10000 records."""
    ...

# Conditional skip
@pytest.mark.skipif(
    not os.getenv("REDIS_URL"),
    reason="REDIS_URL not configured"
)
def test_redis_cache():
    ...

Common Pattern: API Endpoint Testing

# tests/integration/test_products_api.py
import pytest
from fastapi.testclient import TestClient
from myapp.main import app

@pytest.fixture(scope="module")
def client():
    return TestClient(app)

@pytest.fixture
def auth_headers(client):
    response = client.post("/auth/login", json={
        "email": "[e-mail]",
        "password": "testpass"
    })
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

class TestProductsEndpoints:
    def test_list_products(self, client):
        response = client.get("/products/")
        assert response.status_code == 200
        assert isinstance(response.json()["data"], list)

    def test_create_product_authenticated(self, client, auth_headers):
        payload = {"name": "Widget", "price": 9.99, "stock": 100}
        response = client.post("/products/", json=payload, headers=auth_headers)
         response.status_code == 
        product = response.json()
         product[] == 

     ():
        response = client.post(, json={: })
         response.status_code == 

     ():
        payload = {: , : -}
        response = client.post(, json=payload, headers=auth_headers)
         response.status_code == 

     ():
        
        create_response = client.post(
            ,
            json={: , : },
            headers=auth_headers
        )
        product_id = create_response.json()[]

        
        delete_response = client.delete(
            ,
            headers=auth_headers
        )
         delete_response.status_code == 

        
        get_response = client.get()
         get_response.status_code == 

Common Pattern: Database Testing

# tests/integration/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from myapp.database import Base

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

@pytest.fixture
def db_session(engine):
    """Provide transactional database session."""
    Session = sessionmaker(bind=engine)
    session = Session()
    session.begin_nested()  # Savepoint for rollback

    yield session

    session.rollback()
    session.close()

# tests/integration/test_user_repository.py
class TestUserRepository:
    def test_save_and_retrieve(self, db_session):
        repo = UserRepository(db_session)
        user = User(name="Alice", email="[e-mail]")

        saved = repo.save(user)
        retrieved = repo.find_by_id(saved.id)

        assert retrieved.name == "Alice"

    def test_find_by_email(self, db_session):
        repo = UserRepository(db_session)
        repo.save(User(name="Alice", email="[e-mail]"))

        user = repo.find_by_email("[e-mail]")

         user   
         user.name == 

     ():
        repo = UserRepository(db_session)
        repo.save(User(name=, email=))
        repo.save(User(name=, email=))

        users = repo.find_all()
         (users) == 

pytest Configuration

pytest.ini

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

# Output
addopts =
    --strict-markers
    --tb=short
    -v
    --cov=mypackage
    --cov-report=term-missing
    --cov-fail-under=80

# Custom markers
markers =
    slow: marks tests as slow (deselect with '-m "not slow"')
    integration: requires external services
    unit: pure unit tests

# asyncio
asyncio_mode = auto

pyproject.toml

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = [
    "--strict-markers",
    "--tb=short",
    "-v",
    "--cov=mypackage",
    "--cov-report=term-missing",
    "--cov-fail-under=80",
]
markers = [
    "slow: marks tests as slow",
    "integration: requires external services",
    "unit: pure unit tests",
]
asyncio_mode = "auto"

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

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

CLI Quick Reference

# Run all tests
pytest

# Run with coverage
pytest --cov=mypackage --cov-report=term-missing

# Run specific test file
pytest tests/test_users.py

# Run specific test
pytest tests/test_users.py::TestUserService::test_create_user

# Run by marker
pytest -m unit
pytest -m "not slow"
pytest -m "integration and not slow"

# Run failed tests from last run
pytest --lf

# Run tests that contain keyword
pytest -k "user and not delete"

# Show locals on failure
pytest -l

# Stop after first failure
pytest -x

# Stop after N failures
pytest --maxfail=3

# Verbose output
pytest -v

# Show captured output even for passing tests
pytest -s

# Parallel execution (requires pytest-xdist)
pytest -n auto

# Generate HTML report
pytest --html=report.html --self-contained-html

# Update snapshots (requires pytest-snapshot)
pytest --snapshot-update

Quick Reference

Feature	Usage
Async test	@pytest.mark.asyncio async def test_...
Async fixture	@pytest.fixture async def ...
Async mock	AsyncMock from unittest.mock
File I/O	tmp_path builtin fixture
Test class	class TestFoo: with @pytest.fixture(autouse=True) def setup
Skip condition	@pytest.mark.skipif(condition, reason=...)
Expected fail	@pytest.mark.xfail
Parametrize IDs	ids=[...] parameter on @pytest.mark.parametrize
Coverage	pytest --cov=pkg --cov-fail-under=80
Parallel	pytest -n auto (pytest-xdist)