moveservicesInterfaces
This commit is contained in:
Michael McGuinness
@@ -0,0 +1,215 @@
|
||||
package integration_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"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 queueContainerConfig struct {
|
||||
Queue queueConfig
|
||||
DB *dbConfig
|
||||
Network *testcontainers.DockerNetwork
|
||||
}
|
||||
|
||||
func createQueueContainer(t *testing.T, ctx context.Context, config *queueContainerConfig) testcontainers.Container {
|
||||
queueCredentials, err := config.Queue.Credentials.Retrieve(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
req := testcontainers.ContainerRequest{
|
||||
Image: "queryorchestration_queue:latest",
|
||||
Env: map[string]string{
|
||||
"QUEUE_URL": config.Queue.URL,
|
||||
"AWS_DEFAULT_REGION": config.Queue.Region,
|
||||
"AWS_REGION": config.Queue.Region,
|
||||
"AWS_ACCESS_KEY_ID": queueCredentials.AccessKeyID,
|
||||
"AWS_SECRET_ACCESS_KEY": queueCredentials.SecretAccessKey,
|
||||
"AWS_SESSION_TOKEN": queueCredentials.SessionToken,
|
||||
"AWS_ENDPOINT_URL_SQS": config.Queue.Endpoint,
|
||||
"DB_USER": config.DB.User,
|
||||
"DB_PASS": config.DB.Password,
|
||||
"DB_HOST": config.DB.Host,
|
||||
"DB_NAME": config.DB.Name,
|
||||
"DB_PORT": strconv.Itoa(config.DB.Port),
|
||||
},
|
||||
Networks: []string{config.Network.Name},
|
||||
}
|
||||
|
||||
container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
|
||||
ContainerRequest: req,
|
||||
Started: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start container: %v", err)
|
||||
}
|
||||
|
||||
return container
|
||||
}
|
||||
|
||||
type queueConfig struct {
|
||||
URL string
|
||||
Region string
|
||||
Endpoint string
|
||||
Credentials aws.CredentialsProvider
|
||||
}
|
||||
|
||||
type queue struct {
|
||||
Container *testcontainers.Container
|
||||
Client *sqs.Client
|
||||
Config *queueConfig
|
||||
}
|
||||
|
||||
func createQueue(t *testing.T, ctx context.Context, network *testcontainers.DockerNetwork) *queue {
|
||||
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:latest",
|
||||
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",
|
||||
},
|
||||
Networks: []string{network.Name},
|
||||
NetworkAliases: map[string][]string{
|
||||
network.Name: {alias},
|
||||
},
|
||||
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)
|
||||
}
|
||||
|
||||
endpoint = fmt.Sprintf("http://%s:%s", alias, port.Port())
|
||||
queueC := &queueConfig{
|
||||
URL: *queueM.QueueUrl,
|
||||
Region: region,
|
||||
Credentials: provider,
|
||||
Endpoint: endpoint,
|
||||
}
|
||||
|
||||
return &queue{
|
||||
Container: &container,
|
||||
Client: client,
|
||||
Config: queueC,
|
||||
}
|
||||
}
|
||||
|
||||
type message struct {
|
||||
URL string
|
||||
Type string
|
||||
Body string
|
||||
}
|
||||
|
||||
func sendMessage(t *testing.T, ctx context.Context, queue *sqs.Client, message message) {
|
||||
_, err := queue.SendMessage(ctx, &sqs.SendMessageInput{
|
||||
MessageAttributes: map[string]types.MessageAttributeValue{
|
||||
"type": {
|
||||
DataType: aws.String("String"),
|
||||
StringValue: aws.String(message.Type),
|
||||
},
|
||||
},
|
||||
QueueUrl: aws.String(message.URL),
|
||||
MessageBody: aws.String(message.Body),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to send message: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func assertMessageWait(t *testing.T, ctx context.Context, queue *queue, msgType string) {
|
||||
time.Sleep(time.Second)
|
||||
|
||||
result, err := queue.Client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
|
||||
QueueUrl: &queue.Config.URL,
|
||||
MaxNumberOfMessages: 1,
|
||||
WaitTimeSeconds: 5,
|
||||
VisibilityTimeout: 2,
|
||||
MessageAttributeNames: []string{
|
||||
"type",
|
||||
},
|
||||
})
|
||||
|
||||
assert.Nil(t, err)
|
||||
for _, message := range result.Messages {
|
||||
assert.Equal(t, msgType, *message.MessageAttributes["type"].StringValue)
|
||||
}
|
||||
}
|
||||
|
||||
func createQueueDependencies(t *testing.T, ctx context.Context) (*queue, func()) {
|
||||
network := createNetwork(t, ctx)
|
||||
|
||||
dbconfig, dbContainer := createDB(t, ctx, network)
|
||||
|
||||
queue := createQueue(t, ctx, network)
|
||||
|
||||
container := createQueueContainer(t, ctx, &queueContainerConfig{
|
||||
Queue: *queue.Config,
|
||||
DB: dbconfig,
|
||||
Network: network,
|
||||
})
|
||||
|
||||
return queue, func() {
|
||||
testcontainers.CleanupNetwork(t, network)
|
||||
dbContainer.Terminate(ctx)
|
||||
(*queue.Container).Terminate(ctx)
|
||||
container.Terminate(ctx)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user