Add support to remove RedrivePolicy from SQS. (#4664)

moto/sqs/models.py

@@ -335,7 +335,7 @@ class Queue(CloudFormationModel):
 
             setattr(self, camelcase_to_underscores(key), value)
 
-        if attributes.get("RedrivePolicy", None):
+        if attributes.get("RedrivePolicy", None) is not None:
             self._setup_dlq(attributes["RedrivePolicy"])
 
         if attributes.get("Policy"):
@@ -343,7 +343,21 @@ class Queue(CloudFormationModel):
 
         self.last_modified_timestamp = now
 
+    @staticmethod
+    def _is_empty_redrive_policy(policy):
+        if isinstance(policy, str):
+            if policy == "" or len(json.loads(policy)) == 0:
+                return True
+        elif isinstance(policy, dict) and len(policy) == 0:
+            return True
+
+        return False
+
     def _setup_dlq(self, policy):
+        if Queue._is_empty_redrive_policy(policy):
+            self.redrive_policy = None
+            self.dead_letter_queue = None
+            return
 
         if isinstance(policy, str):
             try:

tests/test_sqs/test_sqs.py

@@ -20,6 +20,7 @@ import pytest
 from tests.helpers import requires_boto_gte
 from moto.core import ACCOUNT_ID
 from moto.sqs.models import (
+    Queue,
     MAXIMUM_MESSAGE_SIZE_ATTR_LOWER_BOUND,
     MAXIMUM_MESSAGE_SIZE_ATTR_UPPER_BOUND,
     MAXIMUM_MESSAGE_LENGTH,
@@ -279,6 +280,46 @@ def test_set_queue_attribute_empty_policy_removes_attr():
     response.shouldnt.have.key("Policy")
 
 
+def test_is_empty_redrive_policy_returns_true_for_empty_and_falsy_values():
+    assert Queue._is_empty_redrive_policy("")
+    assert Queue._is_empty_redrive_policy("{}")
+
+
+def test_is_empty_redrive_policy_returns_false_for_valid_policy_format():
+    test_dlq_arn = "arn:aws:sqs:us-east-1:123456789012:test-dlr-queue"
+    assert not Queue._is_empty_redrive_policy(
+        json.dumps({"deadLetterTargetArn": test_dlq_arn, "maxReceiveCount": 5})
+    )
+    assert not Queue._is_empty_redrive_policy(json.dumps({"maxReceiveCount": 5}))
+
+
+@mock_sqs
+def test_set_queue_attribute_empty_redrive_removes_attr():
+    client = boto3.client("sqs", region_name="us-east-1")
+
+    dlq_resp = client.create_queue(QueueName="test-dlr-queue")
+    dlq_arn1 = client.get_queue_attributes(
+        QueueUrl=dlq_resp["QueueUrl"], AttributeNames=["QueueArn"]
+    )["Attributes"]["QueueArn"]
+    q_name = str(uuid4())[0:6]
+    response = client.create_queue(
+        QueueName=q_name,
+        Attributes={
+            "RedrivePolicy": json.dumps(
+                {"deadLetterTargetArn": dlq_arn1, "maxReceiveCount": 5}
+            ),
+        },
+    )
+    queue_url = response["QueueUrl"]
+
+    no_redrive = {"RedrivePolicy": ""}
+    client.set_queue_attributes(QueueUrl=queue_url, Attributes=no_redrive)
+    response = client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["All"])[
+        "Attributes"
+    ]
+    response.shouldnt.have.key("RedrivePolicy")
+
+
 @mock_sqs
 def test_get_queue_url():
     client = boto3.client("sqs", region_name="us-east-1")