testcontainers-go

使用 Testcontainers 为 Go 进行集成测试

你是一位精通使用 Testcontainers 进行集成测试的 Go 开发者。启用此技能后，你应：

• 始终优先使用预配置模块，而不是通用容器（如果已有对应模块）
• 在编写的每个测试中遵循先清理、再检查错误的模式
• 使用恰当的等待策略，而不是 time.Sleep()——绝不要建议将 time.Sleep() 用作同步机制
• 生成完整、可运行的测试代码，包括所有必要的导入
• 遵循 Go 测试惯例，例如表驱动测试、t.Parallel()、构建标签和子测试

描述

此技能可帮助你使用 Go 语言库 Testcontainers 编写集成测试。该库可提供常见数据库、消息队列、Web 浏览器或任何可在 Docker 容器中运行内容的轻量级临时实例。

核心功能：

• 使用 62+ 个常见服务（数据库、消息队列、云服务等）的预配置模块
• 在 Go 测试中设置和管理 Docker 容器
• 配置网络、卷和环境变量
• 实现恰当的清理和资源管理
• 调试和排查容器问题

何时使用此技能

触发关键词： 集成测试、testcontainers、docker 测试、容器测试、数据库测试、使用 postgres 测试、使用 redis 测试、使用 kafka 测试、真实数据库测试、端到端测试基础设施、测试环境设置、测试清理、测试隔离。

在以下情况下使用此技能：

• 编写需要真实服务（数据库、消息队列等）的集成测试
• 针对依赖项的多个版本或配置进行测试
• 创建可复现的测试环境
• 在集成测试中避免模拟外部依赖项
• 设置临时测试基础设施
• 在 CI/CD 流水线（GitHub Actions、GitLab CI 等）中运行集成测试
• 在同一软件包的多个测试之间共享容器设置

前提条件

• 已安装并运行 Docker 或 Podman
• Go 1.24+（查看 go.mod 了解项目的特定要求）
• Docker 套接字可通过标准路径访问（macOS/Windows 上的 Docker Desktop，Linux 上的 /var/run/docker.sock）

决策指南

使用此决策树选择合适的方法：

Need a container for testing?
├── Is there a pre-configured module? (check module list below)
│   ├── YES → Use the module (Section 2)
│   └── NO  → Use a generic container (Section 3)
│
├── Need multiple containers to communicate?
│   └── YES → Create a custom network (Section 5)
│
├── Need shared setup across tests in a package?
│   └── YES → Use TestMain pattern (Section 10)
│
├── Need to separate integration tests from unit tests?
│   └── YES → Use build tags (Section 10)
│
└── Running in CI/CD?
    └── YES → See CI/CD Integration (Section 11)

说明

1. 安装与设置

将 testcontainers-go 添加到项目中：

go get github.com/testcontainers/testcontainers-go

对于预配置模块（推荐）：

# Example: PostgreSQL module
go get github.com/testcontainers/testcontainers-go/modules/postgres

# Example: Kafka module
go get github.com/testcontainers/testcontainers-go/modules/kafka

# Example: Redis module
go get github.com/testcontainers/testcontainers-go/modules/redis

验证 Docker 是否可用：

func TestDockerAvailable(t *testing.T) {
    testcontainers.SkipIfProviderIsNotHealthy(t)
    // Test will skip if Docker is not running
}

---

2. 使用预配置模块（推荐方法）

Testcontainers for Go 提供 62+ 个预配置模块，这些模块提供可用于生产环境的配置、合理的默认值和辅助方法。如果有可用模块，始终优先使用模块，而不是通用容器。

为什么使用模块？

• 合理的默认值：预配置的端口、环境变量和等待策略
• 连接辅助方法：内置的 ConnectionString()、Endpoint() 等方法
• 专用功能：模块特有的功能（例如 Postgres 快照、Kafka 主题管理）
• 自动凭据：安全的凭据生成和管理
• 经受实战检验：数千个项目已在生产环境中使用

可用模块类别

数据库（17 个模块）：

• postgres、mysql、mariadb、mongodb、redis、valkey
• cockroachdb、clickhouse、memcached、influxdb
• arangodb、cassandra、scylladb、dynamodb
• dolt、databend、surrealdb

消息队列（6 个模块）：

• kafka、rabbitmq、nats、pulsar、redpanda、solace

搜索与向量数据库（9 个模块）：

• elasticsearch、opensearch、meilisearch
• weaviate、qdrant、chroma、milvus、vearch、pinecone

云与基础设施（6 个模块）：

• gcloud、azure、azurite、localstack、dind、k3s

服务与工具（13 个模块）：

• consul、etcd、neo4j、couchbase、vault、openldap
• artemis、inbucket、mockserver、nebulagraph、minio
• toxiproxy、aerospike

开发（10 个模块）：

• compose、registry、k6、ollama、grafana-lgtm
• dockermodelrunner、dockermcpgateway、socat、mssql

基本模块使用模式

package myapp_test

import (
    "context"
    "testing"

    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/postgres"
)

func TestWithPostgres(t *testing.T) {
    ctx := context.Background()

    // Start PostgreSQL container with sensible defaults
    pgContainer, err := postgres.Run(ctx, "postgres:16-alpine")
    testcontainers.CleanupContainer(t, pgContainer)
    require.NoError(t, err)

    // Get connection string - credentials auto-generated
    connStr, err := pgContainer.ConnectionString(ctx)
    require.NoError(t, err)
    // connStr: "postgres://postgres:password@localhost:49153/postgres?sslmode=disable"

    // Use connection string with your database driver
    db, err := sql.Open("postgres", connStr)
    require.NoError(t, err)
    defer db.Close()

    // Run your tests...
}

使用选项配置模块

模块支持三个级别的自定义：

级别 1：简单选项（通过 testcontainers.CustomizeRequestOption）

pgContainer, err := postgres.Run(
    ctx,
    "postgres:16-alpine",
    testcontainers.WithEnv(map[string]string{
        "POSTGRES_DB": "myapp_test",
    }),
    testcontainers.WithLabels(map[string]string{
        "env": "test",
    }),
)

级别 2：模块专属选项

// PostgreSQL with init scripts
pgContainer, err := postgres.Run(
    ctx,
    "postgres:16-alpine",
    postgres.WithInitScripts("./testdata/init.sql"),
    postgres.WithDatabase("myapp_test"),
    postgres.WithUsername("custom_user"),
    postgres.WithPassword("custom_pass"),
)

// Redis with configuration
redisContainer, err := redis.Run(
    ctx,
    "redis:7-alpine",
    redis.WithSnapshotting(10, 1),
    redis.WithLogLevel(redis.LogLevelVerbose),
)

// Kafka with custom config
kafkaContainer, err := kafka.Run(
    ctx,
    "confluentinc/confluent-local:7.5.0",
    kafka.WithClusterID("test-cluster"),
)

级别 3：使用生命周期钩子的高级配置

// PostgreSQL with custom initialization
pgContainer, err := postgres.Run(
    ctx,
    "postgres:16-alpine",
    postgres.WithDatabase("myapp"),
    testcontainers.WithLifecycleHooks(
        testcontainers.ContainerLifecycleHooks{
            PostStarts: []testcontainers.ContainerHook{
                func(ctx context.Context, c testcontainers.Container) error {
                    // Custom initialization after container starts
                    return nil
                },
            },
        },
    ),
)

模块专属辅助方法

大多数模块除了 ConnectionString() 之外，还提供便捷方法：

// PostgreSQL: Snapshot & Restore for test isolation
func TestDatabaseIsolation(t *testing.T) {
    ctx := context.Background()

    pgContainer, err := postgres.Run(ctx, "postgres:16-alpine")
    testcontainers.CleanupContainer(t, pgContainer)
    require.NoError(t, err)

    connStr, _ := pgContainer.ConnectionString(ctx)
    db, _ := sql.Open("postgres", connStr)

    // Create initial data
    db.Exec("CREATE TABLE users (id SERIAL PRIMARY KEY, name TEXT)")
    db.Exec("INSERT INTO users (name) VALUES ('Alice')")

    // Close connection before snapshot (PostgreSQL can't snapshot with active connections)
    db.Close()

    // Take snapshot
    err = pgContainer.Snapshot(ctx, postgres.WithSnapshotName("initial"))
    require.NoError(t, err)

    // Reconnect and make changes
    db, _ = sql.Open("postgres", connStr)
    db.Exec("INSERT INTO users (name) VALUES ('Bob')")

    // Close connection before restore
    db.Close()

    // Restore to snapshot
    err = pgContainer.Restore(ctx, postgres.WithSnapshotName("initial"))
    require.NoError(t, err)

    // Reconnect — Bob is gone, only Alice remains
    db, _ = sql.Open("postgres", connStr)
    defer db.Close()
}

// Kafka: Get bootstrap servers
kafkaContainer, _ := kafka.Run(ctx, "confluentinc/confluent-local:7.5.0")
brokers, _ := kafkaContainer.Brokers(ctx)

查找合适的模块

1. 浏览可用模块：https://testcontainers.com/modules/?language=go（完整且最新的列表）
2. 查看 modules 目录：testcontainers-go GitHub 仓库中的 /modules/
3. 模块文档：https://golang.testcontainers.org/modules/（各模块的在线文档）
4. 按类别浏览（见上方列表）
5. 搜索示例：每个模块的目录中都有 examples_test.go

模块位置模式：

github.com/testcontainers/testcontainers-go/modules/<module-name>

---

3. 使用通用容器（备用方案）

没有预配置模块时，请使用通用容器。

重要：公开端口时务必添加等待策略，以确保容器在测试运行前已就绪。这对可靠性至关重要，尤其是在 CI 环境中。切勿用 time.Sleep 替代——这是反模式，会导致测试不稳定。

func TestCustomContainer(t *testing.T) {
    ctx := context.Background()

    ctr, err := testcontainers.Run(
        ctx,
        "custom-image:latest",
        testcontainers.WithExposedPorts("8080/tcp"),
        testcontainers.WithEnv(map[string]string{
            "APP_ENV": "test",
        }),
        // CRITICAL: Always add wait strategy for exposed ports
        testcontainers.WithWaitStrategy(
            wait.ForListeningPort("8080/tcp").WithStartupTimeout(time.Second*30),
        ),
    )
    testcontainers.CleanupContainer(t, ctr)
    require.NoError(t, err)

    // Get endpoint
    endpoint, err := ctr.Endpoint(ctx, "http")
    require.NoError(t, err)
}

常见通用容器选项：

testcontainers.Run(
    ctx,
    "image:tag",

    // Ports
    testcontainers.WithExposedPorts("80/tcp", "443/tcp"),

    // Environment
    testcontainers.WithEnv(map[string]string{
        "KEY": "value",
    }),

    // Files
    testcontainers.WithFiles(testcontainers.ContainerFile{
        Reader:            strings.NewReader("content"),
        ContainerFilePath: "/app/config.yml",
        FileMode:          0o644,
    }),

    // Volumes
    testcontainers.WithHostConfigModifier(func(hc *container.HostConfig) {
        hc.Binds = []string{"/host/path:/container/path"}
    }),

    // Wait strategies (REQUIRED when using WithExposedPorts)
    // Use wait.ForListeningPort for reliability - never use time.Sleep!
    testcontainers.WithWaitStrategy(
        wait.ForListeningPort("80/tcp"),
        // Or use other strategies: wait.ForLog(), wait.ForHTTP(), etc.
    ),

    // Commands
    testcontainers.WithAfterReadyCommand(
        testcontainers.NewRawCommand([]string{"echo", "initialized"}),
    ),

    // Labels
    testcontainers.WithLabels(map[string]string{
        "app": "myapp",
    }),
)

---

4. 编写集成测试

测试结构最佳实践

package myapp_test

import (
    "context"
    "testing"

    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/postgres"
)

func TestDatabaseOperations(t *testing.T) {
    // 1. Setup: Create context
    ctx := context.Background()

    // 2. Start container
    pgContainer, err := postgres.Run(ctx, "postgres:16-alpine")

    // 3. CRITICAL: Register cleanup BEFORE error check
    testcontainers.CleanupContainer(t, pgContainer)

    // 4. Check for errors
    require.NoError(t, err)

    // 5. Get connection details
    connStr, err := pgContainer.ConnectionString(ctx)
    require.NoError(t, err)

    // 6. Connect to service
    db, err := sql.Open("postgres", connStr)
    require.NoError(t, err)
    defer db.Close()

    // 7. Run your tests
    err = db.Ping()
    require.NoError(t, err)

    // Test your application logic here...
}

关键模式：先清理，再检查错误

// CORRECT:
ctr, err := testcontainers.Run(ctx, "nginx:alpine")
testcontainers.CleanupContainer(t, ctr)  // Register cleanup immediately
require.NoError(t, err)                   // Then check error

// WRONG: Creates resource leaks
ctr, err := testcontainers.Run(ctx, "nginx:alpine")
require.NoError(t, err)                   // If this fails...
testcontainers.CleanupContainer(t, ctr)  // ...cleanup never registers

使用容器的表驱动测试

func TestMultipleVersions(t *testing.T) {
    ctx := context.Background()

    versions := []struct {
        name  string
        image string
    }{
        {"Postgres 14", "postgres:14-alpine"},
        {"Postgres 15", "postgres:15-alpine"},
        {"Postgres 16", "postgres:16-alpine"},
    }

    for _, tc := range versions {
        t.Run(tc.name, func(t *testing.T) {
            pgContainer, err := postgres.Run(ctx, tc.image)
            testcontainers.CleanupContainer(t, pgContainer)
            require.NoError(t, err)

            // Run tests against this version...
        })
    }
}

并行执行测试

func TestParallelContainers(t *testing.T) {
    t.Parallel()  // Enable parallel execution

    ctx := context.Background()

    pgContainer, err := postgres.Run(ctx, "postgres:16-alpine")
    testcontainers.CleanupContainer(t, pgContainer)
    require.NoError(t, err)

    // Each parallel test gets its own container
}

---

5. 容器网络

连接多个容器

import "github.com/testcontainers/testcontainers-go/network"

func TestMultipleServices(t *testing.T) {
    ctx := context.Background()

    // Create custom network
    nw, err := network.New(ctx)
    testcontainers.CleanupNetwork(t, nw)
    require.NoError(t, err)

    // Start database on network
    pgContainer, err := postgres.Run(
        ctx,
        "postgres:16-alpine",
        network.WithNetwork([]string{"database"}, nw),
    )
    testcontainers.CleanupContainer(t, pgContainer)
    require.NoError(t, err)

    // Start application on same network
    appContainer, err := testcontainers.Run(
        ctx,
        "myapp:latest",
        testcontainers.WithEnv(map[string]string{
            "DB_HOST": "database",  // Can reach via network alias
            "DB_PORT": "5432",      // Use internal port, not mapped port
        }),
        network.WithNetwork([]string{"app"}, nw),
    )
    testcontainers.CleanupContainer(t, appContainer)
    require.NoError(t, err)

    // Test application can communicate with database...
}

访问容器端口

func TestPortAccess(t *testing.T) {
    ctx := context.Background()

    ctr, err := testcontainers.Run(
        ctx,
        "nginx:alpine",
        testcontainers.WithExposedPorts("80/tcp"),
    )
    testcontainers.CleanupContainer(t, ctr)
    require.NoError(t, err)

    // Method 1: Get full endpoint (recommended)
    endpoint, err := ctr.Endpoint(ctx, "http")
    require.NoError(t, err)
    // endpoint = "http://localhost:49153"

    // Method 2: Get mapped port only
    port, err := ctr.MappedPort(ctx, "80/tcp")
    require.NoError(t, err)
    portNum := port.Int()  // e.g., 49153

    // Method 3: Get host and port separately
    host, err := ctr.Host(ctx)
    require.NoError(t, err)
    // host = "localhost" (or docker host IP)
}

---

6. 资源管理与清理

清理方法

方法 1：testcontainers.CleanupContainer()（推荐）

func TestRecommendedCleanup(t *testing.T) {
    ctx := context.Background()

    ctr, err := testcontainers.Run(ctx, "nginx:alpine")
    testcontainers.CleanupContainer(t, ctr)  // Registers with t.Cleanup
    require.NoError(t, err)

    // Container automatically cleaned up when test ends
}

方法 2：t.Cleanup()（手动）

func TestManualCleanup(t *testing.T) {
    ctx := context.Background()

    ctr, err := testcontainers.Run(ctx, "nginx:alpine")
    require.NoError(t, err)

    t.Cleanup(func() {
        err := testcontainers.TerminateContainer(ctr)
        require.NoError(t, err)
    })
}

方法 3：defer（旧式）

func TestDeferCleanup(t *testing.T) {
    ctx := context.Background()

    ctr, err := testcontainers.Run(ctx, "nginx:alpine")
    require.NoError(t, err)

    defer func() {
        err := testcontainers.TerminateContainer(ctr)
        require.NoError(t, err)
    }()
}

清理选项

// Cleanup with custom timeout
testcontainers.CleanupContainer(t, ctr,
    testcontainers.StopTimeout(10*time.Second),
)

// Cleanup and remove volumes
testcontainers.CleanupContainer(t, ctr,
    testcontainers.RemoveVolumes("volume1", "volume2"),
)

// Combine options
testcontainers.CleanupContainer(t, ctr,
    testcontainers.StopTimeout(5*time.Second),
    testcontainers.RemoveVolumes("data"),
)

使用 Ryuk 自动清理

Go 的 Testcontainers 使用 Ryuk 作为垃圾回收器，即使测试崩溃或超时，也会自动清理容器：

• 以 sidecar 容器运行（testcontainers/ryuk:0.13.0）
• 监控测试会话生命周期
• 会话结束时清理容器
• 处理并行测试执行

控制 Ryuk 的行为：

// Disable Ryuk (not recommended)
os.Setenv("TESTCONTAINERS_RYUK_DISABLED", "true")

// Enable verbose logging
os.Setenv("RYUK_VERBOSE", "true")

// Adjust timeouts
os.Setenv("RYUK_CONNECTION_TIMEOUT", "2m")
os.Setenv("RYUK_RECONNECTION_TIMEOUT", "30s")

---

7. 配置模式

环境变量

testcontainers.Run(
    ctx,
    "myapp:latest",
    testcontainers.WithEnv(map[string]string{
        "DATABASE_URL": "postgres://localhost/db",
        "LOG_LEVEL":    "debug",
        "API_KEY":      "test-key",
    }),
)

在容器中执行命令

使用 Exec() 执行命令时，建议使用 exec.Multiplexed() 正确处理 Docker 的输出格式：

import "github.com/testcontainers/testcontainers-go/exec"

// Execute command with Multiplexed option
exitCode, reader, err := ctr.Exec(ctx, []string{"sh", "-c", "echo 'hello'"}, exec.Multiplexed())
require.NoError(t, err)
require.Equal(t, 0, exitCode)

// Read the output
output, err := io.ReadAll(reader)
require.NoError(t, err)
fmt.Println(string(output))

为什么使用 exec.Multiplexed()？

• 从输出中移除 Docker 的多路复用标头
• 将 stdout 和 stderr 合并为一个干净的数据流
• 让输出更易于阅读和解析

如果不使用 exec.Multiplexed()，将会得到 Docker 的原始多路复用数据流，其中包含难以解析的标头字节。

文件和目录

// Copy single file
testcontainers.Run(
    ctx,
    "nginx:alpine",
    testcontainers.WithFiles(testcontainers.ContainerFile{
        Reader:            strings.NewReader("server { listen 80; }"),
        ContainerFilePath: "/etc/nginx/conf.d/default.conf",
        FileMode:          0o644,
    }),
)

// Copy multiple files
testcontainers.Run(
    ctx,
    "myapp:latest",
    testcontainers.WithFiles(
        testcontainers.ContainerFile{...},  // config.yml
        testcontainers.ContainerFile{...},  // secrets.json
    ),
)

// Copy from container after start
ctr, _ := testcontainers.Run(ctx, "nginx:alpine")
reader, err := ctr.CopyFileFromContainer(ctx, "/etc/nginx/nginx.conf")
content, _ := io.ReadAll(reader)

卷挂载

testcontainers.Run(
    ctx,
    "postgres:16",
    testcontainers.WithHostConfigModifier(func(hc *container.HostConfig) {
        // Bind mount
        hc.Binds = []string{
            "/host/data:/var/lib/postgresql/data",
        }

        // Named volume
        hc.Mounts = []mount.Mount{
            {
                Type:   mount.TypeVolume,
                Source: "pgdata",
                Target: "/var/lib/postgresql/data",
            },
        }
    }),
)

临时文件系统

testcontainers.Run(
    ctx,
    "myapp:latest",
    testcontainers.WithTmpfs(map[string]string{
        "/tmp":      "rw",
        "/app/temp": "rw,size=100m,mode=1777",
    }),
)

---

8. 等待策略

等待策略对于可靠的测试至关重要。 它们可确保容器在测试运行前完全就绪，这在时序可能变化的 CI 环境中尤其重要。

最佳实践：

• ✅ 暴露端口时始终使用 wait.ForListeningPort()——这是最可靠的做法
• ✅ 根据服务选择合适的等待策略（HTTP 健康检查、日志模式等）
• ❌ 切勿使用 time.Sleep()——这是反模式，会导致测试不稳定
• ✅ 设置合理的超时时间，以应对运行缓慢的 CI 环境

基于端口的等待（暴露端口时推荐）

import "github.com/testcontainers/testcontainers-go/wait"

testcontainers.Run(
    ctx,
    "postgres:16",
    testcontainers.WithWaitStrategy(
        wait.ForListeningPort("5432/tcp").
            WithStartupTimeout(30*time.Second).
            WithPollInterval(1*time.Second),
    ),
)

基于日志的等待

testcontainers.Run(
    ctx,
    "elasticsearch:8.7.0",
    testcontainers.WithWaitStrategy(
        wait.ForLog("started").
            WithStartupTimeout(60*time.Second).
            WithOccurrence(1),
    ),
)

基于 HTTP 的等待

testcontainers.Run(
    ctx,
    "myapp:latest",
    testcontainers.WithWaitStrategy(
        wait.ForHTTP("/health").
            WithPort("8080/tcp").
            WithStatusCodeMatcher(func(status int) bool {
                return status == 200
            }).
            WithStartupTimeout(30*time.Second),
    ),
)

基于 SQL 的等待

testcontainers.Run(
    ctx,
    "postgres:16",
    testcontainers.WithWaitStrategy(
        wait.ForSQL("5432/tcp", "postgres", func(host string, port nat.Port) string {
            return fmt.Sprintf("postgres://user:pass@%s:%s/db?sslmode=disable",
                host, port.Port())
        }).WithStartupTimeout(30*time.Second),
    ),
)

多种等待策略

testcontainers.Run(
    ctx,
    "myapp:latest",
    testcontainers.WithWaitStrategy(
        wait.ForAll(
            wait.ForListeningPort("8080/tcp"),
            wait.ForLog("Application started"),
            wait.ForHTTP("/health"),
        ),
    ),
)

---

9. 故障排除

检查 Docker 是否可用

func TestDockerConnection(t *testing.T) {
    testcontainers.SkipIfProviderIsNotHealthy(t)

    ctx := context.Background()
    cli, err := testcontainers.NewDockerClientWithOpts(ctx)
    require.NoError(t, err)

    info, err := cli.Info(ctx)
    require.NoError(t, err)

    t.Logf("Docker version: %s", info.ServerVersion)
    t.Logf("OS: %s", info.OperatingSystem)
}

调试容器日志

func TestWithLogging(t *testing.T) {
    ctx := context.Background()

    // Method 1: Stream to stdout
    ctr, _ := testcontainers.Run(
        ctx,
        "myapp:latest",
        testcontainers.WithLogConsumers(
            &testcontainers.StdoutLogConsumer{},
        ),
    )
    testcontainers.CleanupContainer(t, ctr)

    // Method 2: Read logs manually
    rc, _ := ctr.Logs(ctx)
    defer rc.Close()
    logs, _ := io.ReadAll(rc)
    t.Logf("Container logs:\n%s", string(logs))

    // Method 3: Inspect container
    info, _ := ctr.Inspect(ctx)
    t.Logf("Container state: %+v", info.State)
}

常见问题

问题：容器启动超时

// Increase wait timeout
testcontainers.WithWaitStrategy(
    wait.ForListeningPort("5432/tcp").
        WithStartupTimeout(60*time.Second),  // Increase from default
)

// Check logs to see what's happening
testcontainers.WithLogConsumers(&testcontainers.StdoutLogConsumer{})

问题：端口已被占用

• Testcontainers 会自动分配随机端口
• 除非必要，否则不要手动指定主机端口
• 检查是否有未清理的容器：docker ps -a

问题：拉取镜像失败

# Pull manually first to verify
docker pull postgres:16

# For private registries, login first
docker login registry.example.com
# Testcontainers will use credentials from ~/.docker/config.json

问题：容器未清理

// Verify Ryuk is running
docker ps | grep ryuk

// Check cleanup is registered correctly
testcontainers.CleanupContainer(t, ctr)  // Before error check!

用于调试的环境变量

# Enable Ryuk verbose logging
export RYUK_VERBOSE=true

# Adjust timeouts
export RYUK_CONNECTION_TIMEOUT=2m
export RYUK_RECONNECTION_TIMEOUT=30s

# Custom Docker socket
export DOCKER_HOST=unix:///var/run/docker.sock

# Registry prefix for private registry
export TESTCONTAINERS_HUB_IMAGE_NAME_PREFIX=private.registry.com

---

10. 共享设置与构建标签

集成测试的构建标签

使用 Go 构建标签将集成测试与单元测试分开。这样，除非明确选择运行集成测试，否则执行 go test ./... 时不会运行它们。

//go:build integration

package myapp_test

import (
    "context"
    "testing"

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

func TestIntegrationWithPostgres(t *testing.T) {
    ctx := context.Background()

    pgContainer, err := postgres.Run(ctx, "postgres:16-alpine")
    testcontainers.CleanupContainer(t, pgContainer)
    require.NoError(t, err)

    // Integration test logic...
}

显式运行集成测试：

# Run only integration tests
go test -tags=integration ./...

# Run only unit tests (default, no tag needed)
go test ./...

使用 TestMain 共享容器设置

如果一个包中的多个测试需要使用同一个容器，可以使用 TestMain 启动一次容器，并让所有测试共享它。这能大幅缩短测试执行时间。

package myapp_test

import (
    "context"
    "database/sql"
    "fmt"
    "os"
    "testing"

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

var testDB *sql.DB

func TestMain(m *testing.M) {
    ctx := context.Background()

    pgContainer, err := postgres.Run(ctx, "postgres:16-alpine",
        postgres.WithDatabase("testdb"),
    )
    if err != nil {
        fmt.Fprintf(os.Stderr, "failed to start postgres: %s\n", err)
        os.Exit(1)
    }

    connStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable")
    if err != nil {
        fmt.Fprintf(os.Stderr, "failed to get connection string: %s\n", err)
        os.Exit(1)
    }

    testDB, err = sql.Open("postgres", connStr)
    if err != nil {
        fmt.Fprintf(os.Stderr, "failed to open db: %s\n", err)
        os.Exit(1)
    }

    // Run tests
    code := m.Run()

    // Cleanup
    testDB.Close()
    testcontainers.TerminateContainer(pgContainer)

    os.Exit(code)
}

func TestCreateUser(t *testing.T) {
    // testDB is shared across all tests — no container startup delay
    _, err := testDB.Exec("INSERT INTO users (name) VALUES ($1)", "Alice")
    require.NoError(t, err)
}

func TestGetUser(t *testing.T) {
    var name string
    err := testDB.QueryRow("SELECT name FROM users LIMIT 1").Scan(&name)
    require.NoError(t, err)
}

何时使用 TestMain，何时为每个测试单独创建容器：

• TestMain：包中的所有测试都需要相同的服务，且测试速度很重要
• 每个测试单独创建容器：测试需要不同的配置，或测试隔离至关重要

---

11. CI/CD 集成

GitHub Actions

name: Integration Tests
on: [push, pull_request]

jobs:
  integration-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version: '1.24'
      - name: Run integration tests
        run: go test -tags=integration -v -timeout 10m ./...

CI 的关键注意事项：

• GitHub Actions 的 Ubuntu 运行器已预装 Docker
• 增加测试超时时间（-timeout 10m），以适应 CI 中的容器启动
• 调试时使用 -count=1 禁用测试缓存
• 可考虑使用 -race 标志检测竞态条件

CI 的环境配置

// Detect CI environment and adjust timeouts
func getStartupTimeout() time.Duration {
    if os.Getenv("CI") != "" {
        return 120 * time.Second // Longer timeout in CI
    }
    return 30 * time.Second
}

// Use in wait strategy
testcontainers.WithWaitStrategy(
    wait.ForListeningPort("5432/tcp").
        WithStartupTimeout(getStartupTimeout()),
)

---

反模式

使用 Testcontainers 编写测试时，请避免以下常见错误。

❌ 使用 time.Sleep() 进行同步

// WRONG: Flaky, slow, unreliable
ctr, err := testcontainers.Run(ctx, "postgres:16",
    testcontainers.WithExposedPorts("5432/tcp"),
)
time.Sleep(5 * time.Second) // DON'T DO THIS

// CORRECT: Use wait strategies
ctr, err := testcontainers.Run(ctx, "postgres:16",
    testcontainers.WithExposedPorts("5432/tcp"),
    testcontainers.WithWaitStrategy(
        wait.ForListeningPort("5432/tcp").
            WithStartupTimeout(30*time.Second),
    ),
)

❌ 注册清理操作前检查错误

// WRONG: Resource leak if Run succeeds but error is non-nil
ctr, err := testcontainers.Run(ctx, "nginx:alpine")
require.NoError(t, err)                   // If this panics...
testcontainers.CleanupContainer(t, ctr)   // ...cleanup never runs

// CORRECT: Always register cleanup first
ctr, err := testcontainers.Run(ctx, "nginx:alpine")
testcontainers.CleanupContainer(t, ctr)   // Register cleanup immediately
require.NoError(t, err)                   // Then check error

❌ 存在模块时仍使用通用容器

// WRONG: Reinventing the wheel
ctr, err := testcontainers.Run(ctx, "postgres:16-alpine",
    testcontainers.WithExposedPorts("5432/tcp"),
    testcontainers.WithEnv(map[string]string{
        "POSTGRES_USER":     "user",
        "POSTGRES_PASSWORD": "pass",
        "POSTGRES_DB":       "mydb",
    }),
    testcontainers.WithWaitStrategy(
        wait.ForLog("database system is ready to accept connections").
            WithOccurrence(2),
    ),
)

// CORRECT: Use the PostgreSQL module
pgContainer, err := postgres.Run(ctx, "postgres:16-alpine",
    postgres.WithDatabase("mydb"),
    postgres.WithUsername("user"),
    postgres.WithPassword("pass"),
)

❌ 硬编码端口

// WRONG: Port conflicts, breaks in CI
host, _ := ctr.Host(ctx)
url := fmt.Sprintf("http://%s:8080/api", host) // Hardcoded port!

// CORRECT: Use mapped ports
endpoint, err := ctr.Endpoint(ctx, "http")
url := fmt.Sprintf("%s/api", endpoint) // Uses dynamically assigned port

❌ 暴露端口却未设置等待策略

// WRONG: Container may not be ready when tests start
ctr, err := testcontainers.Run(ctx, "myapp:latest",
    testcontainers.WithExposedPorts("8080/tcp"),
    // No wait strategy!
)

// CORRECT: Always add a wait strategy when exposing ports
ctr, err := testcontainers.Run(ctx, "myapp:latest",
    testcontainers.WithExposedPorts("8080/tcp"),
    testcontainers.WithWaitStrategy(
        wait.ForListeningPort("8080/tcp").
            WithStartupTimeout(30*time.Second),
    ),
)

---

验证清单

使用 Testcontainers 编写测试后，请检查以下事项：

• 在错误检查前注册清理操作 — CleanupContainer(t, ctr) 出现在 require.NoError(t, err) 之前
• 有可用模块时使用模块 — 不要对已有模块的服务使用通用容器
• 存在等待策略 — 每个 WithExposedPorts() 都有对应的 WithWaitStrategy()（或模块提供了等待策略）
• 不使用 time.Sleep() 进行同步 — 改用等待策略
• 没有硬编码端口 — 使用 MappedPort() 或 Endpoint() 获取动态分配的端口
• 导入正确 — 已导入所有必需的软件包（模块、wait、network 等）
• 传入上下文 — 创建 context.Background() 并将其传入容器操作
• 关闭资源 — 使用 defer 关闭数据库连接、Redis 客户端等
• 使用构建标签（可选）— 对集成测试文件使用 //go:build integration 标签
• 测试可编译 — 运行 go vet ./... 和 go build ./...，及早发现问题

---

示例

示例 1：PostgreSQL 集成测试

package myapp_test

import (
    "context"
    "database/sql"
    "testing"

    _ "github.com/lib/pq"
    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/postgres"
)

func TestUserRepository(t *testing.T) {
    ctx := context.Background()

    // Start PostgreSQL container
    pgContainer, err := postgres.Run(
        ctx,
        "postgres:16-alpine",
        postgres.WithDatabase("testdb"),
        postgres.WithUsername("testuser"),
        postgres.WithPassword("testpass"),
        postgres.WithInitScripts("./testdata/schema.sql"),
    )
    testcontainers.CleanupContainer(t, pgContainer)
    require.NoError(t, err)

    // Get connection string
    connStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable")
    require.NoError(t, err)

    // Connect to database
    db, err := sql.Open("postgres", connStr)
    require.NoError(t, err)
    defer db.Close()

    // Test your repository
    repo := NewUserRepository(db)

    t.Run("CreateUser", func(t *testing.T) {
        user := &User{Name: "Alice", Email: "[email protected]"}
        err := repo.Create(user)
        require.NoError(t, err)
        require.NotZero(t, user.ID)
    })

    t.Run("GetUser", func(t *testing.T) {
        user, err := repo.GetByEmail("[email protected]")
        require.NoError(t, err)
        require.Equal(t, "Alice", user.Name)
    })
}

示例 2：Redis 缓存测试

package cache_test

import (
    "context"
    "testing"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/redis"
)

func TestRedisCache(t *testing.T) {
    ctx := context.Background()

    // Start Redis container
    redisContainer, err := redis.Run(
        ctx,
        "redis:7-alpine",
        redis.WithSnapshotting(10, 1),
        redis.WithLogLevel(redis.LogLevelVerbose),
    )
    testcontainers.CleanupContainer(t, redisContainer)
    require.NoError(t, err)

    // Get connection string
    connStr, err := redisContainer.ConnectionString(ctx)
    require.NoError(t, err)

    // Connect to Redis
    opt, err := redis.ParseURL(connStr)
    require.NoError(t, err)

    client := redis.NewClient(opt)
    defer client.Close()

    // Test cache operations
    t.Run("SetAndGet", func(t *testing.T) {
        err := client.Set(ctx, "key1", "value1", time.Minute).Err()
        require.NoError(t, err)

        val, err := client.Get(ctx, "key1").Result()
        require.NoError(t, err)
        require.Equal(t, "value1", val)
    })

    t.Run("Expiration", func(t *testing.T) {
        err := client.Set(ctx, "key2", "value2", time.Second).Err()
        require.NoError(t, err)

        time.Sleep(2 * time.Second)

        _, err = client.Get(ctx, "key2").Result()
        require.Equal(t, redis.Nil, err)
    })
}

示例 3：Kafka 生产者/消费者测试

package messaging_test

import (
    "context"
    "testing"
    "time"

    "github.com/segmentio/kafka-go"
    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/kafka"
)

func TestKafkaMessaging(t *testing.T) {
    ctx := context.Background()

    // Start Kafka container
    kafkaContainer, err := kafka.Run(
        ctx,
        "confluentinc/confluent-local:7.5.0",
        kafka.WithClusterID("test-cluster"),
    )
    testcontainers.CleanupContainer(t, kafkaContainer)
    require.NoError(t, err)

    // Get bootstrap servers
    brokers, err := kafkaContainer.Brokers(ctx)
    require.NoError(t, err)

    topic := "test-topic"

    // Create producer
    writer := kafka.NewWriter(kafka.WriterConfig{
        Brokers: brokers,
        Topic:   topic,
    })
    defer writer.Close()

    // Create consumer
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: brokers,
        Topic:   topic,
        GroupID: "test-group",
    })
    defer reader.Close()

    // Test message flow
    t.Run("ProduceAndConsume", func(t *testing.T) {
        // Produce message
        err := writer.WriteMessages(ctx, kafka.Message{
            Key:   []byte("key1"),
            Value: []byte("Hello, Kafka!"),
        })
        require.NoError(t, err)

        // Consume message
        msg, err := reader.ReadMessage(ctx)
        require.NoError(t, err)
        require.Equal(t, "Hello, Kafka!", string(msg.Value))
    })
}

示例 4：多容器应用程序栈

package integration_test

import (
    "context"
    "net/http"
    "testing"

    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/postgres"
    "github.com/testcontainers/testcontainers-go/modules/redis"
    "github.com/testcontainers/testcontainers-go/network"
)

func TestFullStack(t *testing.T) {
    ctx := context.Background()

    // Create custom network
    nw, err := network.New(ctx)
    testcontainers.CleanupNetwork(t, nw)
    require.NoError(t, err)

    // Start PostgreSQL
    pgContainer, err := postgres.Run(
        ctx,
        "postgres:16-alpine",
        network.WithNetwork([]string{"database"}, nw),
    )
    testcontainers.CleanupContainer(t, pgContainer)
    require.NoError(t, err)

    // Start Redis
    redisContainer, err := redis.Run(
        ctx,
        "redis:7-alpine",
        network.WithNetwork([]string{"cache"}, nw),
    )
    testcontainers.CleanupContainer(t, redisContainer)
    require.NoError(t, err)

    // Start application
    appContainer, err := testcontainers.Run(
        ctx,
        "myapp:latest",
        testcontainers.WithEnv(map[string]string{
            "DB_HOST":    "database",
            "DB_PORT":    "5432",
            "REDIS_HOST": "cache",
            "REDIS_PORT": "6379",
        }),
        testcontainers.WithExposedPorts("8080/tcp"),
        network.WithNetwork([]string{"app"}, nw),
    )
    testcontainers.CleanupContainer(t, appContainer)
    require.NoError(t, err)

    // Get application endpoint
    endpoint, err := appContainer.Endpoint(ctx, "http")
    require.NoError(t, err)

    // Test application
    resp, err := http.Get(endpoint + "/health")
    require.NoError(t, err)
    require.Equal(t, 200, resp.StatusCode)
}

示例 5：Docker Compose 栈

package compose_test

import (
    "context"
    "testing"

    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/compose"
)

func TestComposeStack(t *testing.T) {
    ctx := context.Background()

    // Start services from docker-compose.yml
    composeStack, err := compose.NewDockerCompose("./docker-compose.yml")
    require.NoError(t, err)

    t.Cleanup(func() {
        if err := composeStack.Down(ctx); err != nil {
            t.Fatalf("failed to down compose stack: %v", err)
        }
    })

    err = composeStack.Up(ctx, compose.Wait(true))
    require.NoError(t, err)

    // Get service container
    webContainer, err := composeStack.ServiceContainer(ctx, "web")
    require.NoError(t, err)

    // Test service
    endpoint, err := webContainer.Endpoint(ctx, "http")
    require.NoError(t, err)

    // Run tests against the stack...
}

---

最佳实践

1. 有可用的预配置模块时，务必使用 — 它们提供合理的默认值和辅助方法
2. 立即注册清理操作 — 在检查错误前调用 testcontainers.CleanupContainer(t, ctr)
3. 公开端口时务必添加等待策略 — 使用 wait.ForListeningPort() 确保可靠性，尤其是在 CI 中。不要使用 time.Sleep()——这是反模式，会导致测试不稳定
4. 选择合适的等待策略 — 对健康检查端点使用 wait.ForHTTP()，对日志模式使用 wait.ForLog()，或使用 wait.ForListeningPort() 检查端口是否可用
5. 利用表驱动测试 — 针对多个版本或配置进行测试
6. 使用自定义网络 — 用于多容器通信
7. 保持容器短暂存在 — 不要依赖测试之间的状态
8. 检查 Docker 是否可用 — 使用 testcontainers.SkipIfProviderIsNotHealthy(t)
9. 启用并行执行 — 使用 t.Parallel() 加快测试套件运行
10. 使用模块辅助方法 — 例如 ConnectionString()、Snapshot()、Restore()
11. 通过日志进行调试 — 排查问题时使用 WithLogConsumers()

---

其他资源

• 官方文档：https://golang.testcontainers.org/
• 可用模块：https://testcontainers.com/modules/?language=go（完整且最新的列表）
• 模块文档：https://golang.testcontainers.org/modules/（各模块的在线文档）
• GitHub 仓库：https://github.com/testcontainers/testcontainers-go
• 模块示例：查看 GitHub 仓库中的 modules/*/examples_test.go 文件
• 社区 Slack：testcontainers.slack.com