sqs-tests

Overview

Two queue types, per the SQS developer guide:

The semantic difference cascades through every test pattern: a Standard-queue test must assume duplicates can occur; a FIFO test must assume strict ordering.

When to use

• The repo has SQS-publishing or SQS-consuming code (AWS SDK).
• The user writes unit tests against mocked SQS clients.
• The user writes integration tests against LocalStack / real SQS (in a sandbox account).
• A test verifies visibility-timeout, DLQ-routing, or long-polling behavior.

Step 1 - Test approach: mock vs LocalStack vs real

Three approaches, ordered by isolation:

For pure logic tests (does the code call SendMessage with the right body?), use mocks. For semantic tests (does retry-on-failure work end-to-end?), use LocalStack. For pre-prod smoke, use real SQS.

Step 2 - Mock-based unit test (TypeScript)

import { mockClient } from 'aws-sdk-client-mock';
import { SQSClient, SendMessageCommand } from '@aws-sdk/client-sqs';

const sqsMock = mockClient(SQSClient);
beforeEach(() => sqsMock.reset());

it('sends order-placed message to SQS', async () => {
  sqsMock.on(SendMessageCommand).resolves({ MessageId: 'msg-123' });

  await placeOrder({ customerId: 1 });

  expect(sqsMock.commandCalls(SendMessageCommand)).toHaveLength(1);
  expect(sqsMock.commandCalls(SendMessageCommand)[0].args[0].input).toMatchObject({
    QueueUrl: expect.stringContaining('orders'),
    MessageBody: expect.stringContaining('"customerId":1'),
  });
});

Step 3 - Mock-based unit test (Python)

import boto3
from moto import mock_aws

@mock_aws
def test_send_order_message():
    sqs = boto3.client('sqs', region_name='us-east-1')
    queue_url = sqs.create_queue(QueueName='orders')['QueueUrl']

    place_order(customer_id=1)

    response = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=10)
    assert len(response['Messages']) == 1
    body = json.loads(response['Messages'][0]['Body'])
    assert body['customerId'] == 1

moto's @mock_aws decorator intercepts boto3 SQS calls; tests run without network.

Step 4 - LocalStack integration test

# docker-compose.yml
services:
  localstack:
    image: localstack/localstack:latest
    ports: [4566:4566]
    environment:
      SERVICES: sqs

sqs = boto3.client(
    'sqs',
    endpoint_url='http://localhost:4566',
    region_name='us-east-1',
    aws_access_key_id='test', aws_secret_access_key='test',
)
queue_url = sqs.create_queue(QueueName='orders')['QueueUrl']
# ... full SQS API works, including visibility timeouts, DLQ, FIFO

Step 5 - Test visibility-timeout behavior

Visibility timeout keeps an in-flight message invisible to other receivers for a configurable window, preventing duplicate processing (SQS developer guide).

Test pattern (LocalStack or real SQS, NOT mock):

sqs.send_message(QueueUrl=queue_url, MessageBody='test')
msg1 = sqs.receive_message(QueueUrl=queue_url, VisibilityTimeout=30)['Messages'][0]
# Within 30s, the message should be invisible to other receivers:
msg2 = sqs.receive_message(QueueUrl=queue_url)
assert msg2.get('Messages') is None
# After visibility timeout (or explicit ChangeMessageVisibility), it returns:
sqs.change_message_visibility(
    QueueUrl=queue_url,
    ReceiptHandle=msg1['ReceiptHandle'],
    VisibilityTimeout=0,
)
msg3 = sqs.receive_message(QueueUrl=queue_url)
assert msg3['Messages'][0]['MessageId'] == msg1['MessageId']

Step 6 - Test DLQ routing

SQS routes a message to the dead-letter queue after maxReceiveCount failed deliveries (poison-message isolation, per the SQS developer guide). Full LocalStack recipe: references/localstack-recipes.md.

Step 7 - Test FIFO ordering + dedup

FIFO queues (FifoQueue: 'true') preserve per-MessageGroupId order and drop identical bodies within a 5-minute dedup window (ContentBasedDeduplication). Full LocalStack recipe: references/localstack-recipes.md.

Step 8 - Message retention

SQS auto-deletes messages older than the retention period: default 4 days, configurable from 60 seconds to 1,209,600 seconds (14 days) (SQS developer guide). Tests rarely verify retention directly; document the expected retention in queue setup (Terraform / CloudFormation) and review per-team.

Step 9 - CI integration

services:
  localstack:
    image: localstack/localstack:latest
    ports: [4566:4566]
    env: { SERVICES: sqs }

steps:
  - run: pytest tests/integration/sqs/ -v

Anti-patterns

Limitations

• LocalStack ≠ real SQS - some behaviors (extreme latency, AWS throttling) are not emulated.
• FIFO throughput limits (300 messages/sec without high-throughput mode) need real SQS to verify.
• Cost-effective sandbox testing requires per-PR queue cleanup; use random queue-name suffixes + cleanup hooks.

References

• sqs-dg - official SQS developer guide; standard vs FIFO, visibility timeout, DLQ, retention
• docs.aws.amazon.com/AWSSimpleQueueService - full SQS docs
• localstack.cloud - LocalStack SQS emulator
• pypi.org/project/moto - Python AWS mock
• npmjs.com/package/aws-sdk-client-mock - TypeScript AWS mock
• sidekiq-tests, celery-tests, bullmq-tests, rabbitmq-tests - sister tools
• idempotency-test-author - critical companion for at-least-once SQS Standard queues