feat: Handle Fn::If intrinsic for Alarms

integration/helpers/base_test.py

@@ -347,7 +347,9 @@ def verify_stack(self):
         # verify if the stack was successfully created
         self.assertEqual(self.stack_description["Stacks"][0]["StackStatus"], "CREATE_COMPLETE")
         # verify if the stack contains the expected resources
-        self.assertTrue(verify_stack_resources(self.expected_resource_path, self.stack_resources))
+        error = verify_stack_resources(self.expected_resource_path, self.stack_resources)
+        if error:
+            self.fail(error)
 
     def _load_yaml(self, file_path):
         """

integration/helpers/resource.py

@@ -34,7 +34,7 @@ def verify_stack_resources(expected_file_path, stack_resources):
     parsed_resources = _sort_resources(stack_resources["StackResourceSummaries"])
 
     if len(expected_resources) != len(parsed_resources):
-        return False
+        return "'{}' resources expected, '{}' found".format(len(expected_resources), len(parsed_resources))
 
     for i in range(len(expected_resources)):
         exp = expected_resources[i]
@@ -43,10 +43,17 @@ def verify_stack_resources(expected_file_path, stack_resources):
             "^" + exp["LogicalResourceId"] + "([0-9a-f]{" + str(LogicalIdGenerator.HASH_LENGTH) + "})?$",
             parsed["LogicalResourceId"],
         ):
-            return False
+            parsed_trimed_down = {
+                "LogicalResourceId": parsed["LogicalResourceId"],
+                "ResourceType": parsed["ResourceType"],
+            }
+
+            return "'{}' expected, '{}' found (Resources must appear in the same order, don't include the LogicalId random suffix)".format(
+                exp, parsed_trimed_down
+            )
         if exp["ResourceType"] != parsed["ResourceType"]:
-            return False
-    return True
+            return "'{}' expected, '{}' found".format(exp["ResourceType"], parsed["ResourceType"])
+    return None
 
 
 def generate_suffix():

integration/resources/expected/single/function_with_deployment_preference_alarms_intrinsic_if.json

@@ -0,0 +1,30 @@
+[
+  {
+    "LogicalResourceId": "ServerlessDeploymentApplication",
+    "ResourceType": "AWS::CodeDeploy::Application"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "CodeDeployServiceRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionDeploymentGroup",
+    "ResourceType": "AWS::CodeDeploy::DeploymentGroup"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionVersion",
+    "ResourceType": "AWS::Lambda::Version"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionAliaslive",
+    "ResourceType": "AWS::Lambda::Alias"
+  }
+]

integration/resources/templates/single/function_with_deployment_preference_alarms_intrinsic_if.yaml

@@ -0,0 +1,23 @@
+Conditions:
+  MyCondition:
+    Fn::Equals:
+      - true
+      - false
+Resources:
+  MyLambdaFunction:
+    Type: "AWS::Serverless::Function"
+    Properties:
+      CodeUri: ${codeuri}
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        Alarms:
+          Fn::If:
+            - MyCondition
+            - - Alarm1
+              - Alarm2
+              - Alarm3
+            - - Alarm1
+              - Alarm5

integration/single/test_basic_function.py

@@ -40,6 +40,9 @@ def test_function_with_http_api_events(self, file_name):
 
         self.assertEqual(requests.get(endpoint).text, self.FUNCTION_OUTPUT)
 
+    def test_function_with_deployment_preference_alarms_intrinsic_if(self):
+        self.create_and_verify_stack("function_with_deployment_preference_alarms_intrinsic_if")
+
     @parameterized.expand(
         [
             ("basic_function_with_sns_dlq", "sns:Publish"),

samtranslator/model/function_policies.py

@@ -3,7 +3,12 @@
 
 from six import string_types
 
-from samtranslator.model.intrinsics import is_intrinsic, is_intrinsic_if, is_intrinsic_no_value
+from samtranslator.model.intrinsics import (
+    is_intrinsic,
+    is_intrinsic_if,
+    is_intrinsic_no_value,
+    validate_intrinsic_if_items,
+)
 from samtranslator.model.exceptions import InvalidTemplateException
 
 PolicyEntry = namedtuple("PolicyEntry", "data type")
@@ -165,8 +170,10 @@ def _get_type_from_intrinsic_if(self, policy):
         """
         intrinsic_if_value = policy["Fn::If"]
 
-        if not len(intrinsic_if_value) == 3:
-            raise InvalidTemplateException("Fn::If requires 3 arguments")
+        try:
+            validate_intrinsic_if_items(intrinsic_if_value)
+        except ValueError as e:
+            raise InvalidTemplateException(e)
 
         if_data = intrinsic_if_value[1]
         else_data = intrinsic_if_value[2]

samtranslator/model/intrinsics.py

@@ -162,6 +162,24 @@ def is_intrinsic_if(input):
     return key == "Fn::If"
 
 
+def validate_intrinsic_if_items(items):
+    """
+    Validates Fn::If items
+
+    Parameters
+    ----------
+    items : list
+        Fn::If items
+
+    Raises
+    ------
+    ValueError
+        If the items are invalid
+    """
+    if not isinstance(items, list) or len(items) != 3:
+        raise ValueError("Fn::If requires 3 arguments")
+
+
 def is_intrinsic_no_value(input):
     """
     Is the given input an intrinsic Ref: AWS::NoValue? Intrinsic function is a dictionary with single

samtranslator/model/preferences/deployment_preference_collection.py

@@ -1,8 +1,15 @@
 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,
+    validate_intrinsic_if_items,
+)
 from samtranslator.model.update_policy import UpdatePolicy
 from samtranslator.translator.arn_generator import ArnGenerator
 import copy
@@ -125,11 +132,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:
+            deployment_group.AlarmConfiguration = self._convert_alarms(deployment_preference.alarms)
+        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 +158,68 @@ def deployment_group(self, function_logical_id):
 
         return deployment_group
 
+    def _convert_alarms(self, preference_alarms):
+        """
+        Converts deployment preference alarms to an AlarmsConfiguration
+
+        Parameters
+        ----------
+        preference_alarms : dict
+            Deployment preference alarms
+
+        Returns
+        -------
+        dict
+            AlarmsConfiguration if alarms is set, None otherwise
+
+        Raises
+        ------
+        ValueError
+            If Alarms is in the wrong format
+        """
+        if not preference_alarms or is_intrinsic_no_value(preference_alarms):
+            return None
+
+        if is_intrinsic_if(preference_alarms):
+            processed_alarms = copy.deepcopy(preference_alarms)
+            alarms_list = processed_alarms.get("Fn::If")
+            validate_intrinsic_if_items(alarms_list)
+            alarms_list[1] = self._build_alarm_configuration(alarms_list[1])
+            alarms_list[2] = self._build_alarm_configuration(alarms_list[2])
+            return processed_alarms
+
+        return self._build_alarm_configuration(preference_alarms)
+
+    def _build_alarm_configuration(self, alarms):
+        """
+        Builds an AlarmConfiguration 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/model/resource_policies.py

@@ -3,7 +3,12 @@
 
 from six import string_types
 
-from samtranslator.model.intrinsics import is_intrinsic, is_intrinsic_if, is_intrinsic_no_value
+from samtranslator.model.intrinsics import (
+    is_intrinsic,
+    is_intrinsic_if,
+    is_intrinsic_no_value,
+    validate_intrinsic_if_items,
+)
 from samtranslator.model.exceptions import InvalidTemplateException
 
 PolicyEntry = namedtuple("PolicyEntry", "data type")
@@ -165,8 +170,10 @@ def _get_type_from_intrinsic_if(self, policy):
         """
         intrinsic_if_value = policy["Fn::If"]
 
-        if not len(intrinsic_if_value) == 3:
-            raise InvalidTemplateException("Fn::If requires 3 arguments")
+        try:
+            validate_intrinsic_if_items(intrinsic_if_value)
+        except ValueError as e:
+            raise InvalidTemplateException(e)
 
         if_data = intrinsic_if_value[1]
         else_data = intrinsic_if_value[2]

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/model/test_function_policies.py

@@ -3,7 +3,6 @@
 
 from samtranslator.model.function_policies import FunctionPolicies, PolicyTypes, PolicyEntry
 from samtranslator.model.exceptions import InvalidTemplateException
-from samtranslator.model.intrinsics import is_intrinsic_if, is_intrinsic_no_value
 
 
 class TestFunctionPolicies(TestCase):
@@ -271,37 +270,6 @@ def test_is_policy_template_must_return_false_without_the_processor(self):
         self.assertFalse(function_policies_obj._is_policy_template(policy))
         self.policy_template_processor_mock.has.assert_not_called()
 
-    def test_is_intrinsic_if_must_return_true_for_if(self):
-        policy = {"Fn::If": "some value"}
-
-        self.assertTrue(is_intrinsic_if(policy))
-
-    def test_is_intrinsic_if_must_return_false_for_others(self):
-        too_many_keys = {"Fn::If": "some value", "Fn::And": "other value"}
-
-        not_if = {"Fn::Or": "some value"}
-
-        self.assertFalse(is_intrinsic_if(too_many_keys))
-        self.assertFalse(is_intrinsic_if(not_if))
-        self.assertFalse(is_intrinsic_if(None))
-
-    def test_is_intrinsic_no_value_must_return_true_for_no_value(self):
-        policy = {"Ref": "AWS::NoValue"}
-
-        self.assertTrue(is_intrinsic_no_value(policy))
-
-    def test_is_intrinsic_no_value_must_return_false_for_other_value(self):
-        bad_key = {"sRefs": "AWS::NoValue"}
-
-        bad_value = {"Ref": "SWA::NoValue"}
-
-        too_many_keys = {"Ref": "AWS::NoValue", "feR": "SWA::NoValue"}
-
-        self.assertFalse(is_intrinsic_no_value(bad_key))
-        self.assertFalse(is_intrinsic_no_value(bad_value))
-        self.assertFalse(is_intrinsic_no_value(None))
-        self.assertFalse(is_intrinsic_no_value(too_many_keys))
-
     def test_get_type_with_intrinsic_if_must_return_managed_policy_type(self):
         managed_policy = {"Fn::If": ["SomeCondition", "some managed policy arn", "other managed policy arn"]}