Integration Testing - Testing Real System Interactions

Integration tests occupy the middle ground between unit tests and end-to-end tests, verifying that multiple components work together correctly. I’ve learned that the secret to effective integration testing isn’t avoiding external dependencies—it’s controlling them predictably.

The challenge with integration tests is balancing realism with reliability. You want to test real interactions, but you also need tests that run consistently across different environments and don’t break when external services have issues.

Database Integration Testing

Database integration tests verify that your data access layer works correctly with real database operations. The key is using a test database that mirrors your production schema but remains isolated from other tests:

import pytest
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker
from src.myapp.models import User, Base
from src.myapp.repositories import UserRepository

@pytest.fixture(scope="session")
def test_engine():
    """Create a test database engine."""
    engine = sa.create_engine("postgresql://test:test@localhost/test_db")
    Base.metadata.create_all(engine)
    yield engine
    Base.metadata.drop_all(engine)

@pytest.fixture
def db_session(test_engine):
    """Provide a clean database session for each test."""
    Session = sessionmaker(bind=test_engine)
    session = Session()
    
    yield session
    
    session.rollback()
    session.close()

def test_user_repository_integration(db_session):
    """Test user repository with real database operations."""
    repo = UserRepository(db_session)
    
    # Create a user
    user = User(username="alice", email="[email protected]")
    saved_user = repo.save(user)
    
    assert saved_user.id is not None
    
    # Retrieve the user
    retrieved = repo.get_by_username("alice")
    assert retrieved.email == "[email protected]"
    
    # Update the user
    retrieved.email = "[email protected]"
    repo.save(retrieved)
    
    # Verify the update
    updated = repo.get_by_id(saved_user.id)
    assert updated.email == "[email protected]"

This test verifies that your repository correctly handles database transactions, relationships, and constraints without mocking the database layer.

API Integration Testing

When testing APIs, you want to verify that your endpoints handle real HTTP requests correctly while controlling the underlying dependencies:

import pytest
from fastapi.testclient import TestClient
from src.myapp.main import create_app
from src.myapp.database import get_db_session

@pytest.fixture
def test_app(db_session):
    """Create test application with test database."""
    app = create_app()
    
    # Override the database dependency
    def override_get_db():
        yield db_session
    
    app.dependency_overrides[get_db_session] = override_get_db
    return app

@pytest.fixture
def client(test_app):
    """Create test client for making HTTP requests."""
    return TestClient(test_app)

def test_user_api_workflow(client, db_session):
    """Test complete user API workflow."""
    # Create a user
    response = client.post("/users", json={
        "username": "bob",
        "email": "[email protected]",
        "password": "secure_password"
    })
    
    assert response.status_code == 201
    user_data = response.json()
    user_id = user_data["id"]
    
    # Retrieve the user
    response = client.get(f"/users/{user_id}")
    assert response.status_code == 200
    
    retrieved_user = response.json()
    assert retrieved_user["username"] == "bob"
    assert "password" not in retrieved_user  # Ensure password not exposed
    
    # Update the user
    response = client.put(f"/users/{user_id}", json={
        "email": "[email protected]"
    })
    assert response.status_code == 200
    
    # Verify the update
    response = client.get(f"/users/{user_id}")
    updated_user = response.json()
    assert updated_user["email"] == "[email protected]"

This integration test verifies the entire HTTP request/response cycle while using a controlled database environment.

Testing External Service Integration

When your application integrates with external services, create integration tests that use real service calls but in a controlled environment:

import pytest
import requests
from src.myapp.services import PaymentService

@pytest.mark.integration
@pytest.mark.skipif(not os.getenv("STRIPE_TEST_KEY"), 
                   reason="Stripe test key not configured")
def test_payment_service_integration():
    """Test payment processing with Stripe test environment."""
    service = PaymentService(api_key=os.getenv("STRIPE_TEST_KEY"))
    
    # Use Stripe's test card numbers
    payment_data = {
        "amount": 2000,  # $20.00
        "currency": "usd",
        "card_number": "4242424242424242",  # Test card
        "exp_month": 12,
        "exp_year": 2025,
        "cvc": "123"
    }
    
    result = service.process_payment(payment_data)
    
    assert result["status"] == "succeeded"
    assert result["amount"] == 2000
    assert "charge_id" in result
    
    # Verify we can retrieve the charge
    charge = service.get_charge(result["charge_id"])
    assert charge["amount"] == 2000

This test uses Stripe’s test environment to verify real API integration without affecting production data or incurring charges.

Container-Based Integration Testing

For complex integration scenarios, use containers to create reproducible test environments:

import pytest
import docker
import time
from src.myapp.cache import RedisCache

@pytest.fixture(scope="session")
def redis_container():
    """Start Redis container for integration tests."""
    client = docker.from_env()
    
    container = client.containers.run(
        "redis:6-alpine",
        ports={"6379/tcp": None},  # Random host port
        detach=True,
        remove=True
    )
    
    # Wait for Redis to be ready
    port = container.attrs["NetworkSettings"]["Ports"]["6379/tcp"][0]["HostPort"]
    redis_url = f"redis://localhost:{port}"
    
    # Wait for service to be ready
    for _ in range(30):
        try:
            import redis
            r = redis.from_url(redis_url)
            r.ping()
            break
        except:
            time.sleep(0.1)
    
    yield redis_url
    
    container.stop()

def test_redis_cache_integration(redis_container):
    """Test cache operations with real Redis instance."""
    cache = RedisCache(redis_container)
    
    # Test basic operations
    cache.set("test_key", "test_value", ttl=60)
    assert cache.get("test_key") == "test_value"
    
    # Test expiration
    cache.set("expire_key", "value", ttl=1)
    time.sleep(1.1)
    assert cache.get("expire_key") is None
    
    # Test complex data
    data = {"user_id": 123, "preferences": ["dark_mode", "notifications"]}
    cache.set("user_data", data)
    retrieved = cache.get("user_data")
    assert retrieved == data

This approach provides a real Redis instance for testing while ensuring complete isolation and cleanup.

Testing Message Queues and Async Operations

Integration tests for asynchronous systems require special handling to ensure operations complete before assertions:

import pytest
import asyncio
from src.myapp.queue import TaskQueue
from src.myapp.workers import EmailWorker

@pytest.fixture
async def task_queue():
    """Provide in-memory task queue for testing."""
    queue = TaskQueue("memory://")
    await queue.connect()
    yield queue
    await queue.disconnect()

@pytest.mark.asyncio
async def test_email_worker_integration(task_queue):
    """Test email processing workflow."""
    worker = EmailWorker(task_queue)
    
    # Queue an email task
    task_id = await task_queue.enqueue("send_email", {
        "to": "[email protected]",
        "subject": "Test Email",
        "body": "This is a test email"
    })
    
    # Process the task
    result = await worker.process_next_task()
    
    assert result["task_id"] == task_id
    assert result["status"] == "completed"
    
    # Verify task is removed from queue
    pending_tasks = await task_queue.get_pending_count()
    assert pending_tasks == 0

This test verifies the complete message queue workflow while using an in-memory queue for speed and reliability.

Integration Test Organization

Organize integration tests separately from unit tests to enable different execution strategies:

tests/
├── unit/
│   ├── test_models.py
│   └── test_services.py
├── integration/
│   ├── test_database.py
│   ├── test_api.py
│   └── test_external_services.py
└── conftest.py

Use pytest markers to run different test categories:

# Run only unit tests (fast)
pytest tests/unit -m "not integration"

# Run integration tests (slower)
pytest tests/integration -m integration

# Run all tests
pytest

This organization lets developers run fast unit tests during development while ensuring integration tests run in CI pipelines.

In our next part, we’ll explore debugging techniques that help you understand what’s happening when tests fail or when your application behaves unexpectedly. We’ll cover Python’s debugging tools, logging strategies, and techniques for diagnosing complex issues in both development and production environments.