Testing

Testing AWS SQS and SNS: Unit Tests, Moto, and Integration Strategies

HelpMeTest

02 Aug 2026 — 4 min read

AWS SQS and SNS are foundational to event-driven architectures on AWS. Testing code that uses them requires a layered approach: fast unit tests with mocked clients, integration tests against LocalStack or a real AWS sandbox account, and end-to-end tests for Lambda triggers.

Layer 1: Unit Testing with moto

moto mocks AWS services in Python tests. No LocalStack, no AWS account, no network.

pip install moto[sqs,sns] boto3 pytest

import boto3
import pytest
import json
from moto import mock_sqs, mock_sns
from unittest.mock import patch

# The class under test
class OrderNotificationService:
    def __init__(self, sqs_client, sns_client):
        self.sqs = sqs_client
        self.sns = sns_client
    
    def enqueue_order(self, queue_url, order):
        self.sqs.send_message(
            QueueUrl=queue_url,
            MessageBody=json.dumps(order),
            MessageGroupId=order['customer_id'],  # FIFO queue
            MessageDeduplicationId=order['order_id']
        )
    
    def broadcast_order_shipped(self, topic_arn, order_id, tracking_number):
        self.sns.publish(
            TopicArn=topic_arn,
            Message=json.dumps({
                'order_id': order_id,
                'tracking_number': tracking_number,
                'status': 'shipped'
            }),
            Subject='Order Shipped',
            MessageAttributes={
                'event_type': {
                    'DataType': 'String',
                    'StringValue': 'ORDER_SHIPPED'
                }
            }
        )

@mock_sqs
def test_enqueue_order_sends_to_queue():
    sqs = boto3.client('sqs', region_name='us-east-1')
    sns = boto3.client('sns', region_name='us-east-1')
    
    # Create test queue
    queue_url = sqs.create_queue(
        QueueName='orders.fifo',
        Attributes={
            'FifoQueue': 'true',
            'ContentBasedDeduplication': 'false'
        }
    )['QueueUrl']
    
    service = OrderNotificationService(sqs, sns)
    
    order = {
        'order_id': 'ord-001',
        'customer_id': 'cust-123',
        'items': ['item-a', 'item-b'],
        'total': 49.99
    }
    
    service.enqueue_order(queue_url, order)
    
    # Verify message is in the queue
    messages = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=1,
        AttributeNames=['MessageGroupId', 'MessageDeduplicationId']
    ).get('Messages', [])
    
    assert len(messages) == 1
    received_order = json.loads(messages[0]['Body'])
    assert received_order['order_id'] == 'ord-001'
    assert received_order['total'] == 49.99

@mock_sns
@mock_sqs
def test_broadcast_order_shipped():
    sqs = boto3.client('sqs', region_name='us-east-1')
    sns = boto3.client('sns', region_name='us-east-1')
    
    # Create topic
    topic_arn = sns.create_topic(Name='order-events')['TopicArn']
    
    # Subscribe a queue to the topic (fan-out pattern)
    queue_url = sqs.create_queue(QueueName='order-events-subscriber')['QueueUrl']
    queue_arn = sqs.get_queue_attributes(
        QueueUrl=queue_url,
        AttributeNames=['QueueArn']
    )['Attributes']['QueueArn']
    
    sns.subscribe(TopicArn=topic_arn, Protocol='sqs', Endpoint=queue_arn)
    
    service = OrderNotificationService(sqs, sns)
    service.broadcast_order_shipped(topic_arn, 'ord-001', 'TRACK123')
    
    # SNS delivers to the subscribed SQS queue
    messages = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=1
    ).get('Messages', [])
    
    assert len(messages) == 1
    # SNS wraps the message in an envelope
    sns_envelope = json.loads(messages[0]['Body'])
    payload = json.loads(sns_envelope['Message'])
    
    assert payload['tracking_number'] == 'TRACK123'
    assert payload['status'] == 'shipped'

Testing Lambda SQS Triggers

Lambda functions triggered by SQS receive events in a specific format. Test the handler with real SQS event payloads.

# order_processor_lambda.py
import json
import boto3

def handler(event, context):
    """Lambda handler for SQS-triggered order processing"""
    results = []
    
    for record in event['Records']:
        body = json.loads(record['body'])
        
        try:
            process_order(body)
            results.append({'id': record['messageId'], 'status': 'success'})
        except InvalidOrderError as e:
            # Return as a batch item failure — SQS will not retry this message
            results.append({
                'itemIdentifier': record['messageId']
            })
    
    # Report partial batch failures
    failed = [r['itemIdentifier'] for r in results if 'itemIdentifier' in r]
    return {'batchItemFailures': [{'itemIdentifier': id} for id in failed]}

# test_order_processor_lambda.py
import pytest
from unittest.mock import patch, MagicMock
import json

SQS_EVENT_TEMPLATE = {
    "Records": [
        {
            "messageId": "msg-001",
            "receiptHandle": "handle-001",
            "body": "",  # filled per test
            "attributes": {
                "ApproximateReceiveCount": "1",
                "SentTimestamp": "1703001600000",
                "SenderId": "AROAIWPX5BD2BHG722MW4",
                "ApproximateFirstReceiveTimestamp": "1703001600001"
            },
            "messageAttributes": {},
            "md5OfBody": "",
            "eventSource": "aws:sqs",
            "eventSourceARN": "arn:aws:sqs:us-east-1:123456789:orders",
            "awsRegion": "us-east-1"
        }
    ]
}

def make_sqs_event(body_dict, message_id="msg-001"):
    import copy
    event = copy.deepcopy(SQS_EVENT_TEMPLATE)
    event['Records'][0]['body'] = json.dumps(body_dict)
    event['Records'][0]['messageId'] = message_id
    return event

@patch('order_processor_lambda.process_order')
def test_successful_processing_returns_empty_failures(mock_process):
    mock_process.return_value = None  # success
    
    event = make_sqs_event({'order_id': 'ord-001', 'total': 49.99})
    result = handler(event, context={})
    
    assert result['batchItemFailures'] == []
    mock_process.assert_called_once()

@patch('order_processor_lambda.process_order')
def test_invalid_order_reported_as_batch_item_failure(mock_process):
    mock_process.side_effect = InvalidOrderError("Missing required field: total")
    
    event = make_sqs_event({'order_id': 'ord-002'}, message_id='msg-002')
    result = handler(event, context={})
    
    assert len(result['batchItemFailures']) == 1
    assert result['batchItemFailures'][0]['itemIdentifier'] == 'msg-002'

@patch('order_processor_lambda.process_order')
def test_partial_batch_failure_with_multiple_messages(mock_process):
    """Mix of success and failure in the same batch"""
    valid_order = {'order_id': 'ord-001', 'total': 49.99}
    invalid_order = {'order_id': 'ord-002'}  # missing total
    
    def side_effect(order):
        if order.get('total') is None:
            raise InvalidOrderError("Missing total")
    
    mock_process.side_effect = side_effect
    
    event = {
        "Records": [
            {**SQS_EVENT_TEMPLATE['Records'][0], 
             'body': json.dumps(valid_order), 'messageId': 'msg-001'},
            {**SQS_EVENT_TEMPLATE['Records'][0],
             'body': json.dumps(invalid_order), 'messageId': 'msg-002'}
        ]
    }
    
    result = handler(event, context={})
    
    assert len(result['batchItemFailures']) == 1
    assert result['batchItemFailures'][0]['itemIdentifier'] == 'msg-002'

SNS filter policies route messages to specific subscribers based on message attributes. Test that your filters work.

@mock_sns
@mock_sqs
def test_sns_filter_policy_routes_only_matching_messages():
    sqs = boto3.client('sqs', region_name='us-east-1')
    sns = boto3.client('sns', region_name='us-east-1')
    
    topic_arn = sns.create_topic(Name='order-events')['TopicArn']
    
    # Queue 1: only ORDER_SHIPPED events
    shipped_queue_url = sqs.create_queue(QueueName='shipped-orders')['QueueUrl']
    shipped_queue_arn = sqs.get_queue_attributes(
        QueueUrl=shipped_queue_url, AttributeNames=['QueueArn']
    )['Attributes']['QueueArn']
    
    # Queue 2: only ORDER_CANCELLED events
    cancelled_queue_url = sqs.create_queue(QueueName='cancelled-orders')['QueueUrl']
    cancelled_queue_arn = sqs.get_queue_attributes(
        QueueUrl=cancelled_queue_url, AttributeNames=['QueueArn']
    )['Attributes']['QueueArn']
    
    # Subscribe with filter policies
    sns.subscribe(
        TopicArn=topic_arn,
        Protocol='sqs',
        Endpoint=shipped_queue_arn,
        Attributes={
            'FilterPolicy': json.dumps({'event_type': ['ORDER_SHIPPED']})
        }
    )
    
    sns.subscribe(
        TopicArn=topic_arn,
        Protocol='sqs',
        Endpoint=cancelled_queue_arn,
        Attributes={
            'FilterPolicy': json.dumps({'event_type': ['ORDER_CANCELLED']})
        }
    )
    
    # Publish shipped event
    sns.publish(
        TopicArn=topic_arn,
        Message=json.dumps({'order_id': 'ord-001', 'status': 'shipped'}),
        MessageAttributes={
            'event_type': {'DataType': 'String', 'StringValue': 'ORDER_SHIPPED'}
        }
    )
    
    # Shipped queue should have the message
    shipped_msgs = sqs.receive_message(QueueUrl=shipped_queue_url).get('Messages', [])
    assert len(shipped_msgs) == 1
    
    # Cancelled queue should NOT have the message
    cancelled_msgs = sqs.receive_message(QueueUrl=cancelled_queue_url).get('Messages', [])
    assert len(cancelled_msgs) == 0

Integration Testing Against LocalStack

For tests that need real AWS behavior (timing, DLQ routing, FIFO ordering), use LocalStack.

# conftest.py
import pytest
import boto3

@pytest.fixture(scope='session')
def localstack_endpoint():
    return 'http://localhost:4566'

@pytest.fixture(scope='session')
def sqs_client(localstack_endpoint):
    return boto3.client(
        'sqs',
        region_name='us-east-1',
        endpoint_url=localstack_endpoint,
        aws_access_key_id='test',
        aws_secret_access_key='test'
    )

Start LocalStack before the test session:

docker run --rm -p 4566:4566 localstack/localstack

Or with docker-compose:

services:
  localstack:
    image: localstack/localstack
    ports:
      - "4566:4566"
    environment:
      - SERVICES=sqs,sns

The layered strategy — moto for unit tests, LocalStack for integration — keeps your fast unit tests truly fast while giving you confidence that AWS-specific behavior (like FIFO ordering guarantees and DLQ routing) works correctly.

Testing AWS SQS and SNS: Unit Tests, Moto, and Integration Strategies

HelpMeTest

Layer 1: Unit Testing with moto

Testing Lambda SQS Triggers

Integration Testing Against LocalStack

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

Layer 1: Unit Testing with moto

Testing Lambda SQS Triggers

Testing SNS Filter Policies

Integration Testing Against LocalStack

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