90 lines
2.4 KiB
Go
90 lines
2.4 KiB
Go
package sqs
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"wucher/internal/config"
|
|
"wucher/internal/queue"
|
|
)
|
|
|
|
func TestEmailQueueIntegration(t *testing.T) {
|
|
queueURL := os.Getenv("SQS_INTEGRATION_QUEUE_URL")
|
|
if queueURL == "" {
|
|
t.Skip("SQS_INTEGRATION_QUEUE_URL is not set")
|
|
}
|
|
|
|
cfg := config.SQSConfig{
|
|
Region: getenv("SQS_INTEGRATION_REGION", os.Getenv("AWS_REGION")),
|
|
Endpoint: getenv("SQS_INTEGRATION_ENDPOINT", os.Getenv("SQS_ENDPOINT")),
|
|
QueueURL: queueURL,
|
|
QueueType: getenv("SQS_INTEGRATION_QUEUE_TYPE", queue.QueueTypeStandard),
|
|
MessageGroupID: getenv("SQS_INTEGRATION_MESSAGE_GROUP_ID", "integration"),
|
|
MaxMessages: 1,
|
|
WaitTimeSeconds: 5,
|
|
VisibilityTimeout: 30 * time.Second,
|
|
AllowInsecureEndpoint: os.Getenv("SQS_INTEGRATION_ALLOW_INSECURE_ENDPOINT") == "true",
|
|
}
|
|
client, err := NewClient(context.Background(), cfg)
|
|
if err != nil {
|
|
t.Fatalf("new client: %v", err)
|
|
}
|
|
adapter, err := NewEmailQueue(client, cfg)
|
|
if err != nil {
|
|
t.Fatalf("new queue: %v", err)
|
|
}
|
|
|
|
serializer := queue.NewJSONSerializer("v1", true, cfg.QueueType, cfg.MessageGroupID)
|
|
outbound, err := serializer.Serialize(context.Background(), queue.EmailJob{
|
|
Type: "integration",
|
|
To: "integration@example.com",
|
|
Subject: "integration-" + time.Now().UTC().Format(time.RFC3339Nano),
|
|
Body: "hello",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("serialize: %v", err)
|
|
}
|
|
if err := adapter.Publish(context.Background(), *outbound); err != nil {
|
|
t.Fatalf("publish: %v", err)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
|
|
for {
|
|
deliveries, err := adapter.Receive(ctx, 1)
|
|
if err != nil {
|
|
t.Fatalf("receive: %v", err)
|
|
}
|
|
if len(deliveries) == 0 {
|
|
if ctx.Err() != nil {
|
|
t.Fatalf("timed out waiting for message")
|
|
}
|
|
continue
|
|
}
|
|
envelope, err := serializer.Deserialize(deliveries[0].Body, deliveries[0].Attributes)
|
|
if err != nil {
|
|
t.Fatalf("deserialize: %v", err)
|
|
}
|
|
if envelope.MessageID != outbound.ID {
|
|
if err := adapter.Retry(context.Background(), deliveries[0], 0); err != nil {
|
|
t.Fatalf("retry unrelated message: %v", err)
|
|
}
|
|
continue
|
|
}
|
|
if err := adapter.Ack(context.Background(), deliveries); err != nil {
|
|
t.Fatalf("ack: %v", err)
|
|
}
|
|
return
|
|
}
|
|
}
|
|
|
|
func getenv(key, fallback string) string {
|
|
if value := os.Getenv(key); value != "" {
|
|
return value
|
|
}
|
|
return fallback
|
|
}
|