Testing

AWS SQS and SNS Testing: LocalStack, Mocking, and End-to-End Async Flows

HelpMeTest

01 Aug 2026 — 6 min read

AWS SQS and SNS are everywhere in serverless and microservice architectures. They're also surprisingly easy to test badly. This guide covers the three layers of SQS/SNS testing: unit tests with SDK mocks, integration tests with LocalStack, and end-to-end tests for complete async flows.

The Three Testing Layers

Unit tests — test your message handler logic with mocked AWS SDK calls. Fast, no infrastructure, but won't catch IAM policy issues or serialization bugs.

Integration tests — run against LocalStack (a local AWS emulator). Test real SQS/SNS behavior without AWS costs or network calls.

End-to-end tests — run against real AWS in a test account. Most realistic, but slow and requires real AWS credentials.

Start with all three if you can afford it. If you can only do two: unit + LocalStack integration.

Setting Up LocalStack with Testcontainers

LocalStack supports a huge portion of AWS services including SQS, SNS, S3, Lambda, and DynamoDB. With Testcontainers, you get a fresh LocalStack instance per test run:

@Testcontainers
class LocalStackBaseTest {

    @Container
    static LocalStackContainer localstack = new LocalStackContainer(
        DockerImageName.parse("localstack/localstack:3.0")
    )
    .withServices(Service.SQS, Service.SNS);

    protected SqsClient sqsClient() {
        return SqsClient.builder()
            .endpointOverride(localstack.getEndpointOverride(Service.SQS))
            .credentialsProvider(StaticCredentialsProvider.create(
                AwsBasicCredentials.create("test", "test")
            ))
            .region(Region.US_EAST_1)
            .build();
    }

    protected SnsClient snsClient() {
        return SnsClient.builder()
            .endpointOverride(localstack.getEndpointOverride(Service.SNS))
            .credentialsProvider(StaticCredentialsProvider.create(
                AwsBasicCredentials.create("test", "test")
            ))
            .region(Region.US_EAST_1)
            .build();
    }
}

SQS Integration Tests

Basic Send and Receive

class SqsIntegrationTest extends LocalStackBaseTest {

    private String queueUrl;
    private SqsClient sqs;

    @BeforeEach
    void setup() {
        sqs = sqsClient();
        CreateQueueResponse queue = sqs.createQueue(
            CreateQueueRequest.builder().queueName("test-queue").build()
        );
        queueUrl = queue.queueUrl();
    }

    @Test
    void shouldSendAndReceiveMessage() {
        sqs.sendMessage(SendMessageRequest.builder()
            .queueUrl(queueUrl)
            .messageBody("{\"event\":\"ORDER_PLACED\",\"orderId\":\"123\"}")
            .messageAttributes(Map.of(
                "eventType", MessageAttributeValue.builder()
                    .dataType("String")
                    .stringValue("ORDER_PLACED")
                    .build()
            ))
            .build()
        );

        ReceiveMessageResponse response = sqs.receiveMessage(
            ReceiveMessageRequest.builder()
                .queueUrl(queueUrl)
                .messageAttributeNames("All")
                .maxNumberOfMessages(1)
                .waitTimeSeconds(5) // Long polling
                .build()
        );

        assertThat(response.messages()).hasSize(1);
        Message msg = response.messages().get(0);
        assertThat(msg.body()).contains("ORDER_PLACED");
        assertThat(msg.messageAttributes()).containsKey("eventType");
    }
}

Testing FIFO Queues

FIFO queues guarantee ordering and deduplication — critical behavior to test:

@Test
void shouldPreserveMessageOrderInFifoQueue() {
    sqs.createQueue(CreateQueueRequest.builder()
        .queueName("orders.fifo")
        .attributes(Map.of(
            QueueAttributeName.FIFO_QUEUE, "true",
            QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "true"
        ))
        .build()
    );
    String fifoUrl = sqs.getQueueUrl(
        GetQueueUrlRequest.builder().queueName("orders.fifo").build()
    ).queueUrl();

    // Send messages with same MessageGroupId to ensure ordering
    for (int i = 1; i <= 5; i++) {
        sqs.sendMessage(SendMessageRequest.builder()
            .queueUrl(fifoUrl)
            .messageBody("message-" + i)
            .messageGroupId("order-group")
            .messageDeduplicationId("dedup-" + i)
            .build()
        );
    }

    List<String> received = new ArrayList<>();
    for (int i = 0; i < 5; i++) {
        List<Message> messages = sqs.receiveMessage(
            ReceiveMessageRequest.builder()
                .queueUrl(fifoUrl)
                .maxNumberOfMessages(1)
                .build()
        ).messages();
        if (!messages.isEmpty()) {
            received.add(messages.get(0).body());
            sqs.deleteMessage(DeleteMessageRequest.builder()
                .queueUrl(fifoUrl)
                .receiptHandle(messages.get(0).receiptHandle())
                .build()
            );
        }
    }

    assertThat(received).containsExactly(
        "message-1", "message-2", "message-3", "message-4", "message-5"
    );
}

Testing Dead-Letter Queues

@Test
void shouldMoveToDeadLetterAfterMaxReceives() throws Exception {
    // Create DLQ first
    String dlqUrl = sqs.createQueue(
        CreateQueueRequest.builder().queueName("orders-dlq").build()
    ).queueUrl();
    String dlqArn = sqs.getQueueAttributes(
        GetQueueAttributesRequest.builder()
            .queueUrl(dlqUrl)
            .attributeNames(QueueAttributeName.QUEUE_ARN)
            .build()
    ).attributes().get(QueueAttributeName.QUEUE_ARN);

    // Create main queue with redrive policy
    String redrivePolicy = String.format(
        "{\"maxReceiveCount\":\"2\",\"deadLetterTargetArn\":\"%s\"}", dlqArn
    );
    String mainUrl = sqs.createQueue(CreateQueueRequest.builder()
        .queueName("orders-main")
        .attributes(Map.of(
            QueueAttributeName.REDRIVE_POLICY, redrivePolicy,
            QueueAttributeName.VISIBILITY_TIMEOUT, "1" // 1 second for fast testing
        ))
        .build()
    ).queueUrl();

    // Send a message
    sqs.sendMessage(SendMessageRequest.builder()
        .queueUrl(mainUrl)
        .messageBody("poison-pill")
        .build()
    );

    // Receive twice without deleting (simulating failed processing)
    for (int i = 0; i < 2; i++) {
        List<Message> messages = sqs.receiveMessage(
            ReceiveMessageRequest.builder()
                .queueUrl(mainUrl)
                .maxNumberOfMessages(1)
                .build()
        ).messages();
        assertThat(messages).hasSize(1);
        // Don't delete — message becomes visible again after visibility timeout
        Thread.sleep(1500); // Wait for visibility timeout
    }

    // After maxReceiveCount, message should be in DLQ
    await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
        List<Message> dlqMessages = sqs.receiveMessage(
            ReceiveMessageRequest.builder()
                .queueUrl(dlqUrl)
                .maxNumberOfMessages(1)
                .build()
        ).messages();
        assertThat(dlqMessages).hasSize(1);
        assertThat(dlqMessages.get(0).body()).isEqualTo("poison-pill");
    });
}

Fan-Out to Multiple SQS Queues

SNS fan-out is one of the most common patterns — and one of the most broken in test environments:

class SnsIntegrationTest extends LocalStackBaseTest {

    @Test
    void shouldFanOutToAllSubscribedQueues() {
        SnsClient sns = snsClient();
        SqsClient sqs = sqsClient();

        // Create topic
        String topicArn = sns.createTopic(
            CreateTopicRequest.builder().name("order-events").build()
        ).topicArn();

        // Create two subscriber queues
        String q1Url = sqs.createQueue(
            CreateQueueRequest.builder().queueName("inventory-queue").build()
        ).queueUrl();
        String q2Url = sqs.createQueue(
            CreateQueueRequest.builder().queueName("notification-queue").build()
        ).queueUrl();

        // Get queue ARNs
        String q1Arn = getQueueArn(sqs, q1Url);
        String q2Arn = getQueueArn(sqs, q2Url);

        // Subscribe queues to topic
        sns.subscribe(SubscribeRequest.builder()
            .topicArn(topicArn)
            .protocol("sqs")
            .endpoint(q1Arn)
            .attributes(Map.of("RawMessageDelivery", "true"))
            .build()
        );
        sns.subscribe(SubscribeRequest.builder()
            .topicArn(topicArn)
            .protocol("sqs")
            .endpoint(q2Arn)
            .attributes(Map.of("RawMessageDelivery", "true"))
            .build()
        );

        // Publish to topic
        sns.publish(PublishRequest.builder()
            .topicArn(topicArn)
            .message("{\"event\":\"ORDER_PLACED\",\"orderId\":\"789\"}")
            .build()
        );

        // Both queues should receive the message
        assertQueueReceivesMessage(sqs, q1Url, "ORDER_PLACED");
        assertQueueReceivesMessage(sqs, q2Url, "ORDER_PLACED");
    }

    private void assertQueueReceivesMessage(SqsClient sqs, String queueUrl, String expected) {
        await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
            List<Message> messages = sqs.receiveMessage(
                ReceiveMessageRequest.builder()
                    .queueUrl(queueUrl)
                    .maxNumberOfMessages(1)
                    .waitTimeSeconds(1)
                    .build()
            ).messages();
            assertThat(messages).hasSize(1);
            assertThat(messages.get(0).body()).contains(expected);
        });
    }
}

Filter policies let subscribers receive only relevant messages. Test them — they're easy to misconfigure:

@Test
void shouldFilterMessagesByAttribute() {
    String topicArn = sns.createTopic(
        CreateTopicRequest.builder().name("orders").build()
    ).topicArn();

    String priorityQueueUrl = sqs.createQueue(
        CreateQueueRequest.builder().queueName("priority-orders").build()
    ).queueUrl();
    String standardQueueUrl = sqs.createQueue(
        CreateQueueRequest.builder().queueName("standard-orders").build()
    ).queueUrl();

    // Subscribe priority queue with filter for tier=premium
    sns.subscribe(SubscribeRequest.builder()
        .topicArn(topicArn)
        .protocol("sqs")
        .endpoint(getQueueArn(sqs, priorityQueueUrl))
        .attributes(Map.of(
            "FilterPolicy", "{\"tier\":[\"premium\"]}",
            "RawMessageDelivery", "true"
        ))
        .build()
    );

    // Subscribe standard queue for everything
    sns.subscribe(SubscribeRequest.builder()
        .topicArn(topicArn)
        .protocol("sqs")
        .endpoint(getQueueArn(sqs, standardQueueUrl))
        .attributes(Map.of("RawMessageDelivery", "true"))
        .build()
    );

    // Publish premium order
    sns.publish(PublishRequest.builder()
        .topicArn(topicArn)
        .message("premium order message")
        .messageAttributes(Map.of(
            "tier", MessageAttributeValue.builder()
                .dataType("String")
                .stringValue("premium")
                .build()
        ))
        .build()
    );

    // Publish standard order
    sns.publish(PublishRequest.builder()
        .topicArn(topicArn)
        .message("standard order message")
        .messageAttributes(Map.of(
            "tier", MessageAttributeValue.builder()
                .dataType("String")
                .stringValue("standard")
                .build()
        ))
        .build()
    );

    // Priority queue: only premium message
    List<Message> priorityMessages = receiveAll(sqs, priorityQueueUrl);
    assertThat(priorityMessages).hasSize(1);
    assertThat(priorityMessages.get(0).body()).contains("premium");

    // Standard queue: both messages
    List<Message> standardMessages = receiveAll(sqs, standardQueueUrl);
    assertThat(standardMessages).hasSize(2);
}

Unit Testing with SDK Mocks

For fast unit tests of your message handling logic:

class OrderEventHandlerTest {

    private SqsClient sqsClient;
    private OrderEventHandler handler;
    private OrderRepository repository;

    @BeforeEach
    void setup() {
        sqsClient = mock(SqsClient.class);
        repository = mock(OrderRepository.class);
        handler = new OrderEventHandler(sqsClient, repository);
    }

    @Test
    void shouldProcessOrderPlacedEvent() {
        String messageBody = "{\"eventType\":\"ORDER_PLACED\",\"orderId\":\"123\",\"amount\":99.99}";
        Message message = Message.builder()
            .body(messageBody)
            .receiptHandle("receipt-handle-123")
            .build();

        when(sqsClient.receiveMessage(any(ReceiveMessageRequest.class)))
            .thenReturn(ReceiveMessageResponse.builder()
                .messages(List.of(message))
                .build());

        handler.processMessages("https://sqs.us-east-1.amazonaws.com/123/orders");

        verify(repository).save(argThat(order -> 
            order.getOrderId().equals("123") && order.getAmount() == 99.99
        ));
        verify(sqsClient).deleteMessage(argThat(req ->
            req.receiptHandle().equals("receipt-handle-123")
        ));
    }

    @Test
    void shouldNotDeleteMessageOnProcessingFailure() {
        String messageBody = "invalid-json";
        Message message = Message.builder()
            .body(messageBody)
            .receiptHandle("bad-receipt")
            .build();

        when(sqsClient.receiveMessage(any())).thenReturn(
            ReceiveMessageResponse.builder().messages(List.of(message)).build()
        );
        when(repository.save(any())).thenThrow(new IllegalArgumentException("bad data"));

        assertThatThrownBy(() -> 
            handler.processMessages("https://sqs.us-east-1.amazonaws.com/123/orders")
        ).isInstanceOf(MessageProcessingException.class);

        // Message should NOT be deleted — let it become visible again
        verify(sqsClient, never()).deleteMessage(any(DeleteMessageRequest.class));
    }
}

Testing AWS Lambda Triggered by SQS

When Lambda is triggered by SQS, you test it differently — with the SQSEvent input object:

class OrderLambdaHandlerTest {

    private OrderLambdaHandler handler;

    @BeforeEach
    void setup() {
        handler = new OrderLambdaHandler(mock(OrderRepository.class));
    }

    @Test
    void shouldProcessBatchOfMessages() {
        SQSEvent event = new SQSEvent();
        SQSEvent.SQSMessage msg1 = createMessage("order-1", "{\"orderId\":\"1\"}");
        SQSEvent.SQSMessage msg2 = createMessage("order-2", "{\"orderId\":\"2\"}");
        event.setRecords(List.of(msg1, msg2));

        SQSBatchResponse response = handler.handleRequest(event, mock(Context.class));

        // All messages processed successfully
        assertThat(response.getBatchItemFailures()).isEmpty();
    }

    @Test
    void shouldReportPartialBatchFailures() {
        SQSEvent event = new SQSEvent();
        event.setRecords(List.of(
            createMessage("good-msg", "{\"orderId\":\"1\"}"),
            createMessage("bad-msg", "not-json")
        ));

        SQSBatchResponse response = handler.handleRequest(event, mock(Context.class));

        assertThat(response.getBatchItemFailures()).hasSize(1);
        assertThat(response.getBatchItemFailures().get(0).getItemIdentifier())
            .isEqualTo("bad-msg");
    }

    private SQSEvent.SQSMessage createMessage(String messageId, String body) {
        SQSEvent.SQSMessage msg = new SQSEvent.SQSMessage();
        msg.setMessageId(messageId);
        msg.setBody(body);
        msg.setReceiptHandle("receipt-" + messageId);
        return msg;
    }
}

Monitoring SQS/SNS in Production

Integration tests catch configuration bugs, but production issues — message age, approximate number of messages not visible, DLQ depth — need continuous monitoring. HelpMeTest monitors your downstream services that consume from SQS/SNS, flagging processing bottlenecks before they become incidents. Try it free at helpmetest.com.

Summary

Use LocalStack via Testcontainers for SQS/SNS integration tests — no real AWS needed
Test FIFO ordering, DLQ redrive policies, and SNS filter policies explicitly
Unit test message handler logic with SDK mocks for fast feedback
For Lambda-SQS, test partial batch failure (report item failures, not full batch failure)
Always test the unhappy path: what happens when processing fails?

AWS SQS and SNS Testing: LocalStack, Mocking, and End-to-End Async Flows

HelpMeTest

The Three Testing Layers

Setting Up LocalStack with Testcontainers

SQS Integration Tests

Basic Send and Receive

Testing FIFO Queues

Testing Dead-Letter Queues

Fan-Out to Multiple SQS Queues

Unit Testing with SDK Mocks

Testing AWS Lambda Triggered by SQS

Monitoring SQS/SNS in Production

Summary

Read more

Test Result Reporting and Failing Fast in CI Pipelines

SOC 2 Evidence Collection Automation: Stop Emailing Screenshots to Your Auditor

PCI-DSS 4.0 Penetration Testing: Scoping, Methodology, and What Assessors Actually Check

CI/CD Pipeline Testing: GitHub Actions, GitLab CI, and Jenkins Patterns

The Three Testing Layers

Setting Up LocalStack with Testcontainers

SQS Integration Tests

Basic Send and Receive

Testing FIFO Queues

Testing Dead-Letter Queues

SNS Integration Tests

Fan-Out to Multiple SQS Queues

Testing SNS Filter Policies

Unit Testing with SDK Mocks

Testing AWS Lambda Triggered by SQS

Monitoring SQS/SNS in Production

Summary

Read more

Test Result Reporting and Failing Fast in CI Pipelines

SOC 2 Evidence Collection Automation: Stop Emailing Screenshots to Your Auditor

PCI-DSS 4.0 Penetration Testing: Scoping, Methodology, and What Assessors Actually Check

CI/CD Pipeline Testing: GitHub Actions, GitLab CI, and Jenkins Patterns