fb37c2a2-4bd7-49b4-891b-ede18134b4b1

Committed 20 Jan 2025 04:22PM UTC coverage: 86.826% (-0.02%) from 86.844%

Build # fb37c2a2-4bd7-49b4-891b-ede18134b4b1

Build Type

push

circleci

Committed by

web-flow

Commit Message

Add tests for KMS resource tagging and untagging (#12121)

Coverage Stats

61076 of 70343 relevant lines covered (86.83%)

0.87 hits per line

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

98.4

/localstack-core/localstack/services/lambda_/event_source_mapping/pollers/sqs_poller.py

import json
import logging
from collections import defaultdict
from functools import cached_property

from botocore.client import BaseClient

from localstack.aws.api.pipes import PipeSourceSqsQueueParameters
from localstack.aws.api.sqs import MessageSystemAttributeName
from localstack.config import internal_service_url
from localstack.services.lambda_.event_source_mapping.event_processor import (
    EventProcessor,
    PartialBatchFailureError,
)
from localstack.services.lambda_.event_source_mapping.pollers.poller import (
    Poller,
    parse_batch_item_failures,
)
from localstack.services.lambda_.event_source_mapping.senders.sender_utils import batched
from localstack.services.sqs.constants import HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT
from localstack.utils.aws.arns import parse_arn
from localstack.utils.strings import first_char_to_lower

LOG = logging.getLogger(__name__)

DEFAULT_MAX_RECEIVE_COUNT = 10


class SqsPoller(Poller):
    queue_url: str

    def __init__(
        self,
        source_arn: str,
        source_parameters: dict | None = None,
        source_client: BaseClient | None = None,
        processor: EventProcessor | None = None,
    ):
        super().__init__(source_arn, source_parameters, source_client, processor)
        self.queue_url = get_queue_url(self.source_arn)
        self._register_client_hooks()

    @property
    def sqs_queue_parameters(self) -> PipeSourceSqsQueueParameters:
        return self.source_parameters["SqsQueueParameters"]

    @cached_property
    def is_fifo_queue(self) -> bool:
        # Alternative heuristic: self.queue_url.endswith(".fifo"), but we need the call to get_queue_attributes for IAM
        return self.get_queue_attributes().get("FifoQueue", "false").lower() == "true"

    def _register_client_hooks(self):
        event_system = self.source_client.meta.events

        def _handle_receive_message_override(params, context, **kwargs):
            requested_count = params.get("MaxNumberOfMessages")
            if not requested_count or requested_count <= DEFAULT_MAX_RECEIVE_COUNT:
                return

            # Allow overide parameter to be greater than default and less than maximum batch size.
            # Useful for getting remaining records less than the batch size. i.e we need 100 records but BatchSize is 1k.
            override = min(requested_count, self.sqs_queue_parameters["BatchSize"])
            context[HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = str(override)

        def _handle_delete_batch_override(params, context, **kwargs):
            requested_count = len(params.get("Entries", []))
            if not requested_count or requested_count <= DEFAULT_MAX_RECEIVE_COUNT:
                return

            override = min(requested_count, self.sqs_queue_parameters["BatchSize"])
            context[HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = str(override)

        def _handler_inject_header(params, context, **kwargs):
            if override := context.pop(HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT, None):
                params["headers"][HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = override

        event_system.register(
            "provide-client-params.sqs.ReceiveMessage", _handle_receive_message_override
        )
        # Since we delete SQS messages after processing, this allows us to remove up to 10K entries at a time.
        event_system.register(
            "provide-client-params.sqs.DeleteMessageBatch", _handle_delete_batch_override
        )
        event_system.register("before-call.sqs.*", _handler_inject_header)

    def get_queue_attributes(self) -> dict:
        """The API call to sqs:GetQueueAttributes is required for IAM policy streamsing."""
        get_queue_attributes_response = self.source_client.get_queue_attributes(
            QueueUrl=self.queue_url,
            AttributeNames=["FifoQueue"],
        )
        return get_queue_attributes_response.get("Attributes", {})

    def event_source(self) -> str:
        return "aws:sqs"

    def poll_events(self) -> None:
        # SQS pipe source: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
        # "The 9 Ways an SQS Message can be Deleted": https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/
        # TODO: implement batch window expires based on MaximumBatchingWindowInSeconds
        # TODO: implement invocation payload size quota
        # TODO: consider long-polling vs. short-polling trade-off. AWS uses long-polling:
        #  https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html#pipes-sqs-scaling
        response = self.source_client.receive_message(
            QueueUrl=self.queue_url,
            MaxNumberOfMessages=self.sqs_queue_parameters.get(
                "BatchSize", DEFAULT_MAX_RECEIVE_COUNT
            ),
            MessageAttributeNames=["All"],
            MessageSystemAttributeNames=[MessageSystemAttributeName.All],
        )
        if messages := response.get("Messages"):
            LOG.debug("Polled %d events from %s", len(messages), self.source_arn)
            try:
                if self.is_fifo_queue:
                    # TODO: think about starvation behavior because once failing message could block other groups
                    fifo_groups = split_by_message_group_id(messages)
                    for fifo_group_messages in fifo_groups.values():
                        self.handle_messages(fifo_group_messages)
                else:
                    self.handle_messages(messages)

            # TODO: unify exception handling across pollers: should we catch and raise?
            except Exception as e:
                # TODO: improve error messages (produce same failure and design better error messages)
                LOG.warning(
                    "Polling or batch processing failed: %s",
                    e,
                    exc_info=LOG.isEnabledFor(logging.DEBUG),
                )

    def handle_messages(self, messages):
        polled_events = transform_into_events(messages)
        # Filtering: matching vs. discarded (i.e., not matching filter criteria)
        # TODO: implement format detection behavior (e.g., for JSON body):
        #  https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
        #  Check whether we need poller-specific filter-preprocessing here without modifying the actual event!
        # convert to json for filtering (HACK for fixing parity with v1 and getting regression tests passing)
        for event in polled_events:
            try:
                event["body"] = json.loads(event["body"])
            except json.JSONDecodeError:
                LOG.debug(
                    "Unable to convert event body '%s' to json... Event might be dropped.",
                    event["body"],
                )
        matching_events = self.filter_events(polled_events)
        # convert them back (HACK for fixing parity with v1 and getting regression tests passing)
        for event in matching_events:
            event["body"] = (
                json.dumps(event["body"]) if not isinstance(event["body"], str) else event["body"]
            )

        all_message_ids = {message["MessageId"] for message in messages}
        matching_message_ids = {event["messageId"] for event in matching_events}
        discarded_message_ids = all_message_ids.difference(matching_message_ids)
        # Delete discarded events immediately:
        # https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/#7-event-source-mappings-with-filters
        self.delete_messages(messages, discarded_message_ids)

        # Don't trigger upon empty events
        if len(matching_events) == 0:
            return
        # Enrich events with metadata after filtering
        enriched_events = self.add_source_metadata(matching_events)

        # Invoke the processor (e.g., Pipe, ESM) and handle partial batch failures
        try:
            self.processor.process_events_batch(enriched_events)
            successful_message_ids = all_message_ids
        except PartialBatchFailureError as e:
            failed_message_ids = parse_batch_item_failures(
                e.partial_failure_payload, matching_message_ids
            )
            successful_message_ids = matching_message_ids.difference(failed_message_ids)

        # Only delete messages that are processed successfully as described here:
        # https://docs.aws.amazon.com/en_gb/lambda/latest/dg/with-sqs.html
        # When Lambda reads a batch, the messages stay in the queue but are hidden for the length of the queue's
        # visibility timeout. If your function successfully processes the batch, Lambda deletes the messages
        # from the queue. By default, if your function encounters an error while processing a batch,
        # all messages in that batch become visible in the queue again. For this reason, your function code must
        # be able to process the same message multiple times without unintended side effects.
        # Troubleshooting: https://repost.aws/knowledge-center/lambda-sqs-report-batch-item-failures
        # For FIFO queues, AWS also deletes successfully sent messages. Therefore, the AWS docs recommends:
        # "If you're using this feature with a FIFO queue, your function should stop processing messages after the first
        # failure and return all failed and unprocessed messages in batchItemFailures. This helps preserve the ordering
        # of messages in your queue."
        # Following this recommendation could result in the unsolved side effect that valid messages are continuously
        # placed in the same batch as failing messages:
        # * https://stackoverflow.com/questions/78694079/how-to-stop-fifo-sqs-messages-from-being-placed-in-a-batch-with-failing-messages
        # * https://stackoverflow.com/questions/76912394/can-i-report-only-messages-from-failing-group-id-in-reportbatchitemfailures-resp

        # TODO: Test blocking failure behavior for FIFO queues to guarantee strict ordering
        #  -> might require some checkpointing or retry control on the poller side?!
        # The poller should only proceed processing FIFO queues after having retried failing messages:
        # "If your pipe returns an error, the pipe attempts all retries on the affected messages before EventBridge
        # receives additional messages from the same group."
        # https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
        self.delete_messages(messages, successful_message_ids)

    def delete_messages(self, messages: list[dict], message_ids_to_delete: set):
        """Delete SQS `messages` from the source queue that match a MessageId within `message_ids_to_delete`"""
        # TODO: unclear how (partial) failures for deleting are handled, retry or fail batch? Hard to test against AWS
        if len(message_ids_to_delete) > 0:
            entries = [
                {"Id": str(count), "ReceiptHandle": message["ReceiptHandle"]}
                for count, message in enumerate(messages)
                if message["MessageId"] in message_ids_to_delete
            ]
            batch_size = self.sqs_queue_parameters.get("BatchSize", DEFAULT_MAX_RECEIVE_COUNT)
            for batched_entries in batched(entries, batch_size):
                self.source_client.delete_message_batch(
                    QueueUrl=self.queue_url, Entries=batched_entries
                )


def split_by_message_group_id(messages) -> defaultdict[str, list[dict]]:
    """Splitting SQS messages by MessageGroupId to ensure strict ordering for FIFO queues"""
    fifo_groups = defaultdict(list)
    for message in messages:
        message_group_id = message["Attributes"]["MessageGroupId"]
        fifo_groups[message_group_id].append(message)
    return fifo_groups


def transform_into_events(messages: list[dict]) -> list[dict]:
    events = []
    for message in messages:
        # TODO: consolidate with SQS event source listener:
        #  localstack.services.lambda_.event_source_listeners.sqs_event_source_listener.SQSEventSourceListener._send_event_to_lambda
        message_attrs = message_attributes_to_lower(message.get("MessageAttributes"))
        event = {
            # Original SQS message attributes
            "messageId": message["MessageId"],
            "receiptHandle": message["ReceiptHandle"],
            # TODO: test with empty body
            # TODO: implement heuristic based on content type: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
            "body": message.get("Body", "MessageBody"),
            "attributes": message.get("Attributes", {}),
            "messageAttributes": message_attrs,
            # TODO: test with empty body
            "md5OfBody": message.get("MD5OfBody") or message.get("MD5OfMessageBody"),
        }
        # TODO: test Pipe with message attributes (only covered by Lambda ESM SQS test so far)
        if md5_of_message_attributes := message.get("MD5OfMessageAttributes"):
            event["md5OfMessageAttributes"] = md5_of_message_attributes
        events.append(event)
    return events


def get_queue_url(queue_arn: str) -> str:
    # TODO: consolidate this method with localstack.services.sqs.models.SqsQueue.url
    # * Do we need to support different endpoint strategies?
    # * If so, how can we achieve this without having a request context
    host_url = internal_service_url()
    host = host_url.rstrip("/")
    parsed_arn = parse_arn(queue_arn)
    account_id = parsed_arn["account"]
    name = parsed_arn["resource"]
    return f"{host}/{account_id}/{name}"


def message_attributes_to_lower(message_attrs):
    """Convert message attribute details (first characters) to lower 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_lower(key)] = attr.pop(key)
    return message_attrs

1	import json	1✔
2	import logging	1✔
3	from collections import defaultdict	1✔
4	from functools import cached_property	1✔
5
6	from botocore.client import BaseClient	1✔
7
8	from localstack.aws.api.pipes import PipeSourceSqsQueueParameters	1✔
9	from localstack.aws.api.sqs import MessageSystemAttributeName	1✔
10	from localstack.config import internal_service_url	1✔
11	from localstack.services.lambda_.event_source_mapping.event_processor import (	1✔
12	EventProcessor,
13	PartialBatchFailureError,
14	)
15	from localstack.services.lambda_.event_source_mapping.pollers.poller import (	1✔
16	Poller,
17	parse_batch_item_failures,
18	)
19	from localstack.services.lambda_.event_source_mapping.senders.sender_utils import batched	1✔
20	from localstack.services.sqs.constants import HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT	1✔
21	from localstack.utils.aws.arns import parse_arn	1✔
22	from localstack.utils.strings import first_char_to_lower	1✔
23
24	LOG = logging.getLogger(__name__)	1✔
25
26	DEFAULT_MAX_RECEIVE_COUNT = 10	1✔
27
28
29	class SqsPoller(Poller):	1✔
30	queue_url: str	1✔
31
32	def __init__(	1✔
33	self,
34	source_arn: str,
35	source_parameters: dict \| None = None,
36	source_client: BaseClient \| None = None,
37	processor: EventProcessor \| None = None,
38	):
39	super().__init__(source_arn, source_parameters, source_client, processor)	1✔
40	self.queue_url = get_queue_url(self.source_arn)	1✔
41	self._register_client_hooks()	1✔
42
43	@property	1✔
44	def sqs_queue_parameters(self) -> PipeSourceSqsQueueParameters:	1✔
45	return self.source_parameters["SqsQueueParameters"]	1✔
46
47	@cached_property	1✔
48	def is_fifo_queue(self) -> bool:	1✔
49	# Alternative heuristic: self.queue_url.endswith(".fifo"), but we need the call to get_queue_attributes for IAM
50	return self.get_queue_attributes().get("FifoQueue", "false").lower() == "true"	1✔
51
52	def _register_client_hooks(self):	1✔
53	event_system = self.source_client.meta.events	1✔
54
55	def _handle_receive_message_override(params, context, **kwargs):	1✔
56	requested_count = params.get("MaxNumberOfMessages")	1✔
57	if not requested_count or requested_count <= DEFAULT_MAX_RECEIVE_COUNT:	1✔
58	return	1✔
59
60	# Allow overide parameter to be greater than default and less than maximum batch size.
61	# Useful for getting remaining records less than the batch size. i.e we need 100 records but BatchSize is 1k.
62	override = min(requested_count, self.sqs_queue_parameters["BatchSize"])	1✔
63	context[HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = str(override)	1✔
64
65	def _handle_delete_batch_override(params, context, **kwargs):	1✔
66	requested_count = len(params.get("Entries", []))	1✔
67	if not requested_count or requested_count <= DEFAULT_MAX_RECEIVE_COUNT:	1✔
68	return	1✔
69
70	override = min(requested_count, self.sqs_queue_parameters["BatchSize"])	1✔
71	context[HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = str(override)	1✔
72
73	def _handler_inject_header(params, context, **kwargs):	1✔
74	if override := context.pop(HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT, None):	1✔
75	params["headers"][HEADER_LOCALSTACK_SQS_OVERRIDE_MESSAGE_COUNT] = override	1✔
76
77	event_system.register(	1✔
78	"provide-client-params.sqs.ReceiveMessage", _handle_receive_message_override
79	)
80	# Since we delete SQS messages after processing, this allows us to remove up to 10K entries at a time.
81	event_system.register(	1✔
82	"provide-client-params.sqs.DeleteMessageBatch", _handle_delete_batch_override
83	)
84	event_system.register("before-call.sqs.*", _handler_inject_header)	1✔
85
86	def get_queue_attributes(self) -> dict:	1✔
87	"""The API call to sqs:GetQueueAttributes is required for IAM policy streamsing."""
88	get_queue_attributes_response = self.source_client.get_queue_attributes(	1✔
89	QueueUrl=self.queue_url,
90	AttributeNames=["FifoQueue"],
91	)
92	return get_queue_attributes_response.get("Attributes", {})	1✔
93
94	def event_source(self) -> str:	1✔
95	return "aws:sqs"	1✔
96
97	def poll_events(self) -> None:	1✔
98	# SQS pipe source: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
99	# "The 9 Ways an SQS Message can be Deleted": https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/
100	# TODO: implement batch window expires based on MaximumBatchingWindowInSeconds
101	# TODO: implement invocation payload size quota
102	# TODO: consider long-polling vs. short-polling trade-off. AWS uses long-polling:
103	# https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html#pipes-sqs-scaling
104	response = self.source_client.receive_message(	1✔
105	QueueUrl=self.queue_url,
106	MaxNumberOfMessages=self.sqs_queue_parameters.get(
107	"BatchSize", DEFAULT_MAX_RECEIVE_COUNT
108	),
109	MessageAttributeNames=["All"],
110	MessageSystemAttributeNames=[MessageSystemAttributeName.All],
111	)
112	if messages := response.get("Messages"):	1✔
113	LOG.debug("Polled %d events from %s", len(messages), self.source_arn)	1✔
114	try:	1✔
115	if self.is_fifo_queue:	1✔
116	# TODO: think about starvation behavior because once failing message could block other groups
117	fifo_groups = split_by_message_group_id(messages)	1✔
118	for fifo_group_messages in fifo_groups.values():	1✔
119	self.handle_messages(fifo_group_messages)	1✔
120	else:
121	self.handle_messages(messages)	1✔
122
123	# TODO: unify exception handling across pollers: should we catch and raise?
124	except Exception as e:	1✔
125	# TODO: improve error messages (produce same failure and design better error messages)
126	LOG.warning(	1✔
127	"Polling or batch processing failed: %s",
128	e,
129	exc_info=LOG.isEnabledFor(logging.DEBUG),
130	)
131
132	def handle_messages(self, messages):	1✔
133	polled_events = transform_into_events(messages)	1✔
134	# Filtering: matching vs. discarded (i.e., not matching filter criteria)
135	# TODO: implement format detection behavior (e.g., for JSON body):
136	# https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
137	# Check whether we need poller-specific filter-preprocessing here without modifying the actual event!
138	# convert to json for filtering (HACK for fixing parity with v1 and getting regression tests passing)
139	for event in polled_events:	1✔
140	try:	1✔
141	event["body"] = json.loads(event["body"])	1✔
142	except json.JSONDecodeError:	1✔
143	LOG.debug(	1✔
144	"Unable to convert event body '%s' to json... Event might be dropped.",
145	event["body"],
146	)
147	matching_events = self.filter_events(polled_events)	1✔
148	# convert them back (HACK for fixing parity with v1 and getting regression tests passing)
149	for event in matching_events:	1✔
150	event["body"] = (	1✔
151	json.dumps(event["body"]) if not isinstance(event["body"], str) else event["body"]
152	)
153
154	all_message_ids = {message["MessageId"] for message in messages}	1✔
155	matching_message_ids = {event["messageId"] for event in matching_events}	1✔
156	discarded_message_ids = all_message_ids.difference(matching_message_ids)	1✔
157	# Delete discarded events immediately:
158	# https://lucvandonkersgoed.com/2022/01/20/the-9-ways-an-sqs-message-can-be-deleted/#7-event-source-mappings-with-filters
159	self.delete_messages(messages, discarded_message_ids)	1✔
160
161	# Don't trigger upon empty events
162	if len(matching_events) == 0:	1✔
163	return	×
164	# Enrich events with metadata after filtering
165	enriched_events = self.add_source_metadata(matching_events)	1✔
166
167	# Invoke the processor (e.g., Pipe, ESM) and handle partial batch failures
168	try:	1✔
169	self.processor.process_events_batch(enriched_events)	1✔
170	successful_message_ids = all_message_ids	1✔
171	except PartialBatchFailureError as e:	1✔
172	failed_message_ids = parse_batch_item_failures(	1✔
173	e.partial_failure_payload, matching_message_ids
174	)
175	successful_message_ids = matching_message_ids.difference(failed_message_ids)	1✔
176
177	# Only delete messages that are processed successfully as described here:
178	# https://docs.aws.amazon.com/en_gb/lambda/latest/dg/with-sqs.html
179	# When Lambda reads a batch, the messages stay in the queue but are hidden for the length of the queue's
180	# visibility timeout. If your function successfully processes the batch, Lambda deletes the messages
181	# from the queue. By default, if your function encounters an error while processing a batch,
182	# all messages in that batch become visible in the queue again. For this reason, your function code must
183	# be able to process the same message multiple times without unintended side effects.
184	# Troubleshooting: https://repost.aws/knowledge-center/lambda-sqs-report-batch-item-failures
185	# For FIFO queues, AWS also deletes successfully sent messages. Therefore, the AWS docs recommends:
186	# "If you're using this feature with a FIFO queue, your function should stop processing messages after the first
187	# failure and return all failed and unprocessed messages in batchItemFailures. This helps preserve the ordering
188	# of messages in your queue."
189	# Following this recommendation could result in the unsolved side effect that valid messages are continuously
190	# placed in the same batch as failing messages:
191	# * https://stackoverflow.com/questions/78694079/how-to-stop-fifo-sqs-messages-from-being-placed-in-a-batch-with-failing-messages
192	# * https://stackoverflow.com/questions/76912394/can-i-report-only-messages-from-failing-group-id-in-reportbatchitemfailures-resp
193
194	# TODO: Test blocking failure behavior for FIFO queues to guarantee strict ordering
195	# -> might require some checkpointing or retry control on the poller side?!
196	# The poller should only proceed processing FIFO queues after having retried failing messages:
197	# "If your pipe returns an error, the pipe attempts all retries on the affected messages before EventBridge
198	# receives additional messages from the same group."
199	# https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html
200	self.delete_messages(messages, successful_message_ids)	1✔
201
202	def delete_messages(self, messages: list[dict], message_ids_to_delete: set):	1✔
203	"""Delete SQS `messages` from the source queue that match a MessageId within `message_ids_to_delete`"""
204	# TODO: unclear how (partial) failures for deleting are handled, retry or fail batch? Hard to test against AWS
205	if len(message_ids_to_delete) > 0:	1✔
206	entries = [	1✔
207	{"Id": str(count), "ReceiptHandle": message["ReceiptHandle"]}
208	for count, message in enumerate(messages)
209	if message["MessageId"] in message_ids_to_delete
210	]
211	batch_size = self.sqs_queue_parameters.get("BatchSize", DEFAULT_MAX_RECEIVE_COUNT)	1✔
212	for batched_entries in batched(entries, batch_size):	1✔
213	self.source_client.delete_message_batch(	1✔
214	QueueUrl=self.queue_url, Entries=batched_entries
215	)
216
217
218	def split_by_message_group_id(messages) -> defaultdict[str, list[dict]]:	1✔
219	"""Splitting SQS messages by MessageGroupId to ensure strict ordering for FIFO queues"""
220	fifo_groups = defaultdict(list)	1✔
221	for message in messages:	1✔
222	message_group_id = message["Attributes"]["MessageGroupId"]	1✔
223	fifo_groups[message_group_id].append(message)	1✔
224	return fifo_groups	1✔
225
226
227	def transform_into_events(messages: list[dict]) -> list[dict]:	1✔
228	events = []	1✔
229	for message in messages:	1✔
230	# TODO: consolidate with SQS event source listener:
231	# localstack.services.lambda_.event_source_listeners.sqs_event_source_listener.SQSEventSourceListener._send_event_to_lambda
232	message_attrs = message_attributes_to_lower(message.get("MessageAttributes"))	1✔
233	event = {	1✔
234	# Original SQS message attributes
235	"messageId": message["MessageId"],
236	"receiptHandle": message["ReceiptHandle"],
237	# TODO: test with empty body
238	# TODO: implement heuristic based on content type: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html#pipes-filter-sqs
239	"body": message.get("Body", "MessageBody"),
240	"attributes": message.get("Attributes", {}),
241	"messageAttributes": message_attrs,
242	# TODO: test with empty body
243	"md5OfBody": message.get("MD5OfBody") or message.get("MD5OfMessageBody"),
244	}
245	# TODO: test Pipe with message attributes (only covered by Lambda ESM SQS test so far)
246	if md5_of_message_attributes := message.get("MD5OfMessageAttributes"):	1✔
247	event["md5OfMessageAttributes"] = md5_of_message_attributes	1✔
248	events.append(event)	1✔
249	return events	1✔
250
251
252	def get_queue_url(queue_arn: str) -> str:	1✔
253	# TODO: consolidate this method with localstack.services.sqs.models.SqsQueue.url
254	# * Do we need to support different endpoint strategies?
255	# * If so, how can we achieve this without having a request context
256	host_url = internal_service_url()	1✔
257	host = host_url.rstrip("/")	1✔
258	parsed_arn = parse_arn(queue_arn)	1✔
259	account_id = parsed_arn["account"]	1✔
260	name = parsed_arn["resource"]	1✔
261	return f"{host}/{account_id}/{name}"	1✔
262
263
264	def message_attributes_to_lower(message_attrs):	1✔
265	"""Convert message attribute details (first characters) to lower case (e.g., stringValue, dataType)."""
266	message_attrs = message_attrs or {}	1✔
267	for _, attr in message_attrs.items():	1✔
268	if not isinstance(attr, dict):	1✔
269	continue	×
270	for key, value in dict(attr).items():	1✔
271	attr[first_char_to_lower(key)] = attr.pop(key)	1✔
272	return message_attrs	1✔

localstack / localstack / fb37c2a2-4bd7-49b4-891b-ede18134b4b1

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