test: Improve integration test speed with persistent test infra #90 (#93)

* test: Test infra with shared testcontainers & persistent test containers

* test: Refactor infra to support parallel tests

* test: Fix issue finding .env.test at project root dir

* test: RabbitMQ infra

* refactor: Confirm integration test before starting testinfra

* test: Apply testinfra to destination adapter integration tests

* test: Refactor internal/mqs integration tests

* test: Remove unused testutil code

* test: Fix flaky test

* docs: How to use test infra & integration test template

* chore: Remove no-longer-valid comment

Author: alexluong

.env.test

@@ -0,0 +1,3 @@
+TEST_CLICKHOUSE_URL="localhost:39000"
+TEST_LOCALSTACK_URL="localhost:34566"
+TEST_RABBITMQ_URL="localhost:35672"

Makefile

@@ -29,6 +29,12 @@ down/uptrace:
 up/portal:
 	cd internal/portal && npm install && npm run dev
 
+up/test:
+	docker-compose -f build/test/compose.yml up -d
+
+down/test:
+	docker-compose -f build/test/compose.yml down
+
 test:
 	go test $(TEST) $(TESTARGS)
 

build/test/compose.yml

@@ -0,0 +1,19 @@
+name: "outpost-test"
+
+services:
+  clickhouse:
+    image: clickhouse/clickhouse-server:24-alpine
+    ports:
+      - 39000:9000
+  rabbitmq:
+    image: rabbitmq:3-management
+    ports:
+      - 35672:5672
+      - 45672:15672
+  aws:
+    image: localstack/localstack:latest
+    environment:
+      - SERVICES=sns,sts,sqs
+    ports:
+      - 34566:4566
+      - 34571:4571

cmd/e2e/configs/basic.go

@@ -3,9 +3,8 @@ package configs
 import (
 	"testing"
 
-	"github.com/hookdeck/outpost/internal/clickhouse"
 	"github.com/hookdeck/outpost/internal/config"
-	"github.com/hookdeck/outpost/internal/mqs"
+	"github.com/hookdeck/outpost/internal/util/testinfra"
 	"github.com/hookdeck/outpost/internal/util/testutil"
 )
 
@@ -17,27 +16,13 @@ func Basic(t *testing.T) (*config.Config, func(), error) {
 		}
 	}
 
-	// Testcontainer
-	chEndpoint, cleanupCH, err := testutil.StartTestContainerClickHouse()
-	if err != nil {
-		return nil, cleanup, err
-	}
-	cleanupFns = append(cleanupFns, cleanupCH)
-
-	awsEndpoint, cleanupAWS, err := testutil.StartTestcontainerLocalstack()
-	if err != nil {
-		return nil, cleanup, err
-	}
-	cleanupFns = append(cleanupFns, cleanupAWS)
+	t.Cleanup(testinfra.Start(t))
 
 	// Config
 	redisConfig := testutil.CreateTestRedisConfig(t)
-	clickHouseConfig := &clickhouse.ClickHouseConfig{
-		Addr:     chEndpoint,
-		Username: "default",
-		Password: "",
-		Database: "default",
-	}
+	clickHouseConfig := testinfra.NewClickHouseConfig(t)
+	deliveryMQConfig := testinfra.NewMQAWSConfig(t, nil)
+	logMQConfig := testinfra.NewMQAWSConfig(t, nil)
 
 	return &config.Config{
 		Hostname:               "outpost",
@@ -49,11 +34,11 @@ func Basic(t *testing.T) (*config.Config, func(), error) {
 		PortalProxyURL:         "",
 		Topics:                 testutil.TestTopics,
 		Redis:                  redisConfig,
-		ClickHouse:             clickHouseConfig,
+		ClickHouse:             &clickHouseConfig,
 		OpenTelemetry:          nil,
 		PublishQueueConfig:     nil,
-		DeliveryQueueConfig:    &mqs.QueueConfig{AWSSQS: &mqs.AWSSQSConfig{Endpoint: awsEndpoint, Region: "us-east-1", ServiceAccountCredentials: "test:test:", Topic: "delivery"}},
-		LogQueueConfig:         &mqs.QueueConfig{AWSSQS: &mqs.AWSSQSConfig{Endpoint: awsEndpoint, Region: "us-east-1", ServiceAccountCredentials: "test:test:", Topic: "log"}},
+		DeliveryQueueConfig:    &deliveryMQConfig,
+		LogQueueConfig:         &logMQConfig,
 		PublishMaxConcurrency:  3,
 		DeliveryMaxConcurrency: 3,
 		LogMaxConcurrency:      3,

docs/contributing/test.md

@@ -39,7 +39,7 @@ $ TESTARGS='-v -run "TestJWT"'' make test
 # go test $(go list ./...) -v -run "TestJWT"
 ```
 
-Keep in mind you can't use `-run "Test..."` along with `make test/integration` as the integration test already specify integration tests with `-run` option. However, since you're already specifying which test to run, I assume this is a non-issue.
+Keep in mind you can't use `-run "Test..."` along with `make test/integration` as the integration test already specify integration tests with `-run` option. However, since you're already specifying which test to run, we assume this is a non-issue.
 
 ## Coverage
 
@@ -62,3 +62,47 @@ Running the coverage test command above will generate the `coverage.out` file. Y
 $ make test/coverage/html
 # go tool cover -html=coverage.out
 ```
+
+## Integration & E2E Tests
+
+When running integration & e2e tests, we often times require some test infrastructure such as ClickHouse, LocalStack, RabbitMQ, etc. We use [Testcontainers](https://testcontainers.com/) for that. It usually takes a few seconds (10s or so) to spawn the necessary containers. To improve the feedback loop, you can run a persistent test infrastructure and skip spawning testcontainers.
+
+To run the test infrastructure:
+
+```sh
+$ make up/test
+
+## to take the test infra down
+# $ make down/test
+```
+
+It will run a Docker compose stack called `outpost-test` which runs the necessary services at ports ":30000 + port". For example, ClickHouse usually runs on port `:9000`, so in the test infra it will run on port `:39000`.
+
+From here, you can provide env variable `TESTINFRA=1` to tell the test suite to use these services instead of spawning testcontainers.
+
+```sh
+$ TESTINFRA=1 make test
+```
+
+Tip: You can `$ export TESTINFRA=1` to use the test infra for the whole terminal session.
+
+### Integration Test Template
+
+Here's a short template for how you can write integration tests that require an external test infra:
+
+```golang
+// Integration test should always start with "TestIntegration...() {}"
+func TestIntegrationMyIntegrationTest(t *testing.T) {
+  t.Parallel()
+
+  // call testinfra.Start(t) to signal that you require the test infra.
+  // This helps the test runner properly terminate resources at the end.
+  t.Cleanup(testinfra.Start(t))
+
+  // use whichever infra you need
+  chConfig := testinfra.NewClickHouseConfig(t)
+  awsMQConfig := testinfra.NewMQAWSConfig(t, attributesMap)
+  rabbitmqConfig := testinfra.NewMQRabbitMQConfig(t)
+  // ...
+}
+```

internal/destinationadapter/adapters/aws/aws_test.go

@@ -3,21 +3,17 @@ package aws_test
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"log"
 	"testing"
 	"time"
 
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/config"
-	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
-	"github.com/aws/smithy-go"
 	"github.com/google/uuid"
 	"github.com/hookdeck/outpost/internal/destinationadapter/adapters"
 	awsadapter "github.com/hookdeck/outpost/internal/destinationadapter/adapters/aws"
-	"github.com/hookdeck/outpost/internal/util/testutil"
+	"github.com/hookdeck/outpost/internal/util/awsutil"
+	"github.com/hookdeck/outpost/internal/util/testinfra"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -111,41 +107,21 @@ func TestAWSDestination_Validate(t *testing.T) {
 }
 
 func TestIntegrationAWSDestination_Publish(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-
 	t.Parallel()
+	t.Cleanup(testinfra.Start(t))
 
-	// Setup SQS
-	awsEndpoint, terminate, err := testutil.StartTestcontainerLocalstack()
-	require.Nil(t, err)
-	defer terminate()
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	queueName := "destination_sqs_queue"
-	awsRegion := "eu-central-1"
-
-	sdkConfig, err := config.LoadDefaultConfig(ctx,
-		config.WithRegion(awsRegion),
-		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
-	)
-	require.Nil(t, err)
-	sqsClient := sqs.NewFromConfig(sdkConfig, func(o *sqs.Options) {
-		o.BaseEndpoint = aws.String(awsEndpoint)
-	})
-	queueURL, err := ensureQueue(ctx, sqsClient, queueName)
-	require.Nil(t, err)
-
-	// Setup Destination & Event
+	mq := testinfra.NewMQAWSConfig(t, nil)
+	sqsClient, err := awsutil.SQSClientFromConfig(context.Background(), mq.AWSSQS)
+	require.NoError(t, err)
+	queueURL, err := awsutil.EnsureQueue(context.Background(), sqsClient, mq.AWSSQS.Topic, nil)
+	require.NoError(t, err)
 	awsdestination := awsadapter.New()
 
 	destination := adapters.DestinationAdapterValue{
 		ID:   uuid.New().String(),
 		Type: "aws",
 		Config: map[string]string{
-			"endpoint":  awsEndpoint,
+			"endpoint":  mq.AWSSQS.Endpoint,
 			"queue_url": queueURL,
 		},
 		Credentials: map[string]string{
@@ -158,6 +134,9 @@ func TestIntegrationAWSDestination_Publish(t *testing.T) {
 	errchan := make(chan error)
 	msgchan := make(chan *types.Message)
 
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
 	go func() {
 		for {
 			out, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
@@ -212,8 +191,7 @@ func TestIntegrationAWSDestination_Publish(t *testing.T) {
 			"mykey": "myvaluee",
 		},
 	}
-	err = awsdestination.Publish(context.Background(), destination, event)
-	require.Nil(t, err)
+	require.NoError(t, awsdestination.Publish(context.Background(), destination, event))
 
 	// Assert
 	log.Println("waiting for msg...")
@@ -237,29 +215,3 @@ func TestIntegrationAWSDestination_Publish(t *testing.T) {
 		assert.Equal(t, "anothermetadatavalue", *msg.MessageAttributes["another_metadata"].StringValue)
 	}
 }
-
-func ensureQueue(ctx context.Context, sqsClient *sqs.Client, queueName string) (string, error) {
-	queue, err := sqsClient.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{
-		QueueName: aws.String(queueName),
-	})
-	if err != nil {
-		var apiErr smithy.APIError
-		if errors.As(err, &apiErr) {
-			switch apiErr.(type) {
-			case *types.QueueDoesNotExist:
-				log.Println("Queue does not exist, creating...")
-				createdQueue, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{
-					QueueName: aws.String(queueName),
-				})
-				if err != nil {
-					return "", err
-				}
-				return *createdQueue.QueueUrl, nil
-			default:
-				return "", err
-			}
-		}
-		return "", err
-	}
-	return *queue.QueueUrl, nil
-}

internal/destinationadapter/adapters/rabbitmq/rabbitmq_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/hookdeck/outpost/internal/destinationadapter/adapters"
 	"github.com/hookdeck/outpost/internal/destinationadapter/adapters/rabbitmq"
+	"github.com/hookdeck/outpost/internal/util/testinfra"
 	"github.com/hookdeck/outpost/internal/util/testutil"
 	"github.com/rabbitmq/amqp091-go"
 	"github.com/stretchr/testify/assert"
@@ -123,34 +124,22 @@ func TestRabbitMQDestination_Publish(t *testing.T) {
 }
 
 func TestIntegrationRabbitMQDestination_Publish(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-
 	t.Parallel()
+	t.Cleanup(testinfra.Start(t))
 
-	rabbitmqURL, terminate, err := testutil.StartTestcontainerRabbitMQ()
-	require.Nil(t, err)
-	defer terminate()
-
-	RABBIT_SERVER_URL := rabbitmqURL
-	const (
-		RABBIT_EXCHANGE = "destination_exchange"
-		RABBIT_QUEUE    = "destination_queue_test"
-	)
-
+	mq := testinfra.NewMQRabbitMQConfig(t)
 	rabbitmqDestination := rabbitmq.New()
 
 	destination := adapters.DestinationAdapterValue{
 		ID:   uuid.New().String(),
 		Type: "rabbitmq",
 		Config: map[string]string{
-			"server_url": testutil.ExtractRabbitURL(RABBIT_SERVER_URL),
-			"exchange":   RABBIT_EXCHANGE,
+			"server_url": testutil.ExtractRabbitURL(mq.RabbitMQ.ServerURL),
+			"exchange":   mq.RabbitMQ.Exchange,
 		},
 		Credentials: map[string]string{
-			"username": testutil.ExtractRabbitUsername(RABBIT_SERVER_URL),
-			"password": testutil.ExtractRabbitPassword(RABBIT_SERVER_URL),
+			"username": testutil.ExtractRabbitUsername(mq.RabbitMQ.ServerURL),
+			"password": testutil.ExtractRabbitPassword(mq.RabbitMQ.ServerURL),
 		},
 	}
 
@@ -174,44 +163,19 @@ func TestIntegrationRabbitMQDestination_Publish(t *testing.T) {
 	cancelChan := make(chan bool)
 	msgChan := make(chan *amqp091.Delivery)
 	go func() {
-		conn, _ := amqp091.Dial(RABBIT_SERVER_URL)
+		conn, _ := amqp091.Dial(mq.RabbitMQ.ServerURL)
 		defer conn.Close()
 		ch, _ := conn.Channel()
 		defer ch.Close()
 
-		ch.ExchangeDeclare(
-			RABBIT_EXCHANGE, // name
-			"topic",         // type
-			true,            // durable
-			false,           // auto-deleted
-			false,           // internal
-			false,           // no-wait
-			nil,             // arguments
-		)
-		q, _ := ch.QueueDeclare(
-			RABBIT_QUEUE, // name
-			false,        // durable
-			false,        // delete when unused
-			true,         // exclusive
-			false,        // no-wait
-			nil,          // arguments
-		)
-		ch.QueueBind(
-			q.Name,          // queue name
-			"",              // routing key
-			RABBIT_EXCHANGE, // exchange
-			false,
-			nil,
-		)
-
 		msgs, _ := ch.Consume(
-			RABBIT_QUEUE, // queue
-			"",           // consumer
-			true,         // auto-ack
-			false,        // exclusive
-			false,        // no-local
-			false,        // no-wait
-			nil,          // args
+			mq.RabbitMQ.Queue, // queue
+			"",                // consumer
+			true,              // auto-ack
+			false,             // exclusive
+			false,             // no-local
+			false,             // no-wait
+			nil,               // args
 		)
 
 		log.Println("ready to receive messages")
@@ -229,8 +193,7 @@ func TestIntegrationRabbitMQDestination_Publish(t *testing.T) {
 
 	<-readyChan
 	log.Println("publishing message")
-	err = rabbitmqDestination.Publish(context.Background(), destination, event)
-	assert.Nil(t, err)
+	assert.NoError(t, rabbitmqDestination.Publish(context.Background(), destination, event))
 
 	func() {
 		time.Sleep(time.Second / 2)
@@ -244,10 +207,7 @@ func TestIntegrationRabbitMQDestination_Publish(t *testing.T) {
 	}
 	log.Println("message received", msg)
 	body := make(map[string]interface{})
-	err = json.Unmarshal(msg.Body, &body)
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, json.Unmarshal(msg.Body, &body))
 	assert.Equal(t, event.Data, body)
 	// metadata
 	assert.Equal(t, "metadatavalue", msg.Headers["my_metadata"])