jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/s3integration (pull request #47)

Browse Source 
Set Up S3 integration

* cfginterfaceandfirsts3funcs

* addlocalstack

* generallypassesfullsuite

* addedmultipleattemptedpings

* cleanup

* stabiliseplusskip
This commit is contained in:
Michael McGuinness
2025-02-05 12:52:41 +00:00
parent 9254b91d45
commit 92334ad1dd
838 changed files with 139249 additions and 6671 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/test/queue.go
+9 -114
View File
@@ -2,128 +2,25 @@ package test
import (
"context"
"fmt"
"queryorchestration/internal/serviceconfig"
"queryorchestration/internal/serviceconfig/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 queueContainerConfig struct {
Container testcontainers.Container
ExternalEndpoint string
NetworkEndpoint string
}
type CreateQueueConfig struct {
Network *testcontainers.DockerNetwork
Cfg *serviceconfig.BaseConfig
}
func createQueueContainer(t *testing.T, ctx context.Context, cfg *CreateQueueConfig) (*queueContainerConfig, func()) {
alias := "localstack"
provider := credentials.NewStaticCredentialsProvider(cfg.Cfg.AWSKeyID, cfg.Cfg.AWSSecretKey, cfg.Cfg.AWSSessionToken)
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_REGION": cfg.Cfg.AWSRegion,
"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,
func CreateQueue(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, name string) string {
queueM, err := cfg.GetQueueClient().CreateQueue(ctx, &sqs.CreateQueueInput{
QueueName: aws.String(name),
})
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)
}
extEndpoint := fmt.Sprintf("http://%s:%s", host, mappedPort.Port())
t.Setenv("AWS_ENDPOINT_URL_SQS", extEndpoint)
cfg.Cfg.AWSEndpointUrlSQS = extEndpoint
var endpoint string
if cfg.Network != nil {
endpoint = fmt.Sprintf("http://%s:%s", alias, port.Port())
cfg.Cfg.AWSEndpointUrlSQS = endpoint
}
return &queueContainerConfig{
Container: container,
ExternalEndpoint: extEndpoint,
NetworkEndpoint: endpoint,
}, func() {
err := container.Terminate(ctx)
if err != nil {
t.Error(err)
}
}
}
func CreateQueueClient(t *testing.T, ctx context.Context, cfg *CreateQueueConfig) func() {
qcfg, clean := createQueueContainer(t, ctx, cfg)
provider := credentials.NewStaticCredentialsProvider(cfg.Cfg.AWSKeyID, cfg.Cfg.AWSSecretKey, cfg.Cfg.AWSSessionToken)
sqsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(cfg.Cfg.AWSRegion), config.WithCredentialsProvider(provider), config.WithBaseEndpoint(qcfg.ExternalEndpoint))
if err != nil {
t.Fatal(err)
}
cfg.Cfg.QueueClient = sqs.NewFromConfig(sqsCfg)
return clean
}
func CreateQueue(t *testing.T, ctx context.Context, cfg *serviceconfig.BaseConfig, name string) string {
queueM, err := cfg.QueueClient.CreateQueue(ctx, &sqs.CreateQueueInput{
QueueName: aws.String(name),
})
err = cfg.PingQueueByURL(ctx, *queueM.QueueUrl)
if err != nil {
t.Fatal(err)
}
@@ -131,9 +28,7 @@ func CreateQueue(t *testing.T, ctx context.Context, cfg *serviceconfig.BaseConfi
return *queueM.QueueUrl
}
func AssertMessageWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, params *queue.ReceiveParams) types.Message {
time.Sleep(1 * time.Second)
func AssertMessage(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, params *queue.ReceiveParams) types.Message {
result, err := cfg.ReceiveFromQueue(ctx, params)
assert.NoError(t, err)
assert.NotNil(t, result.Messages)
@@ -143,16 +38,16 @@ func AssertMessageWait(t *testing.T, ctx context.Context, cfg serviceconfig.Conf
return result.Messages[0]
}
func AssertMessageBodyWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body string) {
message := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{
func AssertMessageBody(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body string) {
message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{
QueueURL: url,
})
assert.Equal(t, body, *message.Body)
}
func AssertMessageAttrWait(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value string) {
message := AssertMessageWait(t, ctx, cfg, &queue.ReceiveParams{
func AssertMessageAttr(t *testing.T, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value string) {
message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{
QueueURL: url,
Attributes: []string{name},
})