Integration Testing: Best Practices and Patterns

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from myapp.models import Base, User

@pytest.fixture
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()

    saved = db_session.query(User).filter_by(email="alice@example.com").first()
    assert saved is not None
    assert saved.name == "Alice"

import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine

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

def test_postgres_features(postgres):
    # Test PostgreSQL-specific features
    with postgres.connect() as conn:
        conn.execute("""
            CREATE TABLE users (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100),
                email VARCHAR(255) UNIQUE
            )
        """)
        conn.execute(
            "INSERT INTO users (name, email) VALUES (%s, %s)",
            ("Alice", "alice@example.com")
        )
        result = conn.execute("SELECT * FROM users WHERE email = %s", ("alice@example.com",))
        user = result.fetchone()
        assert user.name == "Alice"

# test_api.py
import pytest
from fastapi.testclient import TestClient
from myapp.main import app
from myapp.database import get_db
from myapp.models import Base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Test database
TEST_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(TEST_DATABASE_URL)
TestingSessionLocal = sessionmaker(bind=engine)

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)

@pytest.fixture(autouse=True)
def setup_database():
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

def test_create_user():
    response = client.post("/users/", json={
        "name": "Alice",
        "email": "alice@example.com"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"
    assert data["email"] == "alice@example.com"
    assert "id" in data

def test_get_user_not_found():
    response = client.get("/users/999")
    assert response.status_code == 404
    assert response.json()["detail"] == "User not found"

def test_duplicate_email():
    client.post("/users/", json={
        "name": "Alice",
        "email": "alice@example.com"
    })
    response = client.post("/users/", json={
        "name": "Bob",
        "email": "alice@example.com"
    })
    assert response.status_code == 400
    assert "already exists" in response.json()["detail"]

# conftest.py
@pytest.fixture
def seed_users(db_session):
    users = [
        User(name="Alice", email="alice@example.com", role="admin"),
        User(name="Bob", email="bob@example.com", role="user"),
        User(name="Charlie", email="charlie@example.com", role="user"),
    ]
    for user in users:
        db_session.add(user)
    db_session.commit()
    return users

# test_admin.py
def test_admin_can_view_all_users(client, seed_users):
    login_response = client.post("/login", json={
        "email": "alice@example.com",
        "password": "admin123"
    })
    token = login_response.json()["token"]

    response = client.get("/admin/users", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200
    assert len(response.json()) == 3  # All seeded users

// test/api/users.test.js
const request = require('supertest');
const app = require('../../app');
const { User, sequelize } = require('../../models');

beforeAll(async () => {
  await sequelize.sync({ force: true });
});

afterAll(async () => {
  await sequelize.close();
});

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

    expect(response.body).toMatchObject({
      name: 'Alice',
      email: 'alice@test.com',
    });
    expect(response.body).toHaveProperty('id');
  });

  it('rejects duplicate emails', async () => {
    await User.create({ name: 'Alice', email: 'alice@test.com' });

    await request(app)
      .post('/api/users')
      .send({ name: 'Bob', email: 'alice@test.com' })
      .expect(400);
  });
});

# .github/workflows/integration.yml
name: Integration Tests
on: [pull_request]
jobs:
  integration:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: testpass
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.12' }
      - run: pip install -r requirements.txt
      - run: pytest tests/integration/
        env:
          DATABASE_URL: postgresql://postgres:testpass@localhost:5432/postgres