Article 7: Advanced Testing Strategies

Introduction

Unit tests are the foundation, but real applications require more sophisticated testing strategies. Through building and maintaining production systems, I've learned that mastering mocking, fixtures, and integration testing is essential for testing code that interacts with databases, APIs, and external services.

This article covers advanced pytest techniques including mocking, test doubles, integration testing, and test-driven development (TDD).

The Testing Pyramid

Test Type	Speed	Scope	Purpose
Unit	Fast	Single function/class	Test logic in isolation
Integration	Medium	Multiple components	Test components work together
End-to-End	Slow	Entire system	Test real user scenarios

Mocking with pytest-mock

Why Mock?

Mocking replaces real dependencies with controlled test doubles:

Basic Mocking

# src/weather_service.py
import requests


def get_weather(city: str) -> dict:
    """Fetch weather from external API."""
    response = requests.get(
        f"https://api.weather.com/v1/{city}",
        timeout=10,
    )
    response.raise_for_status()
    return response.json()


def get_weather_summary(city: str) -> str:
    """Get a human-readable weather summary."""
    data = get_weather(city)
    return f"{city}: {data['temperature']}°C, {data['condition']}"

# tests/test_weather_service.py
from unittest.mock import Mock, patch
import pytest


def test_get_weather_summary(mocker):
    """Test weather summary with mocked API."""
    # Arrange: Mock the external API call
    mock_response = Mock()
    mock_response.json.return_value = {
        "temperature": 22,
        "condition": "Sunny",
    }
    mock_response.raise_for_status = Mock()
    
    mocker.patch("weather_service.requests.get", return_value=mock_response)
    
    # Act
    result = get_weather_summary("London")
    
    # Assert
    assert result == "London: 22°C, Sunny"


def test_get_weather_api_error(mocker):
    """Test handling of API errors."""
    import requests
    
    mocker.patch(
        "weather_service.requests.get",
        side_effect=requests.RequestException("API unavailable"),
    )
    
    with pytest.raises(requests.RequestException):
        get_weather("London")

Mocking Patterns

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


# Mock with return value
mock_db = Mock()
mock_db.find_user.return_value = User(id=1, name="Alice")

# Mock with side effects (multiple calls)
mock_counter = Mock()
mock_counter.next.side_effect = [1, 2, 3]
assert mock_counter.next() == 1
assert mock_counter.next() == 2

# Mock raising exceptions
mock_api = Mock()
mock_api.call.side_effect = ConnectionError("Network failed")

# Verify calls
mock_service = Mock()
mock_service.process("data")
mock_service.process.assert_called_once_with("data")

# Verify multiple calls
mock_logger = Mock()
mock_logger.log("info", "Starting")
mock_logger.log("info", "Done")
mock_logger.log.assert_has_calls([
    call("info", "Starting"),
    call("info", "Done"),
])

Patching

from unittest.mock import patch


# Context manager
def test_with_patch():
    with patch("module.external_function") as mock_func:
        mock_func.return_value = "mocked"
        result = function_under_test()
        assert result == "mocked"


# Decorator
@patch("module.ExternalClass")
def test_with_decorator(mock_class):
    mock_instance = mock_class.return_value
    mock_instance.method.return_value = "mocked"
    
    result = function_under_test()
    
    assert result == "mocked"


# Multiple patches
@patch("module.function_a")
@patch("module.function_b")
def test_multiple_patches(mock_b, mock_a):  # Note: reverse order!
    mock_a.return_value = "a"
    mock_b.return_value = "b"
    
    result = function_under_test()
    
    assert "a" in result and "b" in result


# pytest-mock style
def test_with_mocker(mocker):
    mock_func = mocker.patch("module.external_function")
    mock_func.return_value = "mocked"
    
    result = function_under_test()
    
    assert result == "mocked"

Where to Patch

# src/services/user_service.py
from repositories.user_repository import UserRepository

class UserService:
    def __init__(self):
        self.repo = UserRepository()  # Import used here
    
    def get_user(self, user_id: int):
        return self.repo.find_by_id(user_id)

# tests/test_user_service.py

# WRONG: Patching where it's defined
@patch("repositories.user_repository.UserRepository")
def test_wrong(mock_repo):
    # This won't work as expected!
    pass


# CORRECT: Patching where it's used
@patch("services.user_service.UserRepository")
def test_correct(mock_repo_class):
    mock_instance = mock_repo_class.return_value
    mock_instance.find_by_id.return_value = User(id=1, name="Alice")
    
    service = UserService()
    user = service.get_user(1)
    
    assert user.name == "Alice"

Advanced Fixtures

Factory Fixtures

# tests/conftest.py
import pytest
from datetime import date, timedelta


@pytest.fixture
def make_task():
    """Factory fixture for creating tasks with defaults."""
    def _make_task(
        title: str = "Default Task",
        completed: bool = False,
        due_date: date | None = None,
    ) -> Task:
        return Task(
            id=None,
            title=title,
            completed=completed,
            due_date=due_date or date.today() + timedelta(days=7),
        )
    return _make_task


@pytest.fixture
def make_user():
    """Factory fixture for creating users."""
    _id_counter = 0
    
    def _make_user(
        name: str | None = None,
        email: str | None = None,
        is_admin: bool = False,
    ) -> User:
        nonlocal _id_counter
        _id_counter += 1
        return User(
            id=_id_counter,
            name=name or f"User {_id_counter}",
            email=email or f"user{_id_counter}@example.com",
            is_admin=is_admin,
        )
    return _make_user

# tests/test_tasks.py

def test_complete_task(make_task):
    """Test completing a task."""
    task = make_task(title="Learn testing")
    
    task.complete()
    
    assert task.completed is True


def test_overdue_detection(make_task):
    """Test overdue task detection."""
    past_date = date.today() - timedelta(days=1)
    task = make_task(title="Overdue task", due_date=past_date)
    
    assert task.is_overdue is True


def test_user_permissions(make_user):
    """Test user permission checks."""
    admin = make_user(is_admin=True)
    regular = make_user(is_admin=False)
    
    assert admin.can_delete_users is True
    assert regular.can_delete_users is False

Fixture Composition

@pytest.fixture
def database(tmp_path):
    """Create a test database."""
    db_path = tmp_path / "test.db"
    db = Database(str(db_path))
    db.initialize()
    yield db
    db.close()


@pytest.fixture
def user_repository(database):
    """Repository using test database."""
    return UserRepository(database)


@pytest.fixture
def task_repository(database):
    """Repository using test database."""
    return TaskRepository(database)


@pytest.fixture
def task_service(task_repository, user_repository):
    """Service with all dependencies."""
    return TaskService(
        task_repo=task_repository,
        user_repo=user_repository,
    )


# Test uses the composed fixture
def test_assign_task(task_service, make_user, make_task):
    """Test assigning a task to a user."""
    user = make_user(name="Alice")
    task = make_task(title="Review PR")
    
    task_service.assign(task, user)
    
    assert task.assignee_id == user.id

Async Fixtures

import pytest
import asyncio


@pytest.fixture
async def async_client():
    """Async HTTP client fixture."""
    async with aiohttp.ClientSession() as session:
        yield session


@pytest.fixture
async def database():
    """Async database connection."""
    pool = await asyncpg.create_pool(DATABASE_URL)
    yield pool
    await pool.close()


@pytest.mark.asyncio
async def test_fetch_data(async_client):
    """Test async data fetching."""
    async with async_client.get("/api/data") as response:
        data = await response.json()
        assert "results" in data

Integration Testing

Database Integration Tests

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


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


@pytest.fixture
def session(engine):
    """Create a new database session for each test."""
    Session = sessionmaker(bind=engine)
    session = Session()
    
    yield session
    
    session.rollback()
    session.close()


@pytest.fixture
def user_repository(session):
    """Repository with real database session."""
    return SQLUserRepository(session)

# tests/integration/test_user_repository.py
import pytest


@pytest.mark.integration
class TestUserRepository:
    """Integration tests for UserRepository."""
    
    def test_save_and_retrieve(self, user_repository):
        """Test saving and retrieving a user."""
        user = User(name="Alice", email="[email protected]")
        
        saved = user_repository.save(user)
        retrieved = user_repository.find_by_id(saved.id)
        
        assert retrieved is not None
        assert retrieved.name == "Alice"
        assert retrieved.email == "[email protected]"
    
    def test_find_by_email(self, user_repository):
        """Test finding user by email."""
        user = User(name="Bob", email="[email protected]")
        user_repository.save(user)
        
        found = user_repository.find_by_email("[email protected]")
        
        assert found is not None
        assert found.name == "Bob"
    
    def test_update_user(self, user_repository):
        """Test updating user information."""
        user = user_repository.save(User(name="Charlie"))
        
        user.name = "Charles"
        user_repository.save(user)
        
        retrieved = user_repository.find_by_id(user.id)
        assert retrieved.name == "Charles"

API Integration Tests

# tests/integration/test_api.py
import pytest
from fastapi.testclient import TestClient


@pytest.fixture
def client(app, database):
    """Test client with real database."""
    with TestClient(app) as client:
        yield client


@pytest.mark.integration
class TestUserAPI:
    """Integration tests for User API."""
    
    def test_create_user(self, client):
        """Test user creation endpoint."""
        response = client.post(
            "/api/users",
            json={"name": "Alice", "email": "[email protected]"},
        )
        
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Alice"
        assert "id" in data
    
    def test_get_user(self, client):
        """Test get user endpoint."""
        # Create user first
        create_response = client.post(
            "/api/users",
            json={"name": "Bob", "email": "[email protected]"},
        )
        user_id = create_response.json()["id"]
        
        # Get user
        response = client.get(f"/api/users/{user_id}")
        
        assert response.status_code == 200
        assert response.json()["name"] == "Bob"
    
    def test_user_not_found(self, client):
        """Test 404 for missing user."""
        response = client.get("/api/users/99999")
        
        assert response.status_code == 404
        assert "not found" in response.json()["detail"].lower()

External Service Integration

import pytest
import responses


@pytest.fixture
def mock_payment_api():
    """Mock external payment API responses."""
    with responses.RequestsMock() as rsps:
        # Mock successful payment
        rsps.add(
            responses.POST,
            "https://api.payment.com/charge",
            json={"status": "success", "transaction_id": "txn_123"},
            status=200,
        )
        yield rsps


@pytest.mark.integration
def test_process_payment(payment_service, mock_payment_api):
    """Test payment processing with mocked external API."""
    result = payment_service.charge(
        amount=Decimal("99.99"),
        card_token="tok_test",
    )
    
    assert result.success is True
    assert result.transaction_id == "txn_123"


@pytest.mark.integration
def test_payment_failure(payment_service):
    """Test handling of payment failures."""
    with responses.RequestsMock() as rsps:
        rsps.add(
            responses.POST,
            "https://api.payment.com/charge",
            json={"status": "failed", "error": "Card declined"},
            status=402,
        )
        
        result = payment_service.charge(
            amount=Decimal("99.99"),
            card_token="tok_declined",
        )
        
        assert result.success is False
        assert "declined" in result.error

Test-Driven Development (TDD)

The TDD Cycle

TDD Example: Building a Password Validator

Step 1: Red - Write a failing test

# tests/test_password_validator.py

def test_password_minimum_length():
    """Password must be at least 8 characters."""
    validator = PasswordValidator()
    
    assert validator.is_valid("short") is False
    assert validator.is_valid("longenough") is True

Step 2: Green - Make it pass with minimal code

# src/password_validator.py

class PasswordValidator:
    def is_valid(self, password: str) -> bool:
        return len(password) >= 8

Step 3: Red - Add next requirement

def test_password_requires_uppercase():
    """Password must contain at least one uppercase letter."""
    validator = PasswordValidator()
    
    assert validator.is_valid("lowercase") is False
    assert validator.is_valid("Uppercase1") is True

Step 4: Green - Update implementation

class PasswordValidator:
    def is_valid(self, password: str) -> bool:
        if len(password) < 8:
            return False
        if not any(c.isupper() for c in password):
            return False
        return True

Step 5: Refactor - Clean up

class PasswordValidator:
    MIN_LENGTH = 8
    
    def is_valid(self, password: str) -> bool:
        return (
            self._has_minimum_length(password)
            and self._has_uppercase(password)
        )
    
    def _has_minimum_length(self, password: str) -> bool:
        return len(password) >= self.MIN_LENGTH
    
    def _has_uppercase(self, password: str) -> bool:
        return any(c.isupper() for c in password)

Complete TDD Example

# Final test suite after TDD cycles

class TestPasswordValidator:
    """Tests for password validation."""
    
    @pytest.fixture
    def validator(self):
        return PasswordValidator()
    
    @pytest.mark.parametrize("password,expected", [
        ("short", False),
        ("longenough", False),  # No uppercase
        ("ALLUPPERCASE", False),  # No lowercase
        ("NoNumber", False),  # No number
        ("Valid1Password", True),
    ])
    def test_password_validation(self, validator, password, expected):
        """Test various password inputs."""
        assert validator.is_valid(password) == expected
    
    def test_validation_errors(self, validator):
        """Test that errors explain what's wrong."""
        errors = validator.validate("short")
        
        assert "at least 8 characters" in errors
        assert "uppercase letter" in errors
        assert "number" in errors
    
    def test_valid_password_no_errors(self, validator):
        """Test valid password returns no errors."""
        errors = validator.validate("Valid1Password")
        
        assert errors == []

# Final implementation

class PasswordValidator:
    """Validates password strength."""
    
    MIN_LENGTH = 8
    
    def is_valid(self, password: str) -> bool:
        """Check if password meets all requirements."""
        return len(self.validate(password)) == 0
    
    def validate(self, password: str) -> list[str]:
        """Return list of validation errors."""
        errors = []
        
        if len(password) < self.MIN_LENGTH:
            errors.append(f"Password must be at least {self.MIN_LENGTH} characters")
        
        if not any(c.isupper() for c in password):
            errors.append("Password must contain an uppercase letter")
        
        if not any(c.islower() for c in password):
            errors.append("Password must contain a lowercase letter")
        
        if not any(c.isdigit() for c in password):
            errors.append("Password must contain a number")
        
        return errors

Test Doubles

Examples

# Dummy: Just fills a parameter slot
def test_with_dummy():
    dummy_logger = object()  # Never actually used
    service = ServiceThatRequiresLogger(dummy_logger)
    # Test something that doesn't log


# Stub: Returns predetermined values
class StubUserRepository:
    def find_by_id(self, user_id: int) -> User:
        return User(id=user_id, name="Stub User")


# Spy: Records interactions
class SpyEmailSender:
    def __init__(self):
        self.sent_emails = []
    
    def send(self, to: str, subject: str, body: str):
        self.sent_emails.append({"to": to, "subject": subject, "body": body})


def test_user_registration_sends_email():
    spy = SpyEmailSender()
    service = UserService(email_sender=spy)
    
    service.register("[email protected]", "password123")
    
    assert len(spy.sent_emails) == 1
    assert spy.sent_emails[0]["to"] == "[email protected]"
    assert "Welcome" in spy.sent_emails[0]["subject"]


# Fake: Working lightweight implementation
class FakeUserRepository:
    def __init__(self):
        self._users = {}
        self._next_id = 1
    
    def save(self, user: User) -> User:
        if user.id is None:
            user.id = self._next_id
            self._next_id += 1
        self._users[user.id] = user
        return user
    
    def find_by_id(self, user_id: int) -> User | None:
        return self._users.get(user_id)
    
    def find_all(self) -> list[User]:
        return list(self._users.values())

Practical Exercise

Exercise 1: Mock External Dependencies

# Test this function that uses external services

import requests

def get_user_with_weather(user_id: int) -> dict:
    """Get user info with their local weather."""
    # Fetch user from database
    user = user_repository.find_by_id(user_id)
    if not user:
        raise ValueError(f"User {user_id} not found")
    
    # Fetch weather from external API
    response = requests.get(
        f"https://api.weather.com/v1/{user.city}",
        timeout=10,
    )
    weather = response.json()
    
    return {
        "user": user.to_dict(),
        "weather": {
            "temperature": weather["temp"],
            "condition": weather["condition"],
        },
    }

# Solution
def test_get_user_with_weather(mocker):
    # Mock the repository
    mock_repo = mocker.patch("module.user_repository")
    mock_repo.find_by_id.return_value = User(
        id=1,
        name="Alice",
        city="London",
    )
    
    # Mock the API call
    mock_response = mocker.Mock()
    mock_response.json.return_value = {
        "temp": 18,
        "condition": "Cloudy",
    }
    mocker.patch("module.requests.get", return_value=mock_response)
    
    # Call the function
    result = get_user_with_weather(1)
    
    # Assertions
    assert result["user"]["name"] == "Alice"
    assert result["weather"]["temperature"] == 18
    assert result["weather"]["condition"] == "Cloudy"

Exercise 2: TDD a Shopping Cart

Write tests first, then implement:

# Requirements:
# 1. Add items to cart
# 2. Remove items from cart
# 3. Calculate total price
# 4. Apply discount codes
# 5. Clear cart

class TestShoppingCart:
    @pytest.fixture
    def cart(self):
        return ShoppingCart()
    
    @pytest.fixture
    def sample_product(self):
        return Product(id=1, name="Widget", price=Decimal("10.00"))
    
    def test_empty_cart_has_zero_total(self, cart):
        assert cart.total == Decimal("0")
    
    def test_add_item_increases_total(self, cart, sample_product):
        cart.add(sample_product, quantity=2)
        assert cart.total == Decimal("20.00")
    
    # Continue with more tests...

Testing Anti-Patterns

What to Avoid

# ❌ Testing implementation details
def test_internal_cache():
    service = UserService()
    service.get_user(1)
    assert service._cache[1] is not None  # Don't test private members!


# ❌ Tests that depend on order
def test_first():
    global_state.value = 42

def test_second():
    assert global_state.value == 42  # Fails if run alone!


# ❌ Sleeping in tests
def test_async_operation():
    start_async_task()
    time.sleep(5)  # Don't do this!
    assert task_completed()


# ❌ Testing external systems
def test_real_api():
    response = requests.get("https://real-api.com/data")  # Flaky!
    assert response.status_code == 200

Better Approaches

# ✅ Test behavior, not implementation
def test_get_user_returns_cached_result():
    service = UserService(cache_enabled=True)
    
    first_call = service.get_user(1)
    second_call = service.get_user(1)
    
    assert first_call == second_call
    # Don't care HOW it's cached, just that it works


# ✅ Independent tests
@pytest.fixture
def fresh_state():
    state = AppState()
    yield state
    state.cleanup()


# ✅ Use proper async testing
@pytest.mark.asyncio
async def test_async_operation():
    result = await async_function()
    assert result.completed


# ✅ Mock external systems
def test_api_integration(mock_api):
    mock_api.add_response({"data": "test"})
    
    result = fetch_data_from_api()
    
    assert result["data"] == "test"

Key Takeaways

1. Mock external dependencies - Don't let tests depend on networks or databases

2. Use factory fixtures - Create flexible, reusable test data

3. Integration tests complement unit tests - Test that components work together

4. TDD drives better design - Write tests first for cleaner code

5. Choose the right test double - Dummy, stub, spy, mock, or fake

What's Next?

With comprehensive testing skills, let's ensure your code is well-documented. In Article 8: Documentation Best Practices, we'll cover docstrings, README standards, and keeping documentation in sync with code.

---

This article is part of the Software Engineering 101 series.