17265699519

Committed 27 Aug 2025 11:28AM UTC coverage: 86.827% (-0.01%) from 86.837%

Build # 17265699519

Build Type

push

github

Committed by

web-flow

Commit Message

Fix SQS tests failing due to missing snapshot update after #12957 (#13062)

Coverage Stats

67057 of 77231 relevant lines covered (86.83%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.37

/localstack-core/localstack/utils/aws/dead_letter_queue.py

import json
import logging
import uuid

from localstack.aws.connect import connect_to
from localstack.utils.aws import arns
from localstack.utils.strings import convert_to_printable_chars, first_char_to_upper

LOG = logging.getLogger(__name__)


def sns_error_to_dead_letter_queue(
    sns_subscriber: dict,
    message: str,
    error: str,
    **kwargs,
):
    policy = json.loads(sns_subscriber.get("RedrivePolicy") or "{}")
    target_arn = policy.get("deadLetterTargetArn")
    if not target_arn:
        return
    if not_supported := (
        set(kwargs) - {"MessageAttributes", "MessageGroupId", "MessageDeduplicationId"}
    ):
        LOG.warning(
            "Not publishing to the DLQ - invalid arguments passed to the DLQ '%s'", not_supported
        )
        return
    event = {
        "message": message,
        **kwargs,
    }
    return _send_to_dead_letter_queue(sns_subscriber["SubscriptionArn"], target_arn, event, error)


def _send_to_dead_letter_queue(source_arn: str, dlq_arn: str, event: dict, error, role: str = None):
    if not dlq_arn:
        return
    LOG.info("Sending failed execution %s to dead letter queue %s", source_arn, dlq_arn)
    messages = _prepare_messages_to_dlq(source_arn, event, error)
    source_service = arns.extract_service_from_arn(source_arn)
    region = arns.extract_region_from_arn(dlq_arn)
    if role:
        clients = connect_to.with_assumed_role(
            role_arn=role, service_principal=source_service, region_name=region
        )
    else:
        clients = connect_to(region_name=region)
    if ":sqs:" in dlq_arn:
        queue_url = arns.sqs_queue_url_for_arn(dlq_arn)
        sqs_client = clients.sqs.request_metadata(
            source_arn=source_arn, service_principal=source_service
        )
        error = None
        result_code = None
        try:
            result = sqs_client.send_message_batch(QueueUrl=queue_url, Entries=messages)
            result_code = result.get("ResponseMetadata", {}).get("HTTPStatusCode")
        except Exception as e:
            error = e
        if error or not result_code or result_code >= 400:
            msg = f"Unable to send message to dead letter queue {queue_url} (code {result_code}): {error}"
            if "InvalidMessageContents" in str(error):
                msg += f" - messages: {messages}"
            LOG.info(msg)
            raise Exception(msg)
    elif ":sns:" in dlq_arn:
        sns_client = clients.sns.request_metadata(
            source_arn=source_arn, service_principal=source_service
        )
        for message in messages:
            sns_client.publish(
                TopicArn=dlq_arn,
                Message=message["MessageBody"],
                MessageAttributes=message["MessageAttributes"],
            )
    else:
        LOG.warning("Unsupported dead letter queue type: %s", dlq_arn)
    return dlq_arn


def _prepare_messages_to_dlq(source_arn: str, event: dict, error) -> list[dict]:
    messages = []
    custom_attrs = {
        "RequestID": {"DataType": "String", "StringValue": str(uuid.uuid4())},
        "ErrorCode": {"DataType": "String", "StringValue": "200"},
        "ErrorMessage": {"DataType": "String", "StringValue": str(error)},
    }
    if ":sqs:" in source_arn:
        custom_attrs["ErrorMessage"]["StringValue"] = str(error.result)
        for record in event.get("Records", []):
            msg_attrs = message_attributes_to_upper(record.get("messageAttributes"))
            message_attrs = {**msg_attrs, **custom_attrs}
            messages.append(
                {
                    "Id": record.get("messageId"),
                    "MessageBody": record.get("body"),
                    "MessageAttributes": message_attrs,
                }
            )
    elif ":sns:" in source_arn:
        # event can also contain: MessageAttributes, MessageGroupId, MessageDeduplicationId
        message = {
            "Id": str(uuid.uuid4()),
            "MessageBody": event.pop("message"),
            **event,
        }
        messages.append(message)

    elif ":lambda:" in source_arn:
        custom_attrs["ErrorCode"]["DataType"] = "Number"
        # not sure about what type of error can come here
        try:
            error_message = json.loads(error.result)["errorMessage"]
            custom_attrs["ErrorMessage"]["StringValue"] = error_message
        except (ValueError, KeyError):
            # using old behaviour
            custom_attrs["ErrorMessage"]["StringValue"] = str(error)

        messages.append(
            {
                "Id": str(uuid.uuid4()),
                "MessageBody": json.dumps(event),
                "MessageAttributes": custom_attrs,
            }
        )
    # make sure we only have printable strings in the message attributes
    for message in messages:
        if message.get("MessageAttributes"):
            message["MessageAttributes"] = convert_to_printable_chars(message["MessageAttributes"])
    return messages


def message_attributes_to_upper(message_attrs: dict) -> dict:
    """Convert message attribute details (first characters) to upper case (e.g., StringValue, DataType)."""
    message_attrs = message_attrs or {}
    for _, attr in message_attrs.items():
        if not isinstance(attr, dict):
            continue
        for key, value in dict(attr).items():
            attr[first_char_to_upper(key)] = attr.pop(key)
    return message_attrs

1	import json	1✔
2	import logging	1✔
3	import uuid	1✔
4
5	from localstack.aws.connect import connect_to	1✔
6	from localstack.utils.aws import arns	1✔
7	from localstack.utils.strings import convert_to_printable_chars, first_char_to_upper	1✔
8
9	LOG = logging.getLogger(__name__)	1✔
10
11
12	def sns_error_to_dead_letter_queue(	1✔
13	sns_subscriber: dict,
14	message: str,
15	error: str,
16	**kwargs,
17	):
18	policy = json.loads(sns_subscriber.get("RedrivePolicy") or "{}")	1✔
19	target_arn = policy.get("deadLetterTargetArn")	1✔
20	if not target_arn:	1✔
21	return	1✔
22	if not_supported := (	1✔
23	set(kwargs) - {"MessageAttributes", "MessageGroupId", "MessageDeduplicationId"}
24	):
25	LOG.warning(	×
26	"Not publishing to the DLQ - invalid arguments passed to the DLQ '%s'", not_supported
27	)
28	return	×
29	event = {	1✔
30	"message": message,
31	**kwargs,
32	}
33	return _send_to_dead_letter_queue(sns_subscriber["SubscriptionArn"], target_arn, event, error)	1✔
34
35
36	def _send_to_dead_letter_queue(source_arn: str, dlq_arn: str, event: dict, error, role: str = None):	1✔
37	if not dlq_arn:	1✔
38	return	1✔
39	LOG.info("Sending failed execution %s to dead letter queue %s", source_arn, dlq_arn)	1✔
40	messages = _prepare_messages_to_dlq(source_arn, event, error)	1✔
41	source_service = arns.extract_service_from_arn(source_arn)	1✔
42	region = arns.extract_region_from_arn(dlq_arn)	1✔
43	if role:	1✔
44	clients = connect_to.with_assumed_role(	1✔
45	role_arn=role, service_principal=source_service, region_name=region
46	)
47	else:
48	clients = connect_to(region_name=region)	1✔
49	if ":sqs:" in dlq_arn:	1✔
50	queue_url = arns.sqs_queue_url_for_arn(dlq_arn)	1✔
51	sqs_client = clients.sqs.request_metadata(	1✔
52	source_arn=source_arn, service_principal=source_service
53	)
54	error = None	1✔
55	result_code = None	1✔
56	try:	1✔
57	result = sqs_client.send_message_batch(QueueUrl=queue_url, Entries=messages)	1✔
58	result_code = result.get("ResponseMetadata", {}).get("HTTPStatusCode")	1✔
59	except Exception as e:	×
60	error = e	×
61	if error or not result_code or result_code >= 400:	1✔
62	msg = f"Unable to send message to dead letter queue {queue_url} (code {result_code}): {error}"	×
63	if "InvalidMessageContents" in str(error):	×
64	msg += f" - messages: {messages}"	×
65	LOG.info(msg)	×
66	raise Exception(msg)	×
67	elif ":sns:" in dlq_arn:	1✔
68	sns_client = clients.sns.request_metadata(	1✔
69	source_arn=source_arn, service_principal=source_service
70	)
71	for message in messages:	1✔
72	sns_client.publish(	1✔
73	TopicArn=dlq_arn,
74	Message=message["MessageBody"],
75	MessageAttributes=message["MessageAttributes"],
76	)
77	else:
78	LOG.warning("Unsupported dead letter queue type: %s", dlq_arn)	×
79	return dlq_arn	1✔
80
81
82	def _prepare_messages_to_dlq(source_arn: str, event: dict, error) -> list[dict]:	1✔
83	messages = []	1✔
84	custom_attrs = {	1✔
85	"RequestID": {"DataType": "String", "StringValue": str(uuid.uuid4())},
86	"ErrorCode": {"DataType": "String", "StringValue": "200"},
87	"ErrorMessage": {"DataType": "String", "StringValue": str(error)},
88	}
89	if ":sqs:" in source_arn:	1✔
90	custom_attrs["ErrorMessage"]["StringValue"] = str(error.result)	×
91	for record in event.get("Records", []):	×
92	msg_attrs = message_attributes_to_upper(record.get("messageAttributes"))	×
93	message_attrs = {msg_attrs, custom_attrs}	×
94	messages.append(	×
95	{
96	"Id": record.get("messageId"),
97	"MessageBody": record.get("body"),
98	"MessageAttributes": message_attrs,
99	}
100	)
101	elif ":sns:" in source_arn:	1✔
102	# event can also contain: MessageAttributes, MessageGroupId, MessageDeduplicationId
103	message = {	1✔
104	"Id": str(uuid.uuid4()),
105	"MessageBody": event.pop("message"),
106	**event,
107	}
108	messages.append(message)	1✔
109
110	elif ":lambda:" in source_arn:	1✔
111	custom_attrs["ErrorCode"]["DataType"] = "Number"	1✔
112	# not sure about what type of error can come here
113	try:	1✔
114	error_message = json.loads(error.result)["errorMessage"]	1✔
115	custom_attrs["ErrorMessage"]["StringValue"] = error_message	1✔
116	except (ValueError, KeyError):	×
117	# using old behaviour
118	custom_attrs["ErrorMessage"]["StringValue"] = str(error)	×
119
120	messages.append(	1✔
121	{
122	"Id": str(uuid.uuid4()),
123	"MessageBody": json.dumps(event),
124	"MessageAttributes": custom_attrs,
125	}
126	)
127	# make sure we only have printable strings in the message attributes
128	for message in messages:	1✔
129	if message.get("MessageAttributes"):	1✔
130	message["MessageAttributes"] = convert_to_printable_chars(message["MessageAttributes"])	1✔
131	return messages	1✔
132
133
134	def message_attributes_to_upper(message_attrs: dict) -> dict:	1✔
135	"""Convert message attribute details (first characters) to upper case (e.g., StringValue, DataType)."""
136	message_attrs = message_attrs or {}	×
137	for _, attr in message_attrs.items():	×
138	if not isinstance(attr, dict):	×
139	continue	×
140	for key, value in dict(attr).items():	×
141	attr[first_char_to_upper(key)] = attr.pop(key)	×
142	return message_attrs	×

localstack / localstack / 17265699519

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous