query-orchestration/internal/queue/queue_test.go

package queue_test

import (
	"context"
	"fmt"
	"queryorchestration/internal/queue"
	"testing"

	"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/docker/go-connections/nat"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/wait"
)

type queueConfig struct {
	Container *testcontainers.Container
	Config    *queue.Config
}

func createQueue(t *testing.T, ctx context.Context) (*queueConfig, func()) {
	queueName := "test-queue"
	region := "us-east-1"
	alias := "localstack"
	provider := credentials.NewStaticCredentialsProvider("test", "test", "")

	port, err := nat.NewPort("tcp", "4566")
	if err != nil {
		t.Fatalf("Failed to create port: %v", err)
	}

	req := testcontainers.ContainerRequest{
		Image: "localstack/localstack:4.0.3",
		Env: map[string]string{
			"AWS_ACCESS_KEY_ID":      provider.Value.AccessKeyID,
			"AWS_SECRET_ACCESS_KEY":  provider.Value.SecretAccessKey,
			"AWS_SESSION_TOKEN":      provider.Value.SessionToken,
			"AWS_DEFAULT_REGION":     region,
			"AWS_REGION":             region,
			"SERVICES":               "sqs",
			"SKIP_SSL_CERT_DOWNLOAD": "1",
			"LOCALSTACK_HOST":        alias,
			"SQS_ENDPOINT_STRATEGY":  "path",
		},
		ExposedPorts: []string{port.Port()},
		WaitingFor:   wait.ForListeningPort(port),
	}

	container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
		ContainerRequest: req,
		Started:          true,
	})
	if err != nil {
		t.Fatalf("Failed to start container: %v", err)
	}

	host, err := container.Host(ctx)
	if err != nil {
		t.Fatalf("Failed to extract host: %v", err)
	}
	mappedPort, err := container.MappedPort(ctx, port)
	if err != nil {
		t.Fatalf("Failed to extract port: %v", err)
	}

	endpoint := fmt.Sprintf("http://%s:%s", host, mappedPort.Port())
	cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region), config.WithCredentialsProvider(provider), config.WithBaseEndpoint(endpoint))
	if err != nil {
		t.Fatal(err)
	}

	client := sqs.NewFromConfig(cfg)

	queueM, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{
		QueueName: aws.String(queueName),
	})
	if err != nil {
		t.Fatal(err)
	}

	return &queueConfig{
			Container: &container,
			Config: &queue.Config{
				Client: client,
				URL:    *queueM.QueueUrl,
			},
		}, func() {
			err := container.Terminate(ctx)
			if err != nil {
				t.Error(err)
			}
		}
}