refactoring

testhelper/docker/docker.go

@@ -4,9 +4,14 @@ import (
  "regexp"
  "strconv"
  "testing"
+ "time"
 
+ "github.com/ory/dockertest/v3"
  "github.com/ory/dockertest/v3/docker"
  "github.com/stretchr/testify/require"
+
+ "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource"
+ "github.com/rudderlabs/rudder-go-kit/testhelper/rand"
 )
 
 // GetHostPort returns the desired port mapping
@@ -26,3 +31,18 @@ func GetHostPort(t testing.TB, port string, container *docker.Container) int {
 func ToInternalDockerHost(url string) string {
  return regexp.MustCompile(`(localhost|127\.0\.0\.1)`).ReplaceAllString(url, "host.docker.internal")
 }
+
+func CreateNetwork(pool *dockertest.Pool, cln resource.Cleaner, prefix string) (*docker.Network, error) {
+ network, err := pool.Client.CreateNetwork(docker.CreateNetworkOptions{Name: prefix + "_test_" + time.Now().Format("YY-MM-DD-") + rand.String(6)})
+ if err != nil {
+ return nil, err
+ }
+
+ cln.Cleanup(func() {
+ if err := pool.Client.RemoveNetwork(network.ID); err != nil {
+ cln.Logf("Error while removing Docker network: %v", err)
+ }
+ })
+
+ return network, nil
+}

testhelper/docker/resource/etcd/etcd.go

@@ -50,7 +50,7 @@ func Setup(pool *dockertest.Pool, cln resource.Cleaner, opts ...Option) (*Resour
  "ALLOW_NONE_AUTHENTICATION=yes",
  },
  PortBindings: internal.IPv4PortBindings([]string{"2379"}),
- })
+ }, internal.DefaultHostConfig)
  if err != nil {
  return nil, fmt.Errorf("could not create container: %v", err)
  }

testhelper/docker/resource/internal/ports.go

@@ -4,6 +4,8 @@ import (
  "github.com/ory/dockertest/v3/docker"
 )
 
+const BindHostIP = "127.0.0.1"
+
 // IPv4PortBindings returns the port bindings for the given exposed ports forcing ipv4 address.
 func IPv4PortBindings(exposedPorts []string) map[docker.Port][]docker.PortBinding {
  portBindings := make(map[docker.Port][]docker.PortBinding)
@@ -16,11 +18,15 @@ func IPv4PortBindings(exposedPorts []string) map[docker.Port][]docker.PortBindin
  for _, exposedPort := range exposedPorts {
  portBindings[docker.Port(exposedPort)+"/tcp"] = []docker.PortBinding{
  {
- HostIP: "0.0.0.0",
+ HostIP: BindHostIP,
  HostPort: "0",
  },
  }
  }
 
  return portBindings
 }
+
+func DefaultHostConfig(hc *docker.HostConfig) {
+ hc.PublishAllPorts = false
+}

testhelper/docker/resource/kafka/kafka.go

@@ -3,13 +3,15 @@ package kafka
 import (
  _ "encoding/json"
  "fmt"
+ "strconv"
 
  _ "github.com/lib/pq"
  "github.com/ory/dockertest/v3"
  dc "github.com/ory/dockertest/v3/docker"
  "golang.org/x/sync/errgroup"
 
  kithelper "github.com/rudderlabs/rudder-go-kit/testhelper"
+ "github.com/rudderlabs/rudder-go-kit/testhelper/docker"
  "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource"
  "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/internal"
 )
@@ -147,15 +149,10 @@ func Setup(pool *dockertest.Pool, cln resource.Cleaner, opts ...Option) (*Resour
  network := c.network
  if c.network == nil {
  var err error
- network, err = pool.Client.CreateNetwork(dc.CreateNetworkOptions{Name: "kafka_network"})
+ network, err = docker.CreateNetwork(pool, cln, "kafka_network_")
  if err != nil {
- return nil, fmt.Errorf("could not create docker network: %w", err)
+ return nil, err
  }
- cln.Cleanup(func() {
- if err := pool.Client.RemoveNetwork(network.ID); err != nil {
- cln.Log(fmt.Errorf("could not remove kafka network: %w", err))
- }
- })
  }
 
  zookeeperPortInt, err := kithelper.GetFreePort()
@@ -172,7 +169,7 @@ func Setup(pool *dockertest.Pool, cln resource.Cleaner, opts ...Option) (*Resour
  "2181/tcp": {{HostIP: "zookeeper", HostPort: zookeeperPort}},
  },
  Env: []string{"ALLOW_ANONYMOUS_LOGIN=yes"},
- })
+ }, internal.DefaultHostConfig)
  if err != nil {
  return nil, err
  }
@@ -209,7 +206,7 @@ func Setup(pool *dockertest.Pool, cln resource.Cleaner, opts ...Option) (*Resour
  "SCHEMA_REGISTRY_ADVERTISED_HOSTNAME=schemaregistry",
  "SCHEMA_REGISTRY_CLIENT_AUTHENTICATION=NONE",
  },
- })
+ }, internal.DefaultHostConfig)
  if err != nil {
  return nil, err
  }
@@ -336,11 +333,14 @@ func Setup(pool *dockertest.Pool, cln resource.Cleaner, opts ...Option) (*Resour
  NetworkID: network.ID,
  Hostname: hostname,
  PortBindings: map[dc.Port][]dc.PortBinding{
- kafkaClientPort + "/tcp": {{HostIP: "localhost", HostPort: localhostPort}},
+ kafkaClientPort + "/tcp": {{
+ HostIP: internal.BindHostIP,
+ HostPort: strconv.Itoa(localhostPortInt),
+ }},
  },
  Mounts: mounts,
  Env: nodeEnvVars,
- })
+ }, internal.DefaultHostConfig)
  if err != nil {
  return nil, err
  }
@@ -358,6 +358,9 @@ func Setup(pool *dockertest.Pool, cln resource.Cleaner, opts ...Option) (*Resour
  containers: containers,
  }
  for i := 0; i < len(containers); i++ {
+ if containers[i].GetBoundIP(kafkaClientPort+"/tcp") == "" {
+ return nil, fmt.Errorf("could not find kafka broker port")
+ }
  res.Brokers = append(res.Brokers, containers[i].GetBoundIP(kafkaClientPort+"/tcp")+":"+containers[i].GetPort(kafkaClientPort+"/tcp"))
  }
 

testhelper/docker/resource/kafka/kafka_test.go

@@ -19,14 +19,14 @@ import (
  "github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/serde/avro"
  "github.com/linkedin/goavro/v2"
  "github.com/ory/dockertest/v3"
- dc "github.com/ory/dockertest/v3/docker"
  "github.com/segmentio/kafka-go"
  "github.com/segmentio/kafka-go/sasl"
  "github.com/segmentio/kafka-go/sasl/plain"
  "github.com/segmentio/kafka-go/sasl/scram"
  "github.com/stretchr/testify/require"
  "golang.org/x/crypto/ssh"
 
+ "github.com/rudderlabs/rudder-go-kit/testhelper/docker"
  "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/sshserver"
  "github.com/rudderlabs/rudder-go-kit/testhelper/keygen"
 )
@@ -245,14 +245,8 @@ func TestSSH(t *testing.T) {
  pool, err := dockertest.NewPool("")
  require.NoError(t, err)
 
- // Start shared Docker network
- network, err := pool.Client.CreateNetwork(dc.CreateNetworkOptions{Name: "kafka_network"})
+ network, err := docker.CreateNetwork(pool, t, "kafka_ssh_network_")
  require.NoError(t, err)
- t.Cleanup(func() {
- if err := pool.Client.RemoveNetwork(network.ID); err != nil {
- t.Logf("Error while removing Docker network: %v", err)
- }
- })
 
  // Start Kafka cluster with ZooKeeper and three brokers
  _, err = Setup(pool, t,

testhelper/docker/resource/minio/minio.go

@@ -61,7 +61,7 @@ func Setup(pool *dockertest.Pool, d resource.Cleaner, opts ...func(*Config)) (*R
  "MINIO_API_SELECT_PARQUET=on",
  }, c.Options...),
  PortBindings: internal.IPv4PortBindings([]string{"9000"}),
- })
+ }, internal.DefaultHostConfig)
  if err != nil {
  return nil, fmt.Errorf("could not start resource: %s", err)
  }

testhelper/docker/resource/mysql/mysql.go

@@ -43,7 +43,7 @@ func Setup(pool *dockertest.Pool, d resource.Cleaner, opts ...func(*Config)) (*R
  "MYSQL_ROOT_PASSWORD=" + defaultPassword,
  "MYSQL_DATABASE=" + defaultDB,
  },
- ExposedPorts: []string{"3306"},
+ ExposedPorts: []string{"3306/tcp"},
  PortBindings: internal.IPv4PortBindings([]string{"3306"}),
  }, func(hc *dc.HostConfig) {
  hc.ShmSize = c.ShmSize

testhelper/docker/resource/pulsar/pulsar.go

@@ -5,7 +5,6 @@ import (
  "fmt"
 
  "github.com/ory/dockertest/v3"
- "github.com/ory/dockertest/v3/docker"
 
  "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource"
  "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/internal"
@@ -38,9 +37,7 @@ func Setup(pool *dockertest.Pool, d resource.Cleaner, opts ...Option) (*Resource
  PortBindings: internal.IPv4PortBindings([]string{"6650", "8080"}),
  Cmd: []string{"bin/pulsar", "standalone"},
  NetworkID: networkID,
- }, func(hc *docker.HostConfig) {
- // hc.PublishAllPorts = false
- })
+ }, internal.DefaultHostConfig)
  if err != nil {
  return nil, err
  }

testhelper/docker/resource/scylla/scylla.go

@@ -33,8 +33,7 @@ func Setup(pool *dockertest.Pool, d resource.Cleaner, opts ...Option) (*Resource
  ExposedPorts: []string{"9042/tcp"},
  PortBindings: internal.IPv4PortBindings([]string{"9042"}),
  Cmd: []string{"--smp 1"},
- }, func(hc *docker.HostConfig) {
- hc.PublishAllPorts = false
+ }, internal.DefaultHostConfig, func(hc *docker.HostConfig) {
  hc.CPUCount = 1
  hc.Memory = 128 * bytesize.MB
  })

testhelper/docker/resource/types.go

@@ -2,6 +2,7 @@ package resource
 
 type Logger interface {
  Log(...interface{})
+ Logf(string, ...interface{})
 }
 
 type FailIndicator interface {

throttling/util_test.go

@@ -13,6 +13,7 @@ import (
 type tester interface {
  Helper()
  Log(...interface{})
+ Logf(string, ...any)
  Errorf(format string, args ...interface{})
  Fatalf(format string, args ...any)
  Failed() bool