chore: use testcontainers-go in the integration tests

database/cassandra/cassandra_test.go

@@ -3,120 +3,149 @@ package cassandra
 import (
 	"context"
 	"fmt"
-	"github.com/golang-migrate/migrate/v4"
-	"strconv"
 	"testing"
-)
 
-import (
-	"github.com/dhui/dktest"
 	"github.com/gocql/gocql"
-)
+	"github.com/golang-migrate/migrate/v4"
 
-import (
 	dt "github.com/golang-migrate/migrate/v4/database/testing"
-	"github.com/golang-migrate/migrate/v4/dktesting"
+
+	"github.com/testcontainers/testcontainers-go/modules/cassandra"
+
 	_ "github.com/golang-migrate/migrate/v4/source/file"
 )
 
 var (
-	opts = dktest.Options{PortRequired: true, ReadyFunc: isReady}
 	// Supported versions: http://cassandra.apache.org/download/
 	// Although Cassandra 2.x is supported by the Apache Foundation,
 	// the migrate db driver only supports Cassandra 3.x since it uses
 	// the system_schema keyspace.
 	// last ScyllaDB version tested is 5.1.11
-	specs = []dktesting.ContainerSpec{
-		{ImageName: "cassandra:3.0", Options: opts},
-		{ImageName: "cassandra:3.11", Options: opts},
-		{ImageName: "scylladb/scylla:5.1.11", Options: opts},
+	specs = []string{
+		"cassandra:3.0",
+		"cassandra:3.11",
+		"scylladb/scylla:5.1.11",
 	}
 )
 
-func isReady(ctx context.Context, c dktest.ContainerInfo) bool {
-	// Cassandra exposes 5 ports (7000, 7001, 7199, 9042 & 9160)
-	// We only need the port bound to 9042
-	ip, portStr, err := c.Port(9042)
+type CreateKeySpaceCommand struct{}
+
+func Test(t *testing.T) {
+	t.Run("test", test)
+	t.Run("testMigrate", testMigrate)
+}
+
+func startCassandra(spec string) (*cassandra.CassandraContainer, error) {
+	c, err := cassandra.Run(context.Background(), spec)
 	if err != nil {
-		return false
+		return nil, err
 	}
-	port, err := strconv.Atoi(portStr)
+
+	// from here, return the container and the error, so the client code
+	// can defer the container.Terminate call
+
+	// create a keyspace for the tests
+
+	connectionHost, err := c.ConnectionHost(context.Background())
 	if err != nil {
-		return false
+		return c, fmt.Errorf("failed to get connection host: %s", err)
 	}
 
-	cluster := gocql.NewCluster(ip)
-	cluster.Port = port
-	cluster.Consistency = gocql.All
-	p, err := cluster.CreateSession()
+	cluster := gocql.NewCluster(connectionHost)
+	session, err := cluster.CreateSession()
 	if err != nil {
-		return false
+		return c, fmt.Errorf("failed to create session: %s", err)
 	}
-	defer p.Close()
+	defer session.Close()
 	// Create keyspace for tests
-	if err = p.Query("CREATE KEYSPACE testks WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor':1}").Exec(); err != nil {
-		return false
+	if err = session.Query("CREATE KEYSPACE testks WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor':1}").Exec(); err != nil {
+		return c, fmt.Errorf("failed to create keyspace: %s", err)
 	}
-	return true
+
+	return c, nil
 }
 
-func Test(t *testing.T) {
-	t.Run("test", test)
-	t.Run("testMigrate", testMigrate)
+func test(t *testing.T) {
+	for _, spec := range specs {
+		spec := spec // capture range variable, see https://goo.gl/60w3p2
+		t.Run(spec, func(t *testing.T) {
+			t.Parallel()
 
-	t.Cleanup(func() {
-		for _, spec := range specs {
-			t.Log("Cleaning up ", spec.ImageName)
-			if err := spec.Cleanup(); err != nil {
-				t.Error("Error removing ", spec.ImageName, "error:", err)
+			c, err := startCassandra(spec)
+			if err != nil {
+				t.Fatal(err)
 			}
-		}
-	})
-}
+			t.Cleanup(func() {
+				if c == nil {
+					return
+				}
+				if err := c.Terminate(context.Background()); err != nil {
+					t.Error(err)
+				}
+			})
 
-func test(t *testing.T) {
-	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
-		ip, port, err := c.Port(9042)
-		if err != nil {
-			t.Fatal("Unable to get mapped port:", err)
-		}
-		addr := fmt.Sprintf("cassandra://%v:%v/testks", ip, port)
-		p := &Cassandra{}
-		d, err := p.Open(addr)
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer func() {
-			if err := d.Close(); err != nil {
-				t.Error(err)
+			hostPort, err := c.ConnectionHost(context.Background())
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			addr := fmt.Sprintf("cassandra://%s/testks", hostPort)
+			p := &Cassandra{}
+			d, err := p.Open(addr)
+			if err != nil {
+				t.Fatal(err)
 			}
-		}()
-		dt.Test(t, d, []byte("SELECT table_name from system_schema.tables"))
-	})
+			defer func() {
+				if err := d.Close(); err != nil {
+					t.Error(err)
+				}
+			}()
+			dt.Test(t, d, []byte("SELECT table_name from system_schema.tables"))
+		})
+	}
 }
 
 func testMigrate(t *testing.T) {
-	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
-		ip, port, err := c.Port(9042)
-		if err != nil {
-			t.Fatal("Unable to get mapped port:", err)
-		}
-		addr := fmt.Sprintf("cassandra://%v:%v/testks", ip, port)
-		p := &Cassandra{}
-		d, err := p.Open(addr)
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer func() {
-			if err := d.Close(); err != nil {
-				t.Error(err)
+	for _, spec := range specs {
+		spec := spec // capture range variable, see https://goo.gl/60w3p2
+		t.Run(spec, func(t *testing.T) {
+			t.Parallel()
+
+			c, err := startCassandra(spec)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() {
+				if c == nil {
+					return
+				}
+				if err := c.Terminate(context.Background()); err != nil {
+					t.Error(err)
+				}
+			})
+
+			hostPort, err := c.ConnectionHost(context.Background())
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			addr := fmt.Sprintf("cassandra://%s/testks", hostPort)
+			p := &Cassandra{}
+			d, err := p.Open(addr)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() {
+				if err := d.Close(); err != nil {
+					t.Error(err)
+				}
+			}()
+
+			m, err := migrate.NewWithDatabaseInstance("file://./examples/migrations", "testks", d)
+			if err != nil {
+				t.Fatal(err)
 			}
-		}()
-
-		m, err := migrate.NewWithDatabaseInstance("file://./examples/migrations", "testks", d)
-		if err != nil {
-			t.Fatal(err)
-		}
-		dt.TestMigrate(t, m)
-	})
+			dt.TestMigrate(t, m)
+		})
+	}
 }