package test import ( "context" "fmt" "queryorchestration/internal/queue" "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/docker/go-connections/nat" "github.com/stretchr/testify/assert" "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/wait" ) type QueueConfig struct { Container testcontainers.Container Client *sqs.Client URL string External *ExternalQueue } type ExternalQueue struct { Endpoint string Credentials aws.CredentialsProvider Region string URL string } type CreateQueueConfig struct { Network *testcontainers.DockerNetwork } func CreateQueue(t *testing.T, ctx context.Context, cfg *CreateQueueConfig) (*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.ForAll( // wait.ForExposedPort(), wait.ForListeningPort(port), wait.ForLog("Ready."), ), } if cfg.Network != nil { req.Networks = []string{cfg.Network.Name} req.NetworkAliases = map[string][]string{ cfg.Network.Name: {alias}, } } 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()) sqsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region), config.WithCredentialsProvider(provider), config.WithBaseEndpoint(endpoint)) if err != nil { t.Fatal(err) } client := sqs.NewFromConfig(sqsCfg) queueM, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{ QueueName: aws.String(queueName), }) if err != nil { t.Fatal(err) } endpoint = fmt.Sprintf("http://%s:%s", alias, port.Port()) return &QueueConfig{ Container: container, Client: client, URL: *queueM.QueueUrl, External: &ExternalQueue{ Region: region, Credentials: provider, Endpoint: endpoint, }, }, func() { err := container.Terminate(ctx) if err != nil { t.Error(err) } } } func AssertMessageWait(t *testing.T, ctx context.Context, cfg *queue.Config, attrs []string) types.Message { time.Sleep(time.Second) result, err := queue.Receive(ctx, cfg, attrs) assert.Nil(t, err) assert.Len(t, result.Messages, 1) assert.NotNil(t, result.Messages[0]) return result.Messages[0] } func AssertMessageBodyWait(t *testing.T, ctx context.Context, cfg *queue.Config, body string) { message := AssertMessageWait(t, ctx, cfg, []string{}) assert.Equal(t, body, *message.Body) } func AssertMessageAttrWait(t *testing.T, ctx context.Context, cfg *queue.Config, name string, value string) { message := AssertMessageWait(t, ctx, cfg, []string{name}) assert.NotNil(t, message.MessageAttributes[name]) assert.Equal(t, value, *(message.MessageAttributes[name]).StringValue) }