RedrivePolicy added (#2583)

Co-authored-by: Slava Senchenko <[email protected]>

Author: ssenchenko

samtranslator/model/eventsources/push.py

@@ -1,5 +1,6 @@
 import copy
 import re
+from typing import Any, Dict, Optional
 
 from samtranslator.metrics.method_decorator import cw_timer
 from samtranslator.model import ResourceMacro, PropertyType
@@ -451,6 +452,7 @@ class SNS(PushEventSource):
         "Region": PropertyType(False, is_str()),
         "FilterPolicy": PropertyType(False, dict_of(is_str(), list_of(one_of(is_str(), is_type(dict))))),
         "SqsSubscription": PropertyType(False, one_of(is_type(bool), is_type(dict))),
+        "RedrivePolicy": PropertyType(False, is_type(dict)),
     }
 
     @cw_timer(prefix=FUNCTION_EVETSOURCE_METRIC_PREFIX)  # type: ignore[no-untyped-call]
@@ -469,12 +471,13 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
 
         # SNS -> Lambda
         if not self.SqsSubscription:  # type: ignore[attr-defined]
-            subscription = self._inject_subscription(  # type: ignore[no-untyped-call]
+            subscription = self._inject_subscription(
                 "lambda",
                 function.get_runtime_attr("arn"),
                 self.Topic,  # type: ignore[attr-defined]
                 self.Region,  # type: ignore[attr-defined]
                 self.FilterPolicy,  # type: ignore[attr-defined]
+                self.RedrivePolicy,  # type: ignore[attr-defined]
                 function,
             )
             return [self._construct_permission(function, source_arn=self.Topic), subscription]  # type: ignore[attr-defined, no-untyped-call]
@@ -487,8 +490,8 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
             queue_url = queue.get_runtime_attr("queue_url")
 
             queue_policy = self._inject_sqs_queue_policy(self.Topic, queue_arn, queue_url, function)  # type: ignore[attr-defined, no-untyped-call]
-            subscription = self._inject_subscription(  # type: ignore[no-untyped-call]
-                "sqs", queue_arn, self.Topic, self.Region, self.FilterPolicy, function  # type: ignore[attr-defined, attr-defined, attr-defined]
+            subscription = self._inject_subscription(
+                "sqs", queue_arn, self.Topic, self.Region, self.FilterPolicy, self.RedrivePolicy, function  # type: ignore[attr-defined, attr-defined, attr-defined]
             )
             event_source = self._inject_sqs_event_source_mapping(function, role, queue_arn)  # type: ignore[no-untyped-call]
 
@@ -512,15 +515,26 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
         queue_policy = self._inject_sqs_queue_policy(  # type: ignore[no-untyped-call]
             self.Topic, queue_arn, queue_url, function, queue_policy_logical_id  # type: ignore[attr-defined]
         )
-        subscription = self._inject_subscription("sqs", queue_arn, self.Topic, self.Region, self.FilterPolicy, function)  # type: ignore[attr-defined, attr-defined, attr-defined, no-untyped-call]
+        subscription = self._inject_subscription(
+            "sqs", queue_arn, self.Topic, self.Region, self.FilterPolicy, self.RedrivePolicy, function  # type: ignore[attr-defined, attr-defined, attr-defined]
+        )
         event_source = self._inject_sqs_event_source_mapping(function, role, queue_arn, batch_size, enabled)  # type: ignore[no-untyped-call]
 
         resources = resources + event_source
         resources.append(queue_policy)
         resources.append(subscription)
         return resources
 
-    def _inject_subscription(self, protocol, endpoint, topic, region, filterPolicy, function):  # type: ignore[no-untyped-def]
+    def _inject_subscription(
+        self,
+        protocol: str,
+        endpoint: str,
+        topic: str,
+        region: Optional[str],
+        filterPolicy: Optional[Dict[str, Any]],
+        redrivePolicy: Optional[Dict[str, Any]],
+        function: Any,
+    ) -> SNSSubscription:
         subscription = SNSSubscription(self.logical_id, attributes=function.get_passthrough_resource_attributes())
         subscription.Protocol = protocol
         subscription.Endpoint = endpoint
@@ -532,6 +546,9 @@ def _inject_subscription(self, protocol, endpoint, topic, region, filterPolicy,
         if filterPolicy is not None:
             subscription.FilterPolicy = filterPolicy
 
+        if redrivePolicy is not None:
+            subscription.RedrivePolicy = redrivePolicy
+
         return subscription
 
     def _inject_sqs_queue(self, function):  # type: ignore[no-untyped-def]

samtranslator/model/sns.py

@@ -11,6 +11,7 @@ class SNSSubscription(Resource):
         "TopicArn": PropertyType(True, is_str()),
         "Region": PropertyType(False, is_str()),
         "FilterPolicy": PropertyType(False, is_type(dict)),
+        "RedrivePolicy": PropertyType(False, is_type(dict)),
     }
 
 

tests/model/eventsources/test_sns_event_source.py

@@ -29,6 +29,7 @@ def test_to_cloudformation_returns_permission_and_subscription_resources(self):
         self.assertEqual(subscription.Endpoint, "arn:aws:lambda:mock")
         self.assertIsNone(subscription.Region)
         self.assertIsNone(subscription.FilterPolicy)
+        self.assertIsNone(subscription.RedrivePolicy)
 
     def test_to_cloudformation_passes_the_region(self):
         region = "us-west-2"
@@ -54,6 +55,16 @@ def test_to_cloudformation_passes_the_filter_policy(self):
         subscription = resources[1]
         self.assertEqual(subscription.FilterPolicy, filterPolicy)
 
+    def test_to_cloudformation_passes_the_redrive_policy(self):
+        redrive_policy = {"deadLetterTargetArn": "arn:aws:sqs:us-east-2:123456789012:MyDeadLetterQueue"}
+        self.sns_event_source.RedrivePolicy = redrive_policy
+
+        resources = self.sns_event_source.to_cloudformation(function=self.function)
+        self.assertEqual(len(resources), 2)
+        self.assertEqual(resources[1].resource_type, "AWS::SNS::Subscription")
+        subscription = resources[1]
+        self.assertEqual(subscription.RedrivePolicy, redrive_policy)
+
     def test_to_cloudformation_throws_when_no_function(self):
         self.assertRaises(TypeError, self.sns_event_source.to_cloudformation)
 
@@ -77,3 +88,4 @@ def test_to_cloudformation_when_sqs_subscription_disable(self):
         self.assertEqual(subscription.Endpoint, "arn:aws:lambda:mock")
         self.assertIsNone(subscription.Region)
         self.assertIsNone(subscription.FilterPolicy)
+        self.assertIsNone(subscription.RedrivePolicy)