testing-integration-testing

Test interactions between multiple components or services

Integration Testing

When to Use This Skill

Use this skill when you need to:

ACTIVATE THIS SKILL: When testing component interactions, databases, APIs, or external services

Core Concepts

Integration Tests vs Unit Tests

Unit Tests:

Integration Tests:

Test Database Strategies

Strategy 1: In-Memory Database (SQLite)

# Python (pytest)
import pytest
from sqlalchemy import create_engine
from myapp.models import Base

@pytest.fixture(scope="function")
def db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()

def test_create_user(db_session):
    user = User(name="Alice", email="alice@example.com")
    db_session.add(user)
    db_session.commit()

    retrieved = db_session.query(User).filter_by(email="alice@example.com").first()
    assert retrieved.name == "Alice"

Strategy 2: Test Containers (Docker)

# Python (testcontainers)
from testcontainers.postgres import PostgresContainer
import pytest

@pytest.fixture(scope="module")
def postgres_container():
    with PostgresContainer("postgres:16") as postgres:
        yield postgres

@pytest.fixture
def db_connection(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    Base.metadata.create_all(engine)
    connection = engine.connect()
    yield connection
    connection.close()

def test_user_repository(db_connection):
    repo = UserRepository(db_connection)
    user = repo.create(name="Alice", email="alice@example.com")
    assert repo.find_by_id(user.id).name == "Alice"
// TypeScript (testcontainers)
import { PostgreSqlContainer } from '@testcontainers/postgresql';
import { DataSource } from 'typeorm';

describe('UserRepository', () => {
  let container: PostgreSqlContainer;
  let dataSource: DataSource;

  beforeAll(async () => {
    container = await new PostgreSqlContainer('postgres:16').start();
    dataSource = new DataSource({
      type: 'postgres',
      host: container.getHost(),
      port: container.getPort(),
      username: container.getUsername(),
      password: container.getPassword(),
      database: container.getDatabase(),
      entities: [User],
      synchronize: true,
    });
    await dataSource.initialize();
  });

  afterAll(async () => {
    await dataSource.destroy();
    await container.stop();
  });

  it('creates and retrieves user', async () => {
    const repo = dataSource.getRepository(User);
    const user = await repo.save({ name: 'Alice', email: 'alice@example.com' });
    const retrieved = await repo.findOneBy({ id: user.id });
    expect(retrieved.name).toBe('Alice');
  });
});

Strategy 3: Transaction Rollback

# Python (pytest)
@pytest.fixture
def db_session():
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

# Each test gets clean state via rollback
def test_create_user(db_session):
    user = User(name="Alice")
    db_session.add(user)
    db_session.commit()
    assert db_session.query(User).count() == 1
// Go
func TestUserRepository(t *testing.T) {
    db := setupTestDB(t)
    tx := db.Begin()
    defer tx.Rollback()

    repo := NewUserRepository(tx)
    user, err := repo.Create("Alice", "alice@example.com")
    require.NoError(t, err)

    retrieved, err := repo.FindByID(user.ID)
    require.NoError(t, err)
    assert.Equal(t, "Alice", retrieved.Name)
}

Patterns

API Integration Testing

# Python (FastAPI with TestClient)
from fastapi.testclient import TestClient
from myapp.main import app

client = TestClient(app)

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

    # Act
    response = client.post("/users", json=payload)

    # Assert
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"
    assert "id" in data

def test_get_user_endpoint(db_session):
    # Arrange: Create user in DB
    user = User(name="Alice", email="alice@example.com")
    db_session.add(user)
    db_session.commit()

    # Act
    response = client.get(f"/users/{user.id}")

    # Assert
    assert response.status_code == 200
    assert response.json()["name"] == "Alice"

def test_authentication_required():
    response = client.get("/protected-resource")
    assert response.status_code == 401

    response = client.get(
        "/protected-resource",
        headers={"Authorization": "Bearer valid-token"}
    )
    assert response.status_code == 200
// TypeScript (Express + Supertest)
import request from 'supertest';
import { app } from './app';
import { setupTestDB, teardownTestDB } from './test-helpers';

describe('User API', () => {
  beforeAll(async () => {
    await setupTestDB();
  });

  afterAll(async () => {
    await teardownTestDB();
  });

  it('creates user via POST /users', async () => {
    const response = await request(app)
      .post('/users')
      .send({ name: 'Alice', email: 'alice@example.com' })
      .expect(201);

    expect(response.body.name).toBe('Alice');
    expect(response.body.id).toBeDefined();
  });

  it('retrieves user via GET /users/:id', async () => {
    const createRes = await request(app)
      .post('/users')
      .send({ name: 'Bob', email: 'bob@example.com' });

    const userId = createRes.body.id;

    const response = await request(app)
      .get(`/users/${userId}`)
      .expect(200);

    expect(response.body.name).toBe('Bob');
  });
});

Testing Database Migrations

# Python (Alembic)
import pytest
from alembic import command
from alembic.config import Config

def test_migrations_run_successfully():
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", "sqlite:///:memory:")

    # Run all migrations
    command.upgrade(config, "head")

    # Verify schema exists
    engine = create_engine("sqlite:///:memory:")
    inspector = inspect(engine)
    tables = inspector.get_table_names()
    assert "users" in tables
    assert "posts" in tables

def test_migration_rollback():
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", "sqlite:///:memory:")

    # Upgrade to latest
    command.upgrade(config, "head")

    # Downgrade one version
    command.downgrade(config, "-1")

    # Verify rollback worked
    # ... assertions ...

Testing with Redis/Cache

# Python (fakeredis)
import pytest
from fakeredis import FakeRedis

@pytest.fixture
def redis_client():
    return FakeRedis()

def test_cache_set_get(redis_client):
    cache = CacheService(redis_client)
    cache.set("key", "value", ttl=60)
    assert cache.get("key") == "value"

def test_cache_expiration(redis_client):
    cache = CacheService(redis_client)
    cache.set("key", "value", ttl=0)
    time.sleep(0.1)
    assert cache.get("key") is None
// TypeScript (ioredis-mock)
import RedisMock from 'ioredis-mock';

describe('CacheService', () => {
  let redis: RedisMock;
  let cache: CacheService;

  beforeEach(() => {
    redis = new RedisMock();
    cache = new CacheService(redis);
  });

  it('sets and gets values', async () => {
    await cache.set('key', 'value', 60);
    const result = await cache.get('key');
    expect(result).toBe('value');
  });
});

Testing Message Queues

# Python (testing with in-memory queue)
import pytest
from queue import Queue

class InMemoryQueue:
    def __init__(self):
        self.queue = Queue()

    def publish(self, message):
        self.queue.put(message)

    def consume(self, timeout=1):
        return self.queue.get(timeout=timeout)

@pytest.fixture
def message_queue():
    return InMemoryQueue()

def test_order_processing(message_queue):
    # Arrange
    order_service = OrderService(message_queue)
    payment_processor = PaymentProcessor(message_queue)

    # Act
    order_service.create_order(user_id=1, items=["item1"])

    # Assert: Message published
    message = message_queue.consume(timeout=1)
    assert message["type"] == "order_created"
    assert message["user_id"] == 1

    # Act: Process payment
    payment_processor.process(message)

    # Assert: Payment message published
    payment_message = message_queue.consume(timeout=1)
    assert payment_message["type"] == "payment_processed"

Testing External Service Integrations

# Python (using requests-mock)
import pytest
import requests_mock

def test_weather_api_integration():
    with requests_mock.Mocker() as m:
        # Mock external API
        m.get(
            "https://api.weather.com/forecast",
            json={"temperature": 72, "condition": "sunny"}
        )

        # Test service that calls API
        service = WeatherService()
        forecast = service.get_forecast(zip_code="94102")

        assert forecast.temperature == 72
        assert forecast.condition == "sunny"

def test_api_timeout_handling():
    with requests_mock.Mocker() as m:
        m.get("https://api.weather.com/forecast", exc=requests.Timeout)

        service = WeatherService()
        with pytest.raises(WeatherServiceError, match="timeout"):
            service.get_forecast(zip_code="94102")
// TypeScript (MSW - Mock Service Worker)
import { rest } from 'msw';
import { setupServer } from 'msw/node';

const server = setupServer(
  rest.get('https://api.weather.com/forecast', (req, res, ctx) => {
    return res(ctx.json({ temperature: 72, condition: 'sunny' }));
  })
);

beforeAll(() => server.listen());
afterEach(() => server.resetHandlers());
afterAll(() => server.close());

describe('WeatherService', () => {
  it('fetches forecast from API', async () => {
    const service = new WeatherService();
    const forecast = await service.getForecast('94102');

    expect(forecast.temperature).toBe(72);
    expect(forecast.condition).toBe('sunny');
  });

  it('handles API errors', async () => {
    server.use(
      rest.get('https://api.weather.com/forecast', (req, res, ctx) => {
        return res(ctx.status(500));
      })
    );

    const service = new WeatherService();
    await expect(service.getForecast('94102')).rejects.toThrow();
  });
});

Examples by Language

Python (pytest + SQLAlchemy)

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:16") as container:
        yield container

@pytest.fixture(scope="session")
def engine(postgres):
    engine = create_engine(postgres.get_connection_url())
    Base.metadata.create_all(engine)
    return engine

@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

# test_user_repository.py
def test_create_and_find_user(db_session):
    repo = UserRepository(db_session)

    user = repo.create(name="Alice", email="alice@example.com")
    assert user.id is not None

    found = repo.find_by_email("alice@example.com")
    assert found.name == "Alice"

def test_user_unique_email_constraint(db_session):
    repo = UserRepository(db_session)

    repo.create(name="Alice", email="alice@example.com")

    with pytest.raises(IntegrityError):
        repo.create(name="Bob", email="alice@example.com")

Go (testcontainers-go)

// integration_test.go
package repository_test

import (
    "context"
    "testing"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/wait"
)

func setupPostgres(t *testing.T) *sql.DB {
    ctx := context.Background()

    req := testcontainers.ContainerRequest{
        Image:        "postgres:16",
        ExposedPorts: []string{"5432/tcp"},
        Env: map[string]string{
            "POSTGRES_PASSWORD": "test",
            "POSTGRES_DB":       "testdb",
        },
        WaitingFor: wait.ForLog("database system is ready"),
    }

    container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
        ContainerRequest: req,
        Started:          true,
    })
    require.NoError(t, err)

    t.Cleanup(func() {
        container.Terminate(ctx)
    })

    host, _ := container.Host(ctx)
    port, _ := container.MappedPort(ctx, "5432")

    dsn := fmt.Sprintf("host=%s port=%s user=postgres password=test dbname=testdb sslmode=disable",
        host, port.Port())

    db, err := sql.Open("postgres", dsn)
    require.NoError(t, err)

    return db
}

func TestUserRepository(t *testing.T) {
    db := setupPostgres(t)
    defer db.Close()

    repo := NewUserRepository(db)

    user, err := repo.Create("Alice", "alice@example.com")
    require.NoError(t, err)
    assert.NotZero(t, user.ID)

    found, err := repo.FindByEmail("alice@example.com")
    require.NoError(t, err)
    assert.Equal(t, "Alice", found.Name)
}

TypeScript (Vitest + TypeORM)

// test-setup.ts
import { DataSource } from 'typeorm';
import { PostgreSqlContainer } from '@testcontainers/postgresql';

export async function setupTestDB(): Promise<DataSource> {
  const container = await new PostgreSqlContainer('postgres:16').start();

  const dataSource = new DataSource({
    type: 'postgres',
    host: container.getHost(),
    port: container.getPort(),
    username: container.getUsername(),
    password: container.getPassword(),
    database: container.getDatabase(),
    entities: [User, Post],
    synchronize: true,
  });

  await dataSource.initialize();

  return dataSource;
}

// user-repository.test.ts
describe('UserRepository', () => {
  let dataSource: DataSource;
  let repository: Repository<User>;

  beforeAll(async () => {
    dataSource = await setupTestDB();
    repository = dataSource.getRepository(User);
  });

  afterAll(async () => {
    await dataSource.destroy();
  });

  afterEach(async () => {
    await repository.clear();
  });

  it('creates and retrieves user', async () => {
    const user = await repository.save({
      name: 'Alice',
      email: 'alice@example.com'
    });

    const found = await repository.findOneBy({ email: 'alice@example.com' });
    expect(found?.name).toBe('Alice');
  });
});

Checklist

Before Writing Integration Tests:

Setting Up Infrastructure:

Writing Tests:

Performance:

After Writing Tests:

Anti-Patterns

❌ NEVER: Share state between tests without cleanup
   → Flaky tests, order dependencies

❌ NEVER: Use production database for tests
   → Data corruption, slow tests

❌ NEVER: Skip transaction rollback in teardown
   → Polluted database state

❌ NEVER: Test too many layers at once
   → Slow, hard to debug failures

❌ NEVER: Ignore test performance
   → Slow feedback loop, developers skip tests

❌ NEVER: Mock everything in integration tests
   → Defeats purpose of integration testing

❌ NEVER: Leave containers running after tests
   → Resource leaks, port conflicts

Level 3: Resources

Comprehensive Reference

See /resources/REFERENCE.md for in-depth coverage:

Executable Scripts

/resources/scripts/run_integration_tests.sh Orchestrates integration tests with automatic setup/teardown:

/resources/scripts/analyze_test_coverage.py Analyzes integration test coverage across multiple dimensions:

/resources/scripts/generate_test_report.py Generates comprehensive HTML/JSON reports from test results:

Runnable Examples

Python Examples (/resources/examples/python/):

TypeScript Examples (/resources/examples/typescript/):

Docker Infrastructure (/resources/examples/docker/):

All scripts are production-ready with:

Related Skills

Foundation:

Database Testing:

API Testing:

Infrastructure:

Tools: