test: share testcontainers per service via pkg/testutil

broadcast-worker/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package main
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

history-service/internal/cassrepo/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package cassrepo
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

history-service/internal/mongorepo/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package mongorepo
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

history-service/internal/service/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package service_test
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

inbox-worker/integration_test.go

@@ -13,15 +13,13 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	natsmod "github.com/testcontainers/testcontainers-go/modules/nats"
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/mongo"
 
 	"github.com/hmchangw/chat/pkg/model"
 	"github.com/hmchangw/chat/pkg/stream"
 	"github.com/hmchangw/chat/pkg/subject"
 	"github.com/hmchangw/chat/pkg/testutil"
-	"github.com/hmchangw/chat/pkg/testutil/testimages"
 )
 
 func setupMongo(t *testing.T) *mongo.Database {
@@ -577,20 +575,13 @@ func TestHandleMemberAdded_DM_PersistsRemoteCounterpartSub(t *testing.T) {
 	assert.False(t, bobSub.IsSubscribed, "DM does not set IsSubscribed=true")
 }
 
-// setupNATS starts a NATS container with JetStream enabled and returns a
-// JetStream client tied to the test's lifetime.
+// setupNATS connects to the process-shared NATS (JetStream enabled in
+// testutil) and returns a JetStream client tied to the test's lifetime.
 func setupNATS(t *testing.T) (context.Context, jetstream.JetStream) {
 	t.Helper()
 	ctx := context.Background()
 
-	c, err := natsmod.Run(ctx, testimages.NATS)
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = c.Terminate(ctx) })
-
-	url, err := c.ConnectionString(ctx)
-	require.NoError(t, err)
-
-	nc, err := nats.Connect(url)
+	nc, err := nats.Connect(testutil.NATS(t))
 	require.NoError(t, err)
 	t.Cleanup(func() { nc.Close() })
 

inbox-worker/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package main
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

message-worker/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package main
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

notification-worker/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package main
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

pkg/minioutil/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package minioutil
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

pkg/mongoutil/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package mongoutil
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

pkg/natsrouter/integration_test.go

@@ -13,37 +13,21 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	tcnats "github.com/testcontainers/testcontainers-go/modules/nats"
 
 	"github.com/Marz32onE/instrumentation-go/otel-nats/otelnats"
 
 	"github.com/hmchangw/chat/pkg/natsrouter"
-	"github.com/hmchangw/chat/pkg/testutil/testimages"
+	"github.com/hmchangw/chat/pkg/testutil"
 )
 
-// setupNATS starts a real NATS container and returns a connected otelnats
-// client. Required to surface timing races that in-process NATS cannot
+// setupNATS returns an otelnats client connected to the process-shared
+// NATS. Required to surface timing races that in-process NATS cannot
 // reproduce (real TCP, real server dispatch goroutines, real latency).
 func setupNATS(t *testing.T) *otelnats.Conn {
 	t.Helper()
-	ctx := context.Background()
-
-	container, err := tcnats.Run(ctx, testimages.NATS)
-	require.NoError(t, err, "start NATS container")
-	t.Cleanup(func() {
-		// Best-effort container teardown; failures here don't affect outcome.
-		if err := container.Terminate(ctx); err != nil {
-			t.Logf("terminate nats container: %v", err)
-		}
-	})
-
-	url, err := container.ConnectionString(ctx)
-	require.NoError(t, err, "nats connection string")
-
-	nc, err := otelnats.Connect(url)
+	nc, err := otelnats.Connect(testutil.NATS(t))
 	require.NoError(t, err, "connect to NATS")
 	t.Cleanup(nc.Close)
-
 	return nc
 }
 

pkg/natsrouter/main_test.go

@@ -0,0 +1,11 @@
+//go:build integration
+
+package natsrouter_test
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

pkg/roomcrypto/main_test.go

@@ -0,0 +1,14 @@
+//go:build integration
+
+package roomcrypto
+
+// Import testutil for the Ryuk-disable init() side effect even though
+// this package starts its containers per-test (t.Cleanup handles teardown).
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

pkg/roomkeysender/main_test.go

@@ -0,0 +1,14 @@
+//go:build integration
+
+package roomkeysender_test
+
+// Import testutil for the Ryuk-disable init() side effect even though
+// this package starts its containers per-test (t.Cleanup handles teardown).
+
+import (
+	"testing"
+
+	"github.com/hmchangw/chat/pkg/testutil"
+)
+
+func TestMain(m *testing.M) { testutil.RunTests(m) }

pkg/roomkeystore/integration_test.go

@@ -6,47 +6,25 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"fmt"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/testcontainers/testcontainers-go"
-	"github.com/testcontainers/testcontainers-go/wait"
 
-	"github.com/hmchangw/chat/pkg/testutil/testimages"
+	"github.com/hmchangw/chat/pkg/testutil"
 )
 
-// setupValkey starts a valkey/valkey:8 container and returns a connected valkeyStore.
-// The container is terminated via t.Cleanup.
+// setupValkey returns a RoomKeyStore connected to the process-shared
+// Valkey, with FLUSHDB on cleanup so sibling tests start clean.
 func setupValkey(t *testing.T, gracePeriod time.Duration) RoomKeyStore {
 	t.Helper()
-	ctx := context.Background()
-
-	container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
-		ContainerRequest: testcontainers.ContainerRequest{
-			Image:        testimages.Valkey,
-			ExposedPorts: []string{"6379/tcp"},
-			WaitingFor:   wait.ForLog("Ready to accept connections"),
-		},
-		Started: true,
-	})
-	require.NoError(t, err, "start valkey container")
-	t.Cleanup(func() {
-		_ = container.Terminate(ctx) // best-effort; ignore cleanup errors
-	})
-
-	host, err := container.Host(ctx)
-	require.NoError(t, err)
-	port, err := container.MappedPort(ctx, "6379")
-	require.NoError(t, err)
-
 	store, err := NewValkeyStore(Config{
-		Addr:        fmt.Sprintf("%s:%s", host, port.Port()),
+		Addr:        testutil.Valkey(t),
 		GracePeriod: gracePeriod,
 	})
 	require.NoError(t, err, "create valkeyStore")
+	t.Cleanup(func() { testutil.FlushValkey(t) })
 	return store
 }
 

pkg/roomsubcache/integration_test.go

@@ -4,47 +4,27 @@ package roomsubcache_test
 
 import (
 	"context"
-	"fmt"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/testcontainers/testcontainers-go"
-	"github.com/testcontainers/testcontainers-go/wait"
 
 	"github.com/hmchangw/chat/pkg/roomsubcache"
-	"github.com/hmchangw/chat/pkg/testutil/testimages"
+	"github.com/hmchangw/chat/pkg/testutil"
 	"github.com/hmchangw/chat/pkg/valkeyutil"
 )
 
-// setupValkey starts a valkey/valkey:8 container and returns a connected
-// valkeyutil.Client. The container is terminated via t.Cleanup.
+// setupValkey returns a client connected to the process-shared Valkey,
+// with FLUSHDB on cleanup so sibling tests start clean.
 func setupValkey(t *testing.T) valkeyutil.Client {
 	t.Helper()
-	ctx := context.Background()
-
-	container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
-		ContainerRequest: testcontainers.ContainerRequest{
-			Image:        testimages.Valkey,
-			ExposedPorts: []string{"6379/tcp"},
-			WaitingFor:   wait.ForLog("Ready to accept connections"),
-		},
-		Started: true,
-	})
-	require.NoError(t, err, "start valkey container")
+	client, err := valkeyutil.Connect(context.Background(), testutil.Valkey(t), "")
+	require.NoError(t, err, "connect valkey")
 	t.Cleanup(func() {
-		_ = container.Terminate(ctx) // best-effort; ignore cleanup errors
+		testutil.FlushValkey(t)
+		_ = client.Close()
 	})
-
-	host, err := container.Host(ctx)
-	require.NoError(t, err)
-	port, err := container.MappedPort(ctx, "6379")
-	require.NoError(t, err)
-
-	client, err := valkeyutil.Connect(ctx, fmt.Sprintf("%s:%s", host, port.Port()), "")
-	require.NoError(t, err, "connect valkey")
-	t.Cleanup(func() { _ = client.Close() }) // best-effort; ignore cleanup errors
 	return client
 }
 

pkg/testutil/cassandra.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 	"hash/fnv"
+	"os"
 	"sync"
 	"testing"
 	"time"
@@ -18,10 +19,11 @@ import (
 const cassandraImage = "cassandra:5"
 
 var (
-	cassOnce    sync.Once
-	cassHost    string
-	cassSession *gocql.Session
-	cassInitErr error
+	cassOnce      sync.Once
+	cassContainer testcontainers.Container
+	cassHost      string
+	cassSession   *gocql.Session
+	cassInitErr   error
 )
 
 func ensureCassandraSession() (string, *gocql.Session, error) {
@@ -70,10 +72,32 @@ func ensureCassandraSession() (string, *gocql.Session, error) {
 		}
 		cassHost = addr
 		cassSession = s
+		cassContainer = container
 	})
 	return cassHost, cassSession, cassInitErr
 }
 
+// TerminateCassandra closes the shared session and stops the shared
+// container. Best-effort, idempotent.
+func TerminateCassandra() {
+	if cassSession != nil {
+		cassSession.Close()
+		cassSession = nil
+	}
+	if cassContainer != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		if err := cassContainer.Terminate(ctx); err != nil {
+			fmt.Fprintf(os.Stderr, "terminate shared cassandra: %v\n", err)
+		}
+		cassContainer = nil
+	}
+}
+
+// EnsureCassandra starts the shared Cassandra container if not already
+// started. No-t variant intended for TestMain pre-warming.
+func EnsureCassandra() error { _, _, err := ensureCassandraSession(); return err }
+
 // CassandraKeyspace creates an isolated keyspace for the test (SimpleStrategy, RF=1).
 // Returns the keyspace name, an admin session for DDL, and the container host.
 func CassandraKeyspace(t *testing.T, prefix string) (keyspace string, admin *gocql.Session, hostAddr string) {