fix code deploy conditions bug

Author: keetonian

samtranslator/model/preferences/deployment_preference.py

@@ -25,7 +25,16 @@
 """
 DeploymentPreferenceTuple = namedtuple(
     "DeploymentPreferenceTuple",
-    ["deployment_type", "pre_traffic_hook", "post_traffic_hook", "alarms", "enabled", "role", "trigger_configurations"],
+    [
+        "deployment_type",
+        "pre_traffic_hook",
+        "post_traffic_hook",
+        "alarms",
+        "enabled",
+        "role",
+        "trigger_configurations",
+        "condition",
+    ],
 )
 
 
@@ -37,17 +46,18 @@ class DeploymentPreference(DeploymentPreferenceTuple):
     """
 
     @classmethod
-    def from_dict(cls, logical_id, deployment_preference_dict):
+    def from_dict(cls, logical_id, deployment_preference_dict, condition=None):
         """
         :param logical_id: the logical_id of the resource that owns this deployment preference
         :param deployment_preference_dict: the dict object taken from the SAM template
+        :param condition: condition on this deployment preference
         :return:
         """
         enabled = deployment_preference_dict.get("Enabled", True)
         enabled = False if enabled in ["false", "False"] else enabled
 
         if not enabled:
-            return DeploymentPreference(None, None, None, None, False, None, None)
+            return DeploymentPreference(None, None, None, None, False, None, None, None)
 
         if "Type" not in deployment_preference_dict:
             raise InvalidResourceException(logical_id, "'DeploymentPreference' is missing required Property 'Type'")
@@ -65,5 +75,12 @@ def from_dict(cls, logical_id, deployment_preference_dict):
         role = deployment_preference_dict.get("Role", None)
         trigger_configurations = deployment_preference_dict.get("TriggerConfigurations", None)
         return DeploymentPreference(
-            deployment_type, pre_traffic_hook, post_traffic_hook, alarms, enabled, role, trigger_configurations
+            deployment_type,
+            pre_traffic_hook,
+            post_traffic_hook,
+            alarms,
+            enabled,
+            role,
+            trigger_configurations,
+            condition,
         )

samtranslator/model/preferences/deployment_preference_collection.py

@@ -48,13 +48,14 @@ def __init__(self):
         self.codedeploy_application = self._codedeploy_application()
         self.codedeploy_iam_role = self._codedeploy_iam_role()
 
-    def add(self, logical_id, deployment_preference_dict):
+    def add(self, logical_id, deployment_preference_dict, condition=None):
         """
         Add this deployment preference to the collection
 
         :raise ValueError if an existing logical id already exists in the _resource_preferences
         :param logical_id: logical id of the resource where this deployment preference applies
         :param deployment_preference_dict: the input SAM template deployment preference mapping
+        :param condition: the condition (if it exists) on the serverless function
         """
         if logical_id in self._resource_preferences:
             raise ValueError(
@@ -63,7 +64,9 @@ def add(self, logical_id, deployment_preference_dict):
                 )
             )
 
-        self._resource_preferences[logical_id] = DeploymentPreference.from_dict(logical_id, deployment_preference_dict)
+        self._resource_preferences[logical_id] = DeploymentPreference.from_dict(
+            logical_id, deployment_preference_dict, condition
+        )
 
     def get(self, logical_id):
         """
@@ -85,6 +88,13 @@ def can_skip_service_role(self):
         """
         return all(preference.role or not preference.enabled for preference in self._resource_preferences.values())
 
+    def needs_resource_condition(self):
+        """
+        If all preferences have a condition, all code deploy resources need to be conditionally created
+        :return: True, if a condition needs to be created
+        """
+        return all(preference.condition for preference in self._resource_preferences.values())
+
     def enabled_logical_ids(self):
         """
         :return: only the logical id's for the deployment preferences in this collection which are enabled
@@ -156,6 +166,9 @@ def deployment_group(self, function_logical_id):
         if deployment_preference.trigger_configurations:
             deployment_group.TriggerConfigurations = deployment_preference.trigger_configurations
 
+        if deployment_preference.condition:
+            deployment_group.set_resource_attribute("Condition", deployment_preference.condition)
+
         return deployment_group
 
     def _convert_alarms(self, preference_alarms):

samtranslator/model/sam_resources.py

@@ -185,6 +185,7 @@ def to_cloudformation(self, **kwargs):
                 lambda_alias,
                 intrinsics_resolver,
                 mappings_resolver,
+                self.get_passthrough_resource_attributes(),
             )
         event_invoke_policies = []
         if self.EventInvokeConfig:
@@ -827,7 +828,7 @@ def _construct_alias(self, name, function, version):
         return alias
 
     def _validate_deployment_preference_and_add_update_policy(
-        self, deployment_preference_collection, lambda_alias, intrinsics_resolver, mappings_resolver
+        self, deployment_preference_collection, lambda_alias, intrinsics_resolver, mappings_resolver, condition
     ):
         if "Enabled" in self.DeploymentPreference:
             # resolve intrinsics and mappings for Type
@@ -843,10 +844,15 @@ def _validate_deployment_preference_and_add_update_policy(
             preference_type = mappings_resolver.resolve_parameter_refs(preference_type)
             self.DeploymentPreference["Type"] = preference_type
 
+        if condition:
+            condition = condition.get("Condition")
+
         if deployment_preference_collection is None:
             raise ValueError("deployment_preference_collection required for parsing the deployment preference")
 
-        deployment_preference_collection.add(self.logical_id, self.DeploymentPreference)
+        deployment_preference_collection.add(
+            self.logical_id, self.DeploymentPreference, condition,
+        )
 
         if deployment_preference_collection.get(self.logical_id).enabled:
             if self.AutoPublishAlias is None:

tests/translator/model/preferences/test_deployment_preference.py

@@ -16,6 +16,7 @@ def setUp(self):
             "TriggerTargetArn": {"Ref": "MySNSTopic"},
             "TriggerName": "TestTrigger",
         }
+        self.condition = "condition"
         self.expected_deployment_preference = DeploymentPreference(
             self.deployment_type,
             self.pre_traffic_hook,
@@ -24,6 +25,7 @@ def setUp(self):
             True,
             self.role,
             self.trigger_configurations,
+            self.condition,
         )
 
     def test_from_dict_with_intrinsic_function_type(self):
@@ -37,6 +39,7 @@ def test_from_dict_with_intrinsic_function_type(self):
             True,
             self.role,
             self.trigger_configurations,
+            self.condition,
         )
 
         deployment_preference_yaml_dict = dict()
@@ -49,7 +52,7 @@ def test_from_dict_with_intrinsic_function_type(self):
         deployment_preference_yaml_dict["Role"] = self.role
         deployment_preference_yaml_dict["TriggerConfigurations"] = self.trigger_configurations
         deployment_preference_from_yaml_dict = DeploymentPreference.from_dict(
-            "logical_id", deployment_preference_yaml_dict
+            "logical_id", deployment_preference_yaml_dict, self.condition
         )
 
         self.assertEqual(expected_deployment_preference, deployment_preference_from_yaml_dict)
@@ -65,13 +68,13 @@ def test_from_dict(self):
         deployment_preference_yaml_dict["Role"] = self.role
         deployment_preference_yaml_dict["TriggerConfigurations"] = self.trigger_configurations
         deployment_preference_from_yaml_dict = DeploymentPreference.from_dict(
-            "logical_id", deployment_preference_yaml_dict
+            "logical_id", deployment_preference_yaml_dict, self.condition
         )
 
         self.assertEqual(self.expected_deployment_preference, deployment_preference_from_yaml_dict)
 
     def test_from_dict_with_disabled_preference_does_not_require_other_parameters(self):
-        expected_deployment_preference = DeploymentPreference(None, None, None, None, False, None, None)
+        expected_deployment_preference = DeploymentPreference(None, None, None, None, False, None, None, None)
 
         deployment_preference_yaml_dict = dict()
         deployment_preference_yaml_dict["Enabled"] = False
@@ -82,7 +85,7 @@ def test_from_dict_with_disabled_preference_does_not_require_other_parameters(se
         self.assertEqual(expected_deployment_preference, deployment_preference_from_yaml_dict)
 
     def test_from_dict_with_string_disabled_preference_does_not_require_other_parameters(self):
-        expected_deployment_preference = DeploymentPreference(None, None, None, None, False, None, None)
+        expected_deployment_preference = DeploymentPreference(None, None, None, None, False, None, None, None)
 
         deployment_preference_yaml_dict = dict()
         deployment_preference_yaml_dict["Enabled"] = "False"
@@ -93,7 +96,7 @@ def test_from_dict_with_string_disabled_preference_does_not_require_other_parame
         self.assertEqual(expected_deployment_preference, deployment_preference_from_yaml_dict)
 
     def test_from_dict_with_lowercase_string_disabled_preference_does_not_require_other_parameters(self):
-        expected_deployment_preference = DeploymentPreference(None, None, None, None, False, None, None)
+        expected_deployment_preference = DeploymentPreference(None, None, None, None, False, None, None, None)
 
         deployment_preference_yaml_dict = dict()
         deployment_preference_yaml_dict["Enabled"] = "false"

tests/translator/test_function_resources.py

@@ -144,7 +144,9 @@ def test_sam_function_with_deployment_preference(self, get_resolved_alias_name_m
         resources = sam_func.to_cloudformation(**kwargs)
 
         deployment_preference_collection.update_policy.assert_called_once_with(self.sam_func.logical_id)
-        deployment_preference_collection.add.assert_called_once_with(self.sam_func.logical_id, deploy_preference_dict)
+        deployment_preference_collection.add.assert_called_once_with(
+            self.sam_func.logical_id, deploy_preference_dict, None
+        )
 
         aliases = [r.to_dict() for r in resources if r.resource_type == LambdaAlias.resource_type]
 
@@ -222,7 +224,7 @@ def test_sam_function_with_disabled_deployment_preference_does_not_add_update_po
 
         resources = sam_func.to_cloudformation(**kwargs)
 
-        preference_collection.add.assert_called_once_with(sam_func.logical_id, deploy_preference_dict)
+        preference_collection.add.assert_called_once_with(sam_func.logical_id, deploy_preference_dict, None)
         preference_collection.get.assert_called_once_with(sam_func.logical_id)
         self.intrinsics_resolver_mock.resolve_parameter_refs.assert_called_with(enabled)
         aliases = [r.to_dict() for r in resources if r.resource_type == LambdaAlias.resource_type]
@@ -318,7 +320,9 @@ def test_sam_function_with_deployment_preference_intrinsic_ref_enabled_boolean_p
         resources = sam.to_cloudformation(**kwargs)
 
         deployment_preference_collection.update_policy.assert_called_once_with(self.sam_func.logical_id)
-        deployment_preference_collection.add.assert_called_once_with(self.sam_func.logical_id, deploy_preference_dict)
+        deployment_preference_collection.add.assert_called_once_with(
+            self.sam_func.logical_id, deploy_preference_dict, None
+        )
         self.intrinsics_resolver_mock.resolve_parameter_refs.assert_any_call(enabled)
 
         aliases = [r.to_dict() for r in resources if r.resource_type == LambdaAlias.resource_type]