gcp/changestreams: add emulator test harness

Author: mmatczuk

internal/impl/gcp/enterprise/changestreams/emulator_test.go

@@ -0,0 +1,204 @@
+// Copyright 2025 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package changestreams
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"cloud.google.com/go/spanner"
+	adminapi "cloud.google.com/go/spanner/admin/database/apiv1"
+	adminpb "cloud.google.com/go/spanner/admin/database/apiv1/databasepb"
+	instance "cloud.google.com/go/spanner/admin/instance/apiv1"
+	"cloud.google.com/go/spanner/admin/instance/apiv1/instancepb"
+	"github.com/ory/dockertest/v3"
+	"github.com/ory/dockertest/v3/docker"
+	"google.golang.org/api/option"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func startSpannerEmulator(t *testing.T) (addr string) {
+	pool, err := dockertest.NewPool("")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Log("Starting emulator")
+	res, err := pool.RunWithOptions(&dockertest.RunOptions{
+		Repository: "gcr.io/cloud-spanner-emulator/emulator",
+		Tag:        "latest",
+		Env: []string{
+			"SPANNER_EMULATOR_HOST=0.0.0.0:9010",
+		},
+		ExposedPorts: []string{"9010/tcp"},
+		PortBindings: map[docker.Port][]docker.PortBinding{
+			"9010/tcp": {
+				{HostIP: "0.0.0.0", HostPort: "9010"},
+			},
+		},
+	}, func(cfg *docker.HostConfig) {
+		cfg.AutoRemove = true
+		cfg.RestartPolicy = docker.RestartPolicy{
+			Name: "no",
+		}
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	closeFn := func() {
+		if err := pool.Purge(res); err != nil {
+			t.Errorf("Failed to purge resource: %v", err)
+		}
+		t.Log("Emulator stopped")
+	}
+
+	addr = fmt.Sprintf("localhost:%s", res.GetPort("9010/tcp"))
+
+	if err := pool.Retry(func() error {
+		t.Logf("Waiting for emulator to be ready at %s", addr)
+		conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+		if err != nil {
+			return err
+		}
+		conn.Close()
+		return nil
+	}); err != nil {
+		closeFn()
+		t.Fatal(err)
+	}
+
+	t.Cleanup(closeFn)
+	return
+}
+
+func createInstance(ctx context.Context, conn *grpc.ClientConn) (string, error) {
+	const (
+		emulatorProjectID  = "test-project"
+		emulatorInstanceID = "test-instance"
+	)
+
+	adm, err := instance.NewInstanceAdminClient(ctx,
+		option.WithGRPCConn(conn),
+		option.WithoutAuthentication(),
+	)
+	if err != nil {
+		return "", err
+	}
+	// Do not close as it will close the grpc connection
+
+	op, err := adm.CreateInstance(ctx, &instancepb.CreateInstanceRequest{
+		Parent:     "projects/" + emulatorProjectID,
+		InstanceId: emulatorInstanceID,
+		Instance: &instancepb.Instance{
+			Config:          "projects/" + emulatorProjectID + "/instanceConfigs/regional-europe-central2",
+			DisplayName:     emulatorInstanceID,
+			ProcessingUnits: 100,
+		},
+	})
+	if err != nil {
+		return "", err
+	}
+
+	resp, err := op.Wait(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	return resp.Name, nil
+}
+
+type emulatorAdminClient struct {
+	*adminapi.DatabaseAdminClient
+	instanceName string
+
+	t    *testing.T
+	conn *grpc.ClientConn
+}
+
+func newEmulatorHelper(t *testing.T) *emulatorAdminClient {
+	t.Helper()
+
+	// Create a gRPC connection to the emulator
+	conn, err := grpc.NewClient(startSpannerEmulator(t),
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx := t.Context()
+
+	// Create an instance
+	instanceName, err := createInstance(ctx, conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create the database admin client with the gRPC connection
+	adm, err := adminapi.NewDatabaseAdminClient(ctx,
+		option.WithGRPCConn(conn),
+		option.WithoutAuthentication())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return &emulatorAdminClient{
+		DatabaseAdminClient: adm,
+		instanceName:        instanceName,
+
+		t:    t,
+		conn: conn,
+	}
+}
+
+func (a *emulatorAdminClient) fullDatabaseName(dbName string) string {
+	return fmt.Sprintf("%s/databases/%s", a.instanceName, dbName)
+}
+
+func (a *emulatorAdminClient) createTestDatabase(dbName string, opts ...func(*adminpb.CreateDatabaseRequest)) (*spanner.Client, error) {
+	req := &adminpb.CreateDatabaseRequest{
+		Parent:          a.instanceName,
+		CreateStatement: fmt.Sprintf("CREATE DATABASE %s", dbName),
+	}
+	for _, o := range opts {
+		o(req)
+	}
+
+	a.t.Logf("Creating test database %q", dbName)
+	ctx := a.t.Context()
+	op, err := a.CreateDatabase(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := op.Wait(ctx); err != nil {
+		return nil, err
+	}
+	c, err := spanner.NewClient(ctx, a.fullDatabaseName(dbName), option.WithGRPCConn(a.conn))
+	if err != nil {
+		return nil, err
+	}
+	a.t.Cleanup(c.Close)
+
+	return c, nil
+}
+
+func (a *emulatorAdminClient) CreateTestDatabase(dbName string, opts ...func(*adminpb.CreateDatabaseRequest)) *spanner.Client {
+	c, err := a.createTestDatabase(dbName, opts...)
+	if err != nil {
+		a.t.Fatal(err)
+	}
+	return c
+}
+
+func (a *emulatorAdminClient) Close() error {
+	return errors.Join(a.DatabaseAdminClient.Close(), a.conn.Close())
+}