Integration Testing

是什么

这是一份集成测试规范，覆盖真实数据库与消息队列接入、Testcontainers（一次性测试容器）配置、服务边界内端到端测试、服务间契约测试，让团队在上线前就发现单元测试覆盖不到的接口契约和数据集成问题。

怎么用

1. 新模块开发时优先用 Testcontainers 启真实数据库或 Kafka（分布式消息队列），让测试和生产用相同中间件版本。
2. 接口写完单测后立刻补 HTTP 端到端测试，验证完整请求路径包括序列化、鉴权、错误处理。
3. 跨服务调用按文档的契约测试模板编写，消费方和提供方一致才能合并主干。
4. CI（持续集成）流水线把集成测试和单元测试分阶段跑，单测过了再跑集成测试节约资源。
5. 测试用例每月评审一次，移除冗余案例，新增覆盖盲点，让测试套件保持精干。

架构图

flowchart LR
    A[业务代码] --> B[Testcontainers 启依赖]
    B --> C[端到端请求]
    C --> D[契约校验]
    D --> E[测试报告]
    E --> F[CI 流水线]

Integration Testing

Integration tests verify that components work correctly together — real databases, real HTTP routing, real serialization — catching the bugs that unit tests cannot.

When to Activate

• Testing code that reads from or writes to a real database
• Testing an HTTP endpoint end-to-end within the service boundary
• Setting up Testcontainers for a project
• Writing contract tests between two services
• Testing event-driven or message queue workflows
• Managing test fixtures, factories, and teardown strategies
• Diagnosing tests that pass in isolation but fail in CI

Integration vs Unit: The Boundary

Integration tests add real I/O, real wiring, and real serialization. They catch mismatches between your code and the actual database schema, ORM behavior, HTTP middleware ordering, and message envelope formats. What they cost: seconds instead of milliseconds, a Docker dependency, and a higher flakiness risk if not isolated properly.

Testing pyramid starting point: 70% unit / 20% integration / 10% E2E.

Database Integration Tests with Testcontainers

Testcontainers spins up a real Docker container (Postgres, MySQL, Redis, Kafka, etc.) per test suite, giving every developer and CI run an identical, isolated database. No shared test databases, no "works on my machine."

Python

# 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()

TypeScript

// 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

// 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()
}

Fixture Factories

Build test objects with a factory: sensible defaults, every field overridable.

# 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"))

Schema Migrations Before Suite

Always run migrations against the container before tests run — never against a pre-seeded snapshot.

• Python/Alembic: alembic upgrade head pointed at the container URL
• TypeScript/Prisma: prisma migrate deploy with DATABASE_URL set to container URL
• Go/golang-migrate: migrate -path ./migrations -database $DSN up

HTTP API Testing at the Service Level

Test the full request/response cycle — routing, middleware, validation, serialization — without going over the network. The goal is to exercise real handler wiring, not mocked HTTP.

Python (FastAPI / Starlette)

# 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

TypeScript (Express / Fastify)

// 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);
});

Go

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) }
}

Auth in Integration Tests

Injecting real tokens makes tests slow and fragile. Instead:

# 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

Contract Testing with Pact

Consumer-driven contract testing lets two services independently verify the API shape they agree on, without running both services simultaneously.

Consumer side: The consumer writes a Pact — a description of the interactions it expects. Pact runs a mock provider to record the contract.

Provider side: The provider fetches the Pact from the Pact Broker and replays each interaction against its real implementation.

Pact Broker: A shared registry where consumers publish contracts and providers pull them. Teams can see which consumers depend on which provider endpoints.

# 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);
});

When to Use / When Not to Use Pact

Message Queue and Event Testing

Testing Producers

Capture published messages in a list and assert on payload shape and schema.

# 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") }
}

Testing Consumers

Inject a raw message into the handler and assert on side effects (DB row created, email sent, etc.).

# 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")

Test Data Management

Factory Pattern

A factory builds valid domain objects with sensible defaults. Every field is overridable. Never write raw SQL inserts in test bodies.

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

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

Seeders vs Per-Test Factories

Cleanup Strategies

Snapshot Testing

Serialize a complex object to a file; fail if it changes. Useful for stable API response shapes.

# 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();
});

Test Environment Isolation

Never share a test database URL with development or production. Keep a separate .env.test (or config/test.yaml) that is committed to the repo but contains only non-sensitive test-specific values.

# .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/

Schema-per-Test vs Separate DB per Worker

Parallel Test Safety

# BAD — two workers both insert user with email "admin@example.com"
user = UserFactory(email="admin@example.com")

# GOOD — unique per worker run
user = UserFactory(email=f"admin-{uuid.uuid4()}@example.com")
# Or use factory_boy sequences which are process-local
user = UserFactory()  # email = factory.Sequence(lambda n: f"user{n}@example.com")

CI Considerations

• Ensure the CI runner has a Docker socket available (GitHub Actions ubuntu-latest does by default).
• For environments without Docker (some hosted runners, sandboxed CI), use Testcontainers Cloud — it offloads container startup to a remote daemon with no local Docker required.
• Cache container images in CI to reduce startup latency (docker pull postgres:16 as a warm-up step or via registry mirror).
• Set TESTCONTAINERS_RYUK_DISABLED=true only if your CI runner cannot start the Ryuk reaper container (rootless Docker environments).

Red Flags

• Sharing a single test database URL across all developers and CI workers — concurrent test runs corrupt each other's data; each worker needs its own isolated database, schema, or transaction scope
• Using DELETE FROM users WHERE email LIKE 'test%' for cleanup — string-matching cleanup is fragile and misses rows created by factories with generated emails; use transaction rollback or truncate instead
• Starting a Testcontainer per test function instead of per test session — container startup takes 3–10 seconds; starting one per test inflates a 100-test suite from minutes to hours; scope to session or module
• Mocking the ORM or database layer in an integration test — an integration test that mocks db.query() is still a unit test; it cannot catch index mismatches, constraint violations, or ORM-generated SQL bugs
• Hardcoded test user IDs or emails (e.g., "admin@example.com") — parallel workers both insert the same unique email and one fails with a constraint violation; use factories with sequences or UUIDs
• Testing the live external third-party API in integration tests — test reliability becomes coupled to the external service's uptime; use recorded fixtures (VCR cassettes) or a sandbox environment
• Auth bypassed by commenting out middleware — disabling auth middleware entirely hides bugs in how the middleware interacts with handlers; use a test-mode flag that injects a pre-signed test identity instead
• Running Testcontainers in CI without verifying Docker socket access — some CI runners (sandboxed, rootless Docker) cannot start containers; verify access or configure Testcontainers Cloud before the test suite is blocked in CI

Checklist

• Integration tests use a real database/broker (no mocked I/O at the storage layer)
• Testcontainers or equivalent used so tests are reproducible on any machine
• Each test cleans up after itself (transaction rollback or truncate)
• HTTP tests exercise real request parsing, validation, and response serialization
• Test data is created via factories, not hand-crafted raw SQL inserts
• Contract tests exist at every service boundary with different team ownership
• Message queue tests verify both publish payload schema and consumer side effects
• Test environment uses a separate DB from dev — never DATABASE_URL from .env
• Parallel test workers are isolated (no shared mutable rows)
• Integration test suite completes in under 5 minutes in CI
• Schema migrations run against the test container before the suite starts
• Snapshot tests exist for complex, stable API response shapes
• Auth bypassed in tests via test config, not by hard-coding credentials
• Reference/lookup data loaded via seeders; test-specific data via factories