feat: Handle Fn::If intrinsic for Alarms

samtranslator/model/preferences/deployment_preference_collection.py

@@ -1,8 +1,9 @@
 from .deployment_preference import DeploymentPreference
 from samtranslator.model.codedeploy import CodeDeployApplication
 from samtranslator.model.codedeploy import CodeDeployDeploymentGroup
+from samtranslator.model.exceptions import InvalidResourceException
 from samtranslator.model.iam import IAMRole
-from samtranslator.model.intrinsics import fnSub, is_intrinsic
+from samtranslator.model.intrinsics import fnSub, is_intrinsic, is_intrinsic_if, is_intrinsic_no_value
 from samtranslator.model.update_policy import UpdatePolicy
 from samtranslator.translator.arn_generator import ArnGenerator
 import copy
@@ -125,11 +126,10 @@ def deployment_group(self, function_logical_id):
 
         deployment_group = CodeDeployDeploymentGroup(self.deployment_group_logical_id(function_logical_id))
 
-        if deployment_preference.alarms is not None:
-            deployment_group.AlarmConfiguration = {
-                "Enabled": True,
-                "Alarms": [{"Name": alarm} for alarm in deployment_preference.alarms],
-            }
+        try:
+            self._process_alarms(deployment_preference, deployment_group)
+        except ValueError as e:
+            raise InvalidResourceException(function_logical_id, str(e))
 
         deployment_group.ApplicationName = self.codedeploy_application.get_runtime_attr("name")
         deployment_group.AutoRollbackConfiguration = {
@@ -152,6 +152,67 @@ def deployment_group(self, function_logical_id):
 
         return deployment_group
 
+    def _process_alarms(self, preference, group):
+        """
+        Processes the deployment preferences alarms and updates the deployment group accordingly
+
+        Parameters
+        ----------
+        preference : dict
+            Deployment preferences
+        group : dict
+            Deployment group
+
+        Raises
+        ------
+        ValueError
+            If Alarms is in the wrong format
+        """
+        if not preference.alarms or is_intrinsic_no_value(preference.alarms):
+            return
+
+        if is_intrinsic_if(preference.alarms):
+            processed_alarms = copy.deepcopy(preference.alarms)
+            alarms_list = processed_alarms.get("Fn::If")
+            if not isinstance(alarms_list, list) or not len(alarms_list) == 3:
+                raise ValueError("Fn::If requires 3 arguments")
+            alarms_list[1] = self._build_alarms_configuration(alarms_list[1])
+            alarms_list[2] = self._build_alarms_configuration(alarms_list[2])
+            group.AlarmConfiguration = processed_alarms
+            return
+
+        group.AlarmConfiguration = self._build_alarms_configuration(preference.alarms)
+
+    def _build_alarms_configuration(self, alarms):
+        """
+        Builds an Alarms configuration from a list of alarms
+
+        Parameters
+        ----------
+        alarms : list[str]
+            Alarms
+
+        Returns
+        -------
+        dict
+            AlarmsConfiguration for a deployment group
+
+        Raises
+        ------
+        ValueError
+            If alarms is not a list
+        """
+        if not isinstance(alarms, list):
+            raise ValueError("Alarms must be a list")
+
+        if len(alarms) == 0 or is_intrinsic_no_value(alarms[0]):
+            return {}
+
+        return {
+            "Enabled": True,
+            "Alarms": [{"Name": alarm} for alarm in alarms],
+        }
+
     def _replace_deployment_types(self, value, key=None):
         if isinstance(value, list):
             for i in range(len(value)):

samtranslator/translator/translator.py

@@ -152,7 +152,12 @@ def translate(self, sam_template, parameter_values, feature_toggle=None):
                 template["Resources"].update(deployment_preference_collection.codedeploy_iam_role.to_dict())
 
             for logical_id in deployment_preference_collection.enabled_logical_ids():
-                template["Resources"].update(deployment_preference_collection.deployment_group(logical_id).to_dict())
+                try:
+                    template["Resources"].update(
+                        deployment_preference_collection.deployment_group(logical_id).to_dict()
+                    )
+                except InvalidResourceException as e:
+                    document_errors.append(e)
 
         # Run the after-transform plugin target
         try:

tests/translator/input/function_with_deployment_preference_alarms_intrinsic_if.yaml

@@ -0,0 +1,23 @@
+Conditions:
+  MyCondition:
+    Fn::Equals:
+      - true
+      - false
+Resources:
+  MinimalFunction:
+    Type: "AWS::Serverless::Function"
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        Alarms:
+          Fn::If:
+            - MyCondition
+            - - Alarm1
+              - Alarm2
+              - Alarm3
+            - - Alarm1
+              - Alarm5

tests/translator/model/preferences/test_deployment_preference_collection.py

@@ -163,6 +163,170 @@ def test_deployment_group_with_all_parameters(self):
 
         self.assertEqual(deployment_group.to_dict(), expected_deployment_group.to_dict())
 
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": ["Alarm1", "Alarm2", "Alarm3"],
+        }
+        expected_alarm_configuration = {
+            "Enabled": True,
+            "Alarms": [{"Name": "Alarm1"}, {"Name": "Alarm2"}, {"Name": "Alarm3"}],
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertEqual(expected_alarm_configuration, deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_intrinsic_if(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Fn::If": ["MyCondition", ["Alarm1", "Alarm2"], ["Alarm3"]]},
+        }
+        expected_alarm_configuration = {
+            "Fn::If": [
+                "MyCondition",
+                {"Enabled": True, "Alarms": [{"Name": "Alarm1"}, {"Name": "Alarm2"}]},
+                {"Enabled": True, "Alarms": [{"Name": "Alarm3"}]},
+            ],
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertEqual(expected_alarm_configuration, deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_intrinsic_if_empty_then(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Fn::If": ["MyCondition", [], ["Alarm3"]]},
+        }
+        expected_alarm_configuration = {
+            "Fn::If": ["MyCondition", {}, {"Enabled": True, "Alarms": [{"Name": "Alarm3"}]}],
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertEqual(expected_alarm_configuration, deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_intrinsic_if_noref_then(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Fn::If": ["MyCondition", [{"Ref": "AWS::NoValue"}], ["Alarm3"]]},
+        }
+        expected_alarm_configuration = {
+            "Fn::If": ["MyCondition", {}, {"Enabled": True, "Alarms": [{"Name": "Alarm3"}]}],
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertEqual(expected_alarm_configuration, deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_intrinsic_if_empty_else(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Fn::If": ["MyCondition", ["Alarm1", "Alarm2"], []]},
+        }
+        expected_alarm_configuration = {
+            "Fn::If": ["MyCondition", {"Enabled": True, "Alarms": [{"Name": "Alarm1"}, {"Name": "Alarm2"}]}, {}],
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertEqual(expected_alarm_configuration, deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_intrinsic_if_noref_else(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Fn::If": ["MyCondition", ["Alarm1", "Alarm2"], [{"Ref": "AWS::NoValue"}]]},
+        }
+        expected_alarm_configuration = {
+            "Fn::If": ["MyCondition", {"Enabled": True, "Alarms": [{"Name": "Alarm1"}, {"Name": "Alarm2"}]}, {}],
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertEqual(expected_alarm_configuration, deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_ref_novalue(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Ref": "AWS::NoValue"},
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertIsNone(deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_empty(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": [],
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        deployment_group = deployment_preference_collection.deployment_group(self.function_logical_id)
+
+        self.assertIsNone(deployment_group.AlarmConfiguration)
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_not_list(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": "Alarm1",
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        with self.assertRaises(InvalidResourceException) as e:
+            deployment_preference_collection.deployment_group(self.function_logical_id)
+        self.assertEqual(
+            e.exception.message,
+            "Resource with id [{}] is invalid. Alarms must be a list".format(self.function_logical_id),
+        )
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_intrinsic_if_missing_arg(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Fn::If": ["MyCondition", ["Alarm1", "Alarm2"]]},
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        with self.assertRaises(InvalidResourceException) as e:
+            deployment_preference_collection.deployment_group(self.function_logical_id)
+        self.assertEqual(
+            e.exception.message,
+            "Resource with id [{}] is invalid. Fn::If requires 3 arguments".format(self.function_logical_id),
+        )
+
+    @patch("boto3.session.Session.region_name", "ap-southeast-1")
+    def test_deployment_preference_with_alarms_intrinsic_if_not_list(self):
+        deployment_preference = {
+            "Type": "TestDeploymentConfiguration",
+            "Alarms": {"Fn::If": "Alarm1"},
+        }
+        deployment_preference_collection = DeploymentPreferenceCollection()
+        deployment_preference_collection.add(self.function_logical_id, deployment_preference)
+        with self.assertRaises(InvalidResourceException) as e:
+            deployment_preference_collection.deployment_group(self.function_logical_id)
+        self.assertEqual(
+            e.exception.message,
+            "Resource with id [{}] is invalid. Fn::If requires 3 arguments".format(self.function_logical_id),
+        )
+
     @patch("boto3.session.Session.region_name", "ap-southeast-1")
     def test_update_policy_with_minimal_parameters(self):
         expected_update_policy = {