Merge pull request #2994 from redpanda-data/mihaitodor-fix-integratio…

…n-tests

Fix integration tests

internal/impl/aws/integration_test.go

@@ -47,7 +47,7 @@ func getLocalStack(t testing.TB) (port string) {
 		ExposedPorts: []string{port + "/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
 			docker.Port(port + "/tcp"): {
-				docker.PortBinding{HostIP: "", HostPort: port},
+				docker.PortBinding{HostIP: "", HostPort: port + "/tcp"},
 			},
 		},
 		Env: []string{
@@ -72,7 +72,7 @@ func getLocalStack(t testing.TB) (port string) {
 	return
 }
 
-func TestIntegration(t *testing.T) {
+func TestIntegrationAWS(t *testing.T) {
 	integration.CheckSkip(t)
 	t.Parallel()
 

internal/impl/cockroachdb/exploration_test.go

@@ -46,7 +46,7 @@ func TestIntegrationExploration(t *testing.T) {
 		Repository:   "cockroachdb/cockroach",
 		Tag:          "latest",
 		Cmd:          []string{"start-single-node", "--insecure"},
-		ExposedPorts: []string{"8080", "26257"},
+		ExposedPorts: []string{"8080/tcp", "26257/tcp"},
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() {

internal/impl/cockroachdb/integration_test.go

@@ -46,7 +46,7 @@ func TestIntegrationCRDB(t *testing.T) {
 		Repository:   "cockroachdb/cockroach",
 		Tag:          "latest",
 		Cmd:          []string{"start-single-node", "--insecure"},
-		ExposedPorts: []string{"8080", "26257"},
+		ExposedPorts: []string{"8080/tcp", "26257/tcp"},
 	})
 	require.NoError(t, err)
 	t.Cleanup(func() {

internal/impl/cypher/output_test.go

@@ -48,7 +48,7 @@ func makeBatch(args ...string) service.MessageBatch {
 	return batch
 }
 
-func TestIntegration(t *testing.T) {
+func TestIntegrationCypher(t *testing.T) {
 	integration.CheckSkip(t)
 	t.Parallel()
 

internal/impl/elasticsearch/aws/integration_test.go

@@ -53,6 +53,7 @@ var elasticIndex = `{
 }`
 
 func TestIntegrationElasticsearchAWS(t *testing.T) {
+	// TODO: Fix this test after migrating to the new Elasticsearch client libs.
 	t.Skip("Struggling to get localstack es to work, maybe one day")
 
 	integration.CheckSkip(t)

internal/impl/hdfs/integration_test.go

@@ -24,12 +24,13 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	_ "github.com/redpanda-data/benthos/v4/public/components/pure"
 	"github.com/redpanda-data/benthos/v4/public/service/integration"
 )
 
 func TestIntegrationHDFS(t *testing.T) {
 	integration.CheckSkip(t)
-	// t.Skip() // Skip until we fix the static port bindings
+
 	t.Parallel()
 
 	pool, err := dockertest.NewPool("")
@@ -41,12 +42,12 @@ func TestIntegrationHDFS(t *testing.T) {
 		Repository:   "cybermaggedon/hadoop",
 		Tag:          "2.8.2",
 		Hostname:     "localhost",
-		ExposedPorts: []string{"9000", "50075", "50070", "50010"},
+		ExposedPorts: []string{"9000/tcp", "50075/tcp", "50070/tcp", "50010/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9000/tcp":  {{HostIP: "", HostPort: "9000"}},
-			"50070/tcp": {{HostIP: "", HostPort: "50070"}},
-			"50075/tcp": {{HostIP: "", HostPort: "50075"}},
-			"50010/tcp": {{HostIP: "", HostPort: "50010"}},
+			"9000/tcp":  {{HostIP: "", HostPort: "9000/tcp"}},
+			"50070/tcp": {{HostIP: "", HostPort: "50070/tcp"}},
+			"50075/tcp": {{HostIP: "", HostPort: "50075/tcp"}},
+			"50010/tcp": {{HostIP: "", HostPort: "50010/tcp"}},
 		},
 	}
 	resource, err := pool.RunWithOptions(options)

internal/impl/kafka/enterprise/integration_test.go

@@ -91,7 +91,7 @@ func readNKafkaMessages(ctx context.Context, t testing.TB, address, topic string
 	return
 }
 
-func TestIntegration(t *testing.T) {
+func TestKafkaEnterpriseIntegration(t *testing.T) {
 	integration.CheckSkip(t)
 	t.Parallel()
 
@@ -106,9 +106,9 @@ func TestIntegration(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",

internal/impl/kafka/integration_franz_test.go

@@ -47,9 +47,9 @@ func TestIntegrationFranz(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -216,9 +216,9 @@ func TestIntegrationFranzSasl(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -327,9 +327,9 @@ func TestIntegrationFranzOutputFixedTimestamp(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -399,9 +399,9 @@ func BenchmarkIntegrationFranz(b *testing.B) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",

internal/impl/kafka/integration_sarama_test.go

@@ -17,7 +17,6 @@ package kafka_test
 import (
 	"context"
 	"fmt"
-	"runtime"
 	"strconv"
 	"sync"
 	"testing"
@@ -54,9 +53,9 @@ func TestIntegrationSaramaCheckpointOneLockUp(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -217,9 +216,9 @@ func TestIntegrationSaramaRedpanda(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -453,9 +452,6 @@ input:
 
 func TestIntegrationSaramaOld(t *testing.T) {
 	integration.CheckSkip(t)
-	if runtime.GOOS == "darwin" {
-		t.Skip("skipping test on macos")
-	}
 
 	t.Parallel()
 
@@ -472,9 +468,9 @@ func TestIntegrationSaramaOld(t *testing.T) {
 	kafkaResource, err := pool.RunWithOptions(&dockertest.RunOptions{
 		Repository:   "bitnami/kafka",
 		Tag:          "latest",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Env: []string{
 			"KAFKA_CFG_NODE_ID=0",
@@ -683,9 +679,9 @@ func TestIntegrationSaramaOutputFixedTimestamp(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",

internal/impl/kafka/integration_test.go

@@ -115,9 +115,9 @@ func TestIntegrationRedpanda(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -268,9 +268,9 @@ func TestIntegrationRedpandaSasl(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -376,9 +376,9 @@ func TestIntegrationRedpandaOutputFixedTimestamp(t *testing.T) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",
@@ -448,9 +448,9 @@ func BenchmarkIntegrationRedpanda(b *testing.B) {
 		Repository:   "redpandadata/redpanda",
 		Tag:          "latest",
 		Hostname:     "redpanda",
-		ExposedPorts: []string{"9092"},
+		ExposedPorts: []string{"9092/tcp"},
 		PortBindings: map[docker.Port][]docker.PortBinding{
-			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}},
+			"9092/tcp": {{HostIP: "", HostPort: kafkaPortStr + "/tcp"}},
 		},
 		Cmd: []string{
 			"redpanda",

internal/impl/mongodb/input_test.go

@@ -71,7 +71,7 @@ func TestInputIntegration(t *testing.T) {
 			"MONGO_INITDB_ROOT_USERNAME=mongoadmin",
 			"MONGO_INITDB_ROOT_PASSWORD=secret",
 		},
-		ExposedPorts: []string{"27017"},
+		ExposedPorts: []string{"27017/tcp"},
 	})
 	require.NoError(t, err)
 

internal/impl/mongodb/integration_test.go

@@ -52,7 +52,7 @@ func TestIntegrationMongoDB(t *testing.T) {
 			"MONGO_INITDB_ROOT_USERNAME=mongoadmin",
 			"MONGO_INITDB_ROOT_PASSWORD=secret",
 		},
-		ExposedPorts: []string{"27017"},
+		ExposedPorts: []string{"27017/tcp"},
 	})
 	require.NoError(t, err)
 

internal/impl/mongodb/processor_test.go

@@ -49,7 +49,7 @@ func TestProcessorIntegration(t *testing.T) {
 			"MONGO_INITDB_ROOT_USERNAME=mongoadmin",
 			"MONGO_INITDB_ROOT_PASSWORD=secret",
 		},
-		ExposedPorts: []string{"27017"},
+		ExposedPorts: []string{"27017/tcp"},
 	})
 	require.NoError(t, err)
 

internal/impl/nsq/integration_test.go

@@ -22,7 +22,7 @@ import (
 	"github.com/redpanda-data/benthos/v4/public/service/integration"
 )
 
-func TestIntegration(t *testing.T) {
+func TestIntegrationNSQ(t *testing.T) {
 	t.Parallel()
 
 	{

internal/impl/opensearch/integration_test.go

@@ -50,7 +50,7 @@ func outputFromConf(t testing.TB, confStr string, args ...any) *opensearch.Outpu
 	return o
 }
 
-func TestIntegration(t *testing.T) {
+func TestIntegrationOpensearch(t *testing.T) {
 	integration.CheckSkip(t)
 	t.Parallel()
 
@@ -131,7 +131,7 @@ func TestIntegration(t *testing.T) {
 	})
 
 	t.Run("TestOpenSearchErrorHandling", func(te *testing.T) {
-		testOpenSearchErrorHandling(urls, client, te)
+		testOpenSearchErrorHandling(urls, te)
 	})
 
 	t.Run("TestOpenSearchConnect", func(te *testing.T) {
@@ -252,7 +252,7 @@ action: index
 	}
 }
 
-func testOpenSearchErrorHandling(urls []string, client *os.Client, t *testing.T) {
+func testOpenSearchErrorHandling(urls []string, t *testing.T) {
 	ctx, done := context.WithTimeout(context.Background(), time.Second*30)
 	defer done()
 

internal/impl/sftp/integration_test.go

@@ -35,7 +35,7 @@ var (
 	sftpPassword = "pass"
 )
 
-func TestIntegration(t *testing.T) {
+func TestIntegrationSFTP(t *testing.T) {
 	integration.CheckSkip(t)
 	t.Parallel()
 

internal/impl/spicedb/watch_input_test.go

@@ -28,7 +28,7 @@ import (
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
-func TestIntegration(t *testing.T) {
+func TestIntegrationSpiceDB(t *testing.T) {
 	integration.CheckSkip(t)
 	t.Parallel()
 	ctx := context.Background()
@@ -67,22 +67,22 @@ cache: test_cache
 	err = pool.Retry(func() error {
 		r, err := client.WriteSchema(ctx, &v1.WriteSchemaRequest{
 			Schema: `
-	definition user {}
-	 
-	definition document {
-			relation writer: user
-			relation reader: user
-	 
-			/**
-			 * edit determines whether a user can edit the document
-			 */
-			permission edit = writer
-	 
-			/**
-			 * view determines whether a user can view the document
-			 */
-			permission view = reader + writer
-	}`,
+definition user {}
+
+definition document {
+	relation writer: user
+	relation reader: user
+
+	/**
+	* edit determines whether a user can edit the document
+	*/
+	permission edit = writer
+
+	/**
+	* view determines whether a user can view the document
+	*/
+	permission view = reader + writer
+}`,
 		})
 		if err != nil {
 			return err