9.3SQS Testing Patterns

Dead Letter Queue Testing

One of the most critical but often untested patterns is the dead letter queue (DLQ). When a message fails processing multiple times, it should land in the DLQ -- not disappear silently.

# tests/integration/test_sqs_processing.py
import boto3
import pytest
import json
from testcontainers.localstack import LocalStackContainer

@pytest.fixture(scope="module")
def localstack():
    with LocalStackContainer("localstack/localstack:3.0") as ls:
        yield ls

@pytest.fixture
def sqs_client(localstack):
    return boto3.client(
        "sqs",
        endpoint_url=localstack.get_url(),
        region_name="us-east-1",
        aws_access_key_id="test",
        aws_secret_access_key="test",
    )

def test_dead_letter_queue_receives_failed_messages(sqs_client):
    """Messages that fail processing N times must land in the DLQ."""
    # Create DLQ
    dlq = sqs_client.create_queue(QueueName="orders-dlq")
    dlq_arn = sqs_client.get_queue_attributes(
        QueueUrl=dlq["QueueUrl"],
        AttributeNames=["QueueArn"]
    )["Attributes"]["QueueArn"]

    # Create main queue with DLQ redrive policy
    main_queue = sqs_client.create_queue(
        QueueName="orders",
        Attributes={
            "RedrivePolicy": json.dumps({
                "deadLetterTargetArn": dlq_arn,
                "maxReceiveCount": "2"
            })
        }
    )

    # Send a message that will fail processing
    sqs_client.send_message(
        QueueUrl=main_queue["QueueUrl"],
        MessageBody='{"orderId": "INVALID"}'
    )

    # Simulate 2 failed receive attempts (receive but don't delete)
    for _ in range(2):
        msgs = sqs_client.receive_message(QueueUrl=main_queue["QueueUrl"])
        # Not deleting = simulating processing failure

    # Verify message lands in DLQ
    import time
    time.sleep(2)
    dlq_msgs = sqs_client.receive_message(QueueUrl=dlq["QueueUrl"])
    assert len(dlq_msgs.get("Messages", [])) == 1
    failed_body = json.loads(dlq_msgs["Messages"][0]["Body"])
    assert failed_body["orderId"] == "INVALID"

Message Ordering and Deduplication (FIFO Queues)

def test_fifo_queue_preserves_message_order(sqs_client):
    """FIFO queues must deliver messages in the order they were sent."""
    queue = sqs_client.create_queue(
        QueueName="orders.fifo",
        Attributes={
            "FifoQueue": "true",
            "ContentBasedDeduplication": "true"
        }
    )

    # Send messages in a specific order
    for i in range(5):
        sqs_client.send_message(
            QueueUrl=queue["QueueUrl"],
            MessageBody=json.dumps({"sequence": i}),
            MessageGroupId="order-group-1",
        )

    # Receive and verify order
    received_sequences = []
    for _ in range(5):
        msgs = sqs_client.receive_message(QueueUrl=queue["QueueUrl"])
        if "Messages" in msgs:
            body = json.loads(msgs["Messages"][0]["Body"])
            received_sequences.append(body["sequence"])
            sqs_client.delete_message(
                QueueUrl=queue["QueueUrl"],
                ReceiptHandle=msgs["Messages"][0]["ReceiptHandle"]
            )

    assert received_sequences == [0, 1, 2, 3, 4]

def test_fifo_deduplication(sqs_client):
    """FIFO queues must deduplicate messages with the same dedup ID."""
    queue = sqs_client.create_queue(
        QueueName="dedup-test.fifo",
        Attributes={"FifoQueue": "true"}
    )

    # Send the same message twice with the same dedup ID
    for _ in range(2):
        sqs_client.send_message(
            QueueUrl=queue["QueueUrl"],
            MessageBody="duplicate message",
            MessageGroupId="group-1",
            MessageDeduplicationId="same-dedup-id",
        )

    # Should only receive one message
    import time
    time.sleep(1)
    msgs = sqs_client.receive_message(
        QueueUrl=queue["QueueUrl"],
        MaxNumberOfMessages=10,
    )
    assert len(msgs.get("Messages", [])) == 1