Integration Testing

# pip install testcontainers[postgres] psycopg2-binary pytest sqlalchemy

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

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

@pytest.fixture(scope="session")
def engine(postgres):
    engine = create_engine(postgres.get_connection_url())
    # Run migrations before the suite
    # alembic.config.main(argv=["upgrade", "head"])
    return engine

@pytest.fixture
def db(engine):
    # BAD: commit data and delete after — leaves residue if test crashes
    # conn.execute(text("DELETE FROM users WHERE id = :id"), ...)
    # GOOD: wrap in a transaction and roll back — zero cleanup needed
    with engine.begin() as conn:
        savepoint = conn.begin_nested()
        yield conn
        savepoint.rollback()

// npm install testcontainers pg @types/pg

import { PostgreSqlContainer, StartedPostgreSqlContainer } from "testcontainers";
import { Pool } from "pg";
import { runMigrations } from "../db/migrate";

let container: StartedPostgreSqlContainer;
let pool: Pool;

beforeAll(async () => {
  container = await new PostgreSqlContainer("postgres:16").start();
  pool = new Pool({ connectionString: container.getConnectionUri() });
  await runMigrations(pool); // run prisma migrate / knex migrate:latest
}, 60_000);

afterAll(async () => {
  await pool.end();
  await container.stop();
});

beforeEach(async () => {
  await pool.query("BEGIN");
});

afterEach(async () => {
  // GOOD: rollback keeps tests hermetic
  await pool.query("ROLLBACK");
});

// go get github.com/testcontainers/testcontainers-go/modules/postgres

package db_test

import (
    "context"
    "testing"

    tcpostgres "github.com/testcontainers/testcontainers-go/modules/postgres"
    "github.com/testcontainers/testcontainers-go/wait"
)

func TestMain(m *testing.M) {
    ctx := context.Background()
    container, err := tcpostgres.RunContainer(ctx,
        tcpostgres.WithDatabase("testdb"),
        tcpostgres.WithUsername("test"),
        tcpostgres.WithPassword("test"),
        tcpostgres.WithInitScripts("schema.sql"),
        testcontainers.WithWaitStrategy(
            wait.ForLog("database system is ready to accept connections"),
        ),
    )
    if err != nil {
        panic(err)
    }
    defer container.Terminate(ctx)

    connStr, _ := container.ConnectionString(ctx, "sslmode=disable")
    // pass connStr to your repository layer
    m.Run()
}

# Python — factory_boy
import factory
from myapp.models import User

class UserFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = User

    email = factory.Sequence(lambda n: f"user{n}@example.com")
    name = "Test User"
    role = "member"
    is_active = True

# In a test:
admin = UserFactory(role="admin")
inactive = UserFactory(is_active=False)

// TypeScript — plain builder
function buildUser(overrides: Partial<User> = {}): User {
  return {
    id: crypto.randomUUID(),
    email: `user-${Date.now()}@example.com`,
    name: "Test User",
    role: "member",
    isActive: true,
    ...overrides,
  };
}

const admin = buildUser({ role: "admin" });

// Go — functional options
func NewUser(opts ...func(*User)) User {
    u := User{
        ID:       uuid.New(),
        Email:    fmt.Sprintf("user-%d@example.com", time.Now().UnixNano()),
        Name:     "Test User",
        Role:     "member",
        IsActive: true,
    }
    for _, opt := range opts {
        opt(&u)
    }
    return u
}

func WithRole(role string) func(*User) {
    return func(u *User) { u.Role = role }
}

admin := NewUser(WithRole("admin"))

# pip install httpx pytest

from fastapi.testclient import TestClient
from myapp.main import app

client = TestClient(app)

def test_user_lifecycle(db):  # db fixture provides rolled-back session
    # Create
    resp = client.post("/users", json={"email": "a@example.com", "name": "Alice"})
    assert resp.status_code == 201
    user_id = resp.json()["id"]

    # Read back
    resp = client.get(f"/users/{user_id}")
    assert resp.status_code == 200
    assert resp.json()["email"] == "a@example.com"

    # Delete
    resp = client.delete(f"/users/{user_id}")
    assert resp.status_code == 204

    resp = client.get(f"/users/{user_id}")
    assert resp.status_code == 404

// npm install supertest @types/supertest

import request from "supertest";
import { buildApp } from "../src/app";

const app = buildApp({ db: testPool });

test("user lifecycle", async () => {
  const create = await request(app)
    .post("/users")
    .send({ email: "a@example.com", name: "Alice" })
    .expect(201);

  const { id } = create.body;

  await request(app).get(`/users/${id}`).expect(200).expect((res) => {
    expect(res.body.email).toBe("a@example.com");
  });

  await request(app).delete(`/users/${id}`).expect(204);
  await request(app).get(`/users/${id}`).expect(404);
});

import (
    "net/http"
    "net/http/httptest"
    "testing"
    "bytes"
    "encoding/json"
)

func TestUserLifecycle(t *testing.T) {
    handler := buildRouter(testDB)

    // Create
    body, _ := json.Marshal(map[string]string{"email": "a@example.com", "name": "Alice"})
    w := httptest.NewRecorder()
    handler.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/users", bytes.NewReader(body)))
    if w.Code != http.StatusCreated { t.Fatalf("expected 201, got %d", w.Code) }

    var created map[string]any
    json.NewDecoder(w.Body).Decode(&created)
    id := created["id"].(string)

    // Read back
    w = httptest.NewRecorder()
    handler.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/users/"+id, nil))
    if w.Code != http.StatusOK { t.Fatalf("expected 200, got %d", w.Code) }

    // Delete
    w = httptest.NewRecorder()
    handler.ServeHTTP(w, httptest.NewRequest(http.MethodDelete, "/users/"+id, nil))
    if w.Code != http.StatusNoContent { t.Fatalf("expected 204, got %d", w.Code) }
}

# BAD
token = requests.post("/auth/login", json={"password": "..."}).json()["token"]
client.headers["Authorization"] = f"Bearer {token}"

# GOOD — test config bypasses auth middleware, or inject a pre-signed token
client = TestClient(app, headers={"X-Test-User-Id": str(user.id)})
# Middleware reads X-Test-User-Id only when TEST_MODE=true

# Consumer (Python pact-python)
from pact import Consumer, Provider

pact = Consumer("OrderService").has_pact_with(Provider("UserService"))

pact.given("user 42 exists").upon_receiving("a request for user 42").with_request(
    "GET", "/users/42"
).will_respond_with(200, body={"id": 42, "name": Like("Alice")})

with pact:
    # call your actual client code against pact.uri
    user = get_user(42, base_url=pact.uri)
    assert user["id"] == 42

// Consumer (TypeScript pact-js)
import { PactV3, MatchersV3 } from "@pact-foundation/pact";

const provider = new PactV3({ consumer: "OrderService", provider: "UserService" });

provider
  .given("user 42 exists")
  .uponReceiving("a request for user 42")
  .withRequest({ method: "GET", path: "/users/42" })
  .willRespondWith({
    status: 200,
    body: { id: MatchersV3.integer(42), name: MatchersV3.string("Alice") },
  });

await provider.executeTest(async (mockServer) => {
  const user = await getUser(42, mockServer.url);
  expect(user.id).toBe(42);
});

# Python — fake Redis pub/sub with fakeredis
import fakeredis
from myapp.events import publish_order_created

def test_publish_order_created():
    r = fakeredis.FakeRedis()
    pubsub = r.pubsub()
    pubsub.subscribe("orders")

    publish_order_created(r, order_id="abc-123", amount=99.99)

    message = pubsub.get_message(ignore_subscribe_messages=True, timeout=1)
    payload = json.loads(message["data"])
    assert payload["order_id"] == "abc-123"
    assert payload["event"] == "order.created"

// TypeScript — capture calls with a fake queue
const published: unknown[] = [];
const fakeQueue = {
  add: (name: string, data: unknown) => {
    published.push({ name, data });
    return Promise.resolve();
  },
};

await handleCheckout(fakeQueue, { orderId: "abc-123", amount: 99.99 });

expect(published).toHaveLength(1);
expect((published[0] as any).name).toBe("order.created");
expect((published[0] as any).data.orderId).toBe("abc-123");

// Go — channel-based fake
type FakePublisher struct {
    Messages []Event
}

func (f *FakePublisher) Publish(ctx context.Context, e Event) error {
    f.Messages = append(f.Messages, e)
    return nil
}

func TestPublishOrderCreated(t *testing.T) {
    pub := &FakePublisher{}
    HandleCheckout(pub, Order{ID: "abc-123", Amount: 99.99})
    if len(pub.Messages) != 1 { t.Fatal("expected 1 message") }
    if pub.Messages[0].Type != "order.created" { t.Fatal("wrong event type") }
}

# Python — Celery with CELERY_TASK_ALWAYS_EAGER
@pytest.fixture(autouse=True)
def eager_celery(settings):
    settings.CELERY_TASK_ALWAYS_EAGER = True
    settings.CELERY_TASK_EAGER_PROPAGATES = True

def test_consumer_creates_order(db):
    send_order_created.delay({"order_id": "abc-123", "amount": 99.99})
    order = db.query(Order).filter_by(id="abc-123").one()
    assert order.amount == Decimal("99.99")

# BAD
cursor.execute("INSERT INTO users (id, email, role) VALUES ('1', 'a@b.com', 'member')")

# GOOD
user = UserFactory(role="admin")

# Python — syrupy
def test_user_response_shape(snapshot, client):
    resp = client.get("/users/1")
    assert resp.json() == snapshot  # creates __snapshots__/test_users.ambr on first run

// TypeScript — jest --updateSnapshot
test("user response shape", async () => {
  const resp = await request(app).get("/users/1").expect(200);
  expect(resp.body).toMatchSnapshot();
});

# .env.test — committed, non-sensitive
DATABASE_URL=postgres://test:test@localhost:5433/apptest
REDIS_URL=redis://localhost:6380/1
MESSAGE_BROKER_URL=amqp://guest:guest@localhost:5673/