localstack · peter-smith-phd · Dec 11, 2025 · Dec 4, 2025 · baermat · Dec 10, 2025
@@ -275,6 +275,26 @@ def recursive_convert(obj):
     return recursive_convert(input_dict)
 
 
+def resource_tags_to_remove_or_update(
+    prev_tags: list[dict], new_tags: list[dict]
+) -> tuple[list[str], dict[str, str]]:
+    """
+    When updating resources that have tags, we need to determine which tags to remove and which to add/update,
+    as these are typically done in separate API calls. The format of prev_tags and new_tags is expected to
+    be [{ "Key": tagName, "Value": tagValue }, ...]. The return value will be a tuple of (tags_to_remove, tags_to_update),
+    where:
+    - tags_to_remove is a list of tag keys that are present in prev_tags but not in new_tags.
+    - tags_to_update is a dict of tags to add or update, with the format: { tagName: tagValue, ... }.
+    """
+    prev_tag_keys = [tag["Key"] for tag in prev_tags]
+    new_tag_keys = [tag["Key"] for tag in new_tags]
+    tags_to_remove = list(set(prev_tag_keys) - set(new_tag_keys))
+
+    # convert from list of dicts, to a single dict because that's what tag_queue APIs expect.
+    tags_to_update = {tag["Key"]: tag["Value"] for tag in new_tags}
+    return (tags_to_remove, tags_to_update)
+
+
 #  LocalStack specific utilities
 def get_schema_path(file_path: Path) -> dict:
     file_name_base = file_path.name.removesuffix(".py").removesuffix(".py.enc")

@@ -31,6 +31,12 @@
     QueueAttributeName.QueueArn,
 ]
 
+#
+# If these attributes are set to their default values, they are effectively
+# deleted from the queue attributes and not returned in future calls to get_queue_attributes()
+#
+DELETE_IF_DEFAULT = {"KmsMasterKeyId": "", "KmsDataKeyReusePeriodSeconds": "300"}
+
 INVALID_STANDARD_QUEUE_ATTRIBUTES = [
     QueueAttributeName.FifoQueue,
     QueueAttributeName.ContentBasedDeduplication,

@@ -1273,7 +1273,11 @@ def set_queue_attributes(
         for k, v in attributes.items():
             if k in sqs_constants.INTERNAL_QUEUE_ATTRIBUTES:
                 raise InvalidAttributeName(f"Unknown Attribute {k}.")
-            queue.attributes[k] = v
+            if k in sqs_constants.DELETE_IF_DEFAULT and v == sqs_constants.DELETE_IF_DEFAULT[k]:
+                if k in queue.attributes:
+                    del queue.attributes[k]
+            else:
+                queue.attributes[k] = v
 
         # Special cases
         if queue.attributes.get(QueueAttributeName.Policy) == "":

@@ -64,6 +64,35 @@ class SQSQueueProvider(ResourceProvider[SQSQueueProperties]):
     TYPE = "AWS::SQS::Queue"  # Autogenerated. Don't change
     SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
 
+    # Values used when a property is removed from a template and needs to be set to its default.
+    # If AWS changes their defaults in the future, our parity tests should break.
+    DEFAULT_ATTRIBUTE_VALUES = {
+        "ReceiveMessageWaitTimeSeconds": "0",
+        "DelaySeconds": "0",
+        "KmsMasterKeyId": "",
+        "RedrivePolicy": "",
+        "MessageRetentionPeriod": "345600",
+        "MaximumMessageSize": "262144",  # Note: CloudFormation sets this to 256KB on update, but 1MB on create
+        "VisibilityTimeout": "30",
+        "KmsDataKeyReusePeriodSeconds": "300",
+    }
+
+    # Private method for creating a unique queue name, if none is specified.
+    def _autogenerated_queue_name(self, request: ResourceRequest[SQSQueueProperties]) -> str:
-    # Private method for creating a unique queue name, if none is specified.
-    def _autogenerated_queue_name(self, request: ResourceRequest[SQSQueueProperties]) -> str:
+    # Private method for creating a unique queue name, if none is specified.
+    def _generate_queue_name(self, request: ResourceRequest[SQSQueueProperties]) -> str:
-    # Private method for creating a unique queue name, if none is specified.
-    def _autogenerated_queue_name(self, request: ResourceRequest[SQSQueueProperties]) -> str:
+    # Private method for creating a unique queue name, if none is specified.
+    def _generate_queue_name(self, request: ResourceRequest[SQSQueueProperties]) -> str:
+        queue_name = util.generate_default_name(request.stack_name, request.logical_resource_id)
+        isFifoQueue = request.desired_state.get("FifoQueue")
+
+        # Note that it's an SQS FIFO queue only if the FifoQueue property is set to boolean True, or the string "true"
+        # (case insensitive). If it's None (property was omitted) or False, or any type of string (e.g. a typo
+        # such as "Fasle"), then it's not a FIFO queue. This extra check is needed because the CloudFormation engine
+        # doesn't fully validate the FifoQueue property before passing it to the resource provider.
+        if (
+            isFifoQueue == True  # noqa: E712
+            or (isinstance(isFifoQueue, str) and isFifoQueue.lower() == "true")
+        ):
+            queue_name = f"{queue_name[:-5]}.fifo"
+        return queue_name
+
     def create(
         self,
         request: ResourceRequest[SQSQueueProperties],
@@ -74,8 +103,6 @@ def create(
         Primary identifier fields:
           - /properties/QueueUrl
 
-
-
         Create-only properties:
           - /properties/FifoQueue
           - /properties/QueueName
@@ -92,26 +119,13 @@ def create(
           - sqs:TagQueue
 
         """
-        # TODO: validations
+        # TODO: validations - what validations are needed?
         model = request.desired_state
         sqs = request.aws_client_factory.sqs
 
-        if model.get("FifoQueue", False):
-            model["FifoQueue"] = model["FifoQueue"]
-
-        queue_name = model.get("QueueName")
-        if not queue_name:
-            # TODO: verify patterns here
-            if model.get("FifoQueue"):
-                queue_name = util.generate_default_name(
-                    request.stack_name, request.logical_resource_id
-                )[:-5]
-                queue_name = f"{queue_name}.fifo"
-            else:
-                queue_name = util.generate_default_name(
-                    request.stack_name, request.logical_resource_id
-                )
-            model["QueueName"] = queue_name
+        # if no QueueName is specified, automatically generate one
+        if not model.get("QueueName"):
+            model["QueueName"] = self._autogenerated_queue_name(request)
 
         attributes = self._compile_sqs_queue_attributes(model)
         result = request.aws_client_factory.sqs.create_queue(
@@ -184,38 +198,30 @@ def update(
         """
         sqs = request.aws_client_factory.sqs
         model = request.desired_state
+        prev_model = request.previous_state
 
         assert request.previous_state is not None
 
-        should_replace = (
-            request.desired_state.get("QueueName", request.previous_state["QueueName"])
-            != request.previous_state["QueueName"]
-        ) or (
-            request.desired_state.get("FifoQueue", request.previous_state.get("FifoQueue&quo
E988
t;))
-            != request.previous_state.get("FifoQueue")
+        queue_url = prev_model["QueueUrl"]
+        self._populate_missing_attributes_with_defaults(model)
+        sqs.set_queue_attributes(
+            QueueUrl=queue_url, Attributes=self._compile_sqs_queue_attributes(model)
         )
 
-        if not should_replace:
-            return ProgressEvent(OperationStatus.SUCCESS, resource_model=request.previous_state)
-
-        # TODO: copied from the create handler, extract?
-        if model.get("FifoQueue"):
-            queue_name = util.generate_default_name(
-                request.stack_name, request.logical_resource_id
-            )[:-5]
-            queue_name = f"{queue_name}.fifo"
-        else:
-            queue_name = util.generate_default_name(request.stack_name, request.logical_resource_id)
-
-        # replacement (TODO: find out if we should handle this in the provider or outside of it)
-        # delete old queue
-        sqs.delete_queue(QueueUrl=request.previous_state["QueueUrl"])
-        # create new queue (TODO: re-use create logic to make this more robust, e.g. for
-        #  auto-generated queue names)
-        model["QueueUrl"] = sqs.create_queue(QueueName=queue_name)["QueueUrl"]
-        model["Arn"] = sqs.get_queue_attributes(
-            QueueUrl=model["QueueUrl"], AttributeNames=["QueueArn"]
-        )["Attributes"]["QueueArn"]
+        (tags_to_remove, tags_to_add_or_update) = util.resource_tags_to_remove_or_update(
+            prev_model.get("Tags", []), model.get("Tags", [])
+        )
+        sqs.untag_queue(QueueUrl=queue_url, TagKeys=tags_to_remove)
+        sqs.tag_queue(QueueUrl=queue_url, Tags=tags_to_add_or_update)
+
+        model["QueueUrl"] = queue_url
+        model["Arn"] = request.previous_state["Arn"]
+
+        # For QueueName and FifoQueue, always use the value from the previous model. These fields
+        # are create-only, so they cannot be changed via an update (even though they might be omitted)
+        model["QueueName"] = prev_model.get("QueueName")
+        model["FifoQueue"] = prev_model.get("FifoQueue", False)
+
         return ProgressEvent(OperationStatus.SUCCESS, resource_model=model)
 
     def _compile_sqs_queue_attributes(self, properties: SQSQueueProperties) -> dict[str, str]:
@@ -250,6 +256,15 @@ def _compile_sqs_queue_attributes(self, properties: SQSQueueProperties) -> dict[
 
         return result
 
+    def _populate_missing_attributes_with_defaults(self, properties: SQSQueueProperties) -> None:
+        """
+        For any attribute that is missing from the desired state, populate it with the default value.
+        This is the only way to remove an attribute from an existing SQS queue's configuration.
+        :param properties: the properties passed from cloudformation
+        """
+        for k, v in self.DEFAULT_ATTRIBUTE_VALUES.items():
+            properties.setdefault(k, v)
+
     def list(
         self,
         request: ResourceRequest[SQSQueueProperties],

@@ -340,6 +340,7 @@ def test_update_stack_with_same_template_withoutchange_transformation(
         )
 
     @markers.aws.validated
+    @skip_if_legacy_engine()
     def test_update_stack_actual_update(self, deploy_cfn_template, aws_client):
         template = load_file(
             os.path.join(os.path.dirname(__file__), "../../../templates/sqs_queue_update.yml")