Add proptags for codedeploy resources

samtranslator/model/preferences/deployment_preference.py

@@ -22,6 +22,8 @@
 :param enabled: Whether this deployment preference is enabled (true by default)
 :param trigger_configurations: Information about triggers associated with the deployment group. Duplicates are
     not allowed.
+:param tags: Tags to propagate to CodeDeploy resources when propagate_tags is enabled
+:param propagate_tags: Whether to propagate tags to CodeDeploy resources
 """
 DeploymentPreferenceTuple = namedtuple(
     "DeploymentPreferenceTuple",
@@ -34,6 +36,8 @@
         "role",
         "trigger_configurations",
         "condition",
+        "tags",
+        "propagate_tags",
     ],
 )
 
@@ -46,18 +50,20 @@ class DeploymentPreference(DeploymentPreferenceTuple):
     """
 
     @classmethod
-    def from_dict(cls, logical_id, deployment_preference_dict, condition=None):  # type: ignore[no-untyped-def]
+    def from_dict(cls, logical_id, deployment_preference_dict, condition=None, tags=None, propagate_tags=False):  # type: ignore[no-untyped-def]
         """
         :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
+        :param tags: tags from the SAM resource to propagate to CodeDeploy resources
+        :param propagate_tags: whether to propagate tags to CodeDeploy resources
         :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, None)
+            return DeploymentPreference(None, None, None, None, False, None, None, None, None, None)
 
         if "Type" not in deployment_preference_dict:
             raise InvalidResourceException(logical_id, "'DeploymentPreference' is missing required Property 'Type'")
@@ -85,4 +91,6 @@ def from_dict(cls, logical_id, deployment_preference_dict, condition=None):  # t
             role,
             trigger_configurations,
             condition if passthrough_condition else None,
+            tags if propagate_tags and tags else None,
+            propagate_tags,
         )

samtranslator/model/preferences/deployment_preference_collection.py

@@ -14,6 +14,7 @@
     ref,
     validate_intrinsic_if_items,
 )
+from samtranslator.model.tags.resource_tagging import get_tag_list
 from samtranslator.model.update_policy import UpdatePolicy
 from samtranslator.translator.arn_generator import ArnGenerator
 
@@ -52,20 +53,29 @@ def __init__(self) -> None:
         """
         self._resource_preferences: Dict[str, Any] = {}
 
-    def add(self, logical_id: str, deployment_preference_dict: Dict[str, Any], condition: Optional[str] = None) -> None:
+    def add(
+        self,
+        logical_id: str,
+        deployment_preference_dict: Dict[str, Any],
+        condition: Optional[str] = None,
+        tags: Optional[Dict[str, Any]] = None,
+        propagate_tags: Optional[bool] = False,
+    ) -> 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
+        :param tags: tags from the SAM resource to propagate to CodeDeploy resources
+        :param propagate_tags: whether to propagate tags to CodeDeploy resources
         """
         if logical_id in self._resource_preferences:
             raise ValueError(f"logical_id {logical_id} previously added to this deployment_preference_collection")
 
         self._resource_preferences[logical_id] = DeploymentPreference.from_dict(  # type: ignore[no-untyped-call]
-            logical_id, deployment_preference_dict, condition
+            logical_id, deployment_preference_dict, condition, tags, propagate_tags
         )
 
     def get(self, logical_id: str) -> DeploymentPreference:
@@ -127,6 +137,13 @@ def enabled_logical_ids(self) -> List[str]:
     def get_codedeploy_application(self) -> CodeDeployApplication:
         codedeploy_application_resource = CodeDeployApplication(CODEDEPLOY_APPLICATION_LOGICAL_ID)
         codedeploy_application_resource.ComputePlatform = "Lambda"
+
+        merged_tags: Dict[str, Any] = {}
+        for preference in self._resource_preferences.values():
+            if preference.enabled and preference.propagate_tags and preference.tags:
+                merged_tags.update(preference.tags)
+        if merged_tags:
+            codedeploy_application_resource.Tags = get_tag_list(merged_tags)
         if self.needs_resource_condition():
             conditions = self.get_all_deployment_conditions()
             condition_name = CODE_DEPLOY_CONDITION_NAME
@@ -165,6 +182,14 @@ def get_codedeploy_iam_role(self) -> IAMRole:
             if len(conditions) <= 1:
                 condition_name = conditions.pop()
             iam_role.set_resource_attribute("Condition", condition_name)
+
+        merged_tags: Dict[str, Any] = {}
+        for preference in self._resource_preferences.values():
+            if preference.enabled and preference.propagate_tags and preference.tags:
+                merged_tags.update(preference.tags)
+        if merged_tags:
+            iam_role.Tags = get_tag_list(merged_tags)
+
         return iam_role
 
     def deployment_group(self, function_logical_id: str) -> CodeDeployDeploymentGroup:
@@ -201,6 +226,9 @@ def deployment_group(self, function_logical_id: str) -> CodeDeployDeploymentGrou
         if deployment_preference.trigger_configurations:
             deployment_group.TriggerConfigurations = deployment_preference.trigger_configurations
 
+        if deployment_preference.tags:
+            deployment_group.Tags = get_tag_list(deployment_preference.tags)
+
         if deployment_preference.condition:
             deployment_group.set_resource_attribute("Condition", deployment_preference.condition)
 

samtranslator/model/sam_resources.py

@@ -1,4 +1,4 @@
-"""SAM macro definitions"""
+"""SAM macro definitions"""
 
 import copy
 import re
@@ -1308,6 +1308,8 @@ def _validate_deployment_preference_and_add_update_policy(  # noqa: PLR0913
             self.logical_id,
             self.DeploymentPreference,
             passthrough_resource_attributes.get("Condition"),
+            self.Tags,
+            self.PropagateTags,
         )
 
         if deployment_preference_collection.get(self.logical_id).enabled:

tests/translator/input/function_with_propagate_tags_and_no_tags.yaml

@@ -6,7 +6,7 @@ Resources:
         exports.handler = async () => ‘Hello World!'
       Handler: index.handler
       Runtime: nodejs18.x
-      # PropagateTags: True
+      PropagateTags: true
       AutoPublishAlias: Live
       Events:
         CWEvent:

tests/translator/model/preferences/test_deployment_preference.py

@@ -29,6 +29,8 @@ def test_from_dict_with_intrinsic_function_type(self):
             self.role,
             self.trigger_configurations,
             None,
+            None,
+            False,
         )
 
         deployment_preference_yaml_dict = dict()
@@ -56,6 +58,8 @@ def test_from_dict(self):
             self.role,
             self.trigger_configurations,
             None,
+            None,
+            False,
         )
 
         deployment_preference_yaml_dict = dict()
@@ -83,6 +87,8 @@ def test_from_dict_with_passthrough_condition(self):
             self.role,
             self.trigger_configurations,
             self.condition,
+            None,
+            False,
         )
 
         deployment_preference_yaml_dict = dict()
@@ -102,7 +108,9 @@ def test_from_dict_with_passthrough_condition(self):
         self.assertEqual(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, None)
+        expected_deployment_preference = DeploymentPreference(
+            None, None, None, None, False, None, None, None, None, None
+        )
 
         deployment_preference_yaml_dict = dict()
         deployment_preference_yaml_dict["Enabled"] = False
@@ -113,7 +121,9 @@ 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, None)
+        expected_deployment_preference = DeploymentPreference(
+            None, None, None, None, False, None, None, None, None, None
+        )
 
         deployment_preference_yaml_dict = dict()
         deployment_preference_yaml_dict["Enabled"] = "False"
@@ -124,7 +134,9 @@ 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, None)
+        expected_deployment_preference = DeploymentPreference(
+            None, None, None, None, False, None, None, None, None, None
+        )
 
         deployment_preference_yaml_dict = dict()
         deployment_preference_yaml_dict["Enabled"] = "false"

tests/translator/output/aws-cn/function_with_events_and_propagate_tags.json

@@ -311,7 +311,17 @@
             "DeploymentRole",
             "Arn"
           ]
-        }
+        },
+        "Tags": [
+          {
+            "Key": "Key1",
+            "Value": "Value1"
+          },
+          {
+            "Key": "Key2",
+            "Value": "Value2"
+          }
+        ]
       },
       "Type": "AWS::CodeDeploy::DeploymentGroup"
     },
@@ -520,7 +530,17 @@
     },
     "ServerlessDeploymentApplication": {
       "Properties": {
-        "ComputePlatform": "Lambda"
+        "ComputePlatform": "Lambda",
+        "Tags": [
+          {
+            "Key": "Key1",
+            "Value": "Value1"
+          },
+          {
+            "Key": "Key2",
+            "Value": "Value2"
+          }
+        ]
       },
       "Type": "AWS::CodeDeploy::Application"
     },

tests/translator/output/aws-us-gov/function_with_events_and_propagate_tags.json

@@ -311,7 +311,17 @@
             "DeploymentRole",
             "Arn"
           ]
-        }
+        },
+        "Tags": [
+          {
+            "Key": "Key1",
+            "Value": "Value1"
+          },
+          {
+            "Key": "Key2",
+            "Value": "Value2"
+          }
+        ]
       },
       "Type": "AWS::CodeDeploy::DeploymentGroup"
     },
@@ -520,7 +530,17 @@
     },
     "ServerlessDeploymentApplication": {
       "Properties": {
-        "ComputePlatform": "Lambda"
+        "ComputePlatform": "Lambda",
+        "Tags": [
+          {
+            "Key": "Key1",
+            "Value": "Value1"
+          },
+          {
+            "Key": "Key2",
+            "Value": "Value2"
+          }
+        ]
       },
       "Type": "AWS::CodeDeploy::Application"
     },

tests/translator/output/function_with_events_and_propagate_tags.json

@@ -311,7 +311,17 @@
             "DeploymentRole",
             "Arn"
           ]
-        }
+        },
+        "Tags": [
+          {
+            "Key": "Key1",
+            "Value": "Value1"
+          },
+          {
+            "Key": "Key2",
+            "Value": "Value2"
+          }
+        ]
       },
       "Type": "AWS::CodeDeploy::DeploymentGroup"
     },
@@ -520,7 +530,17 @@
     },
     "ServerlessDeploymentApplication": {
       "Properties": {
-        "ComputePlatform": "Lambda"
+        "ComputePlatform": "Lambda",
+        "Tags": [
+          {
+            "Key": "Key1",
+            "Value": "Value1"
+          },
+          {
+            "Key": "Key2",
+            "Value": "Value2"
+          }
+        ]
       },
       "Type": "AWS::CodeDeploy::Application"
     },

tests/translator/test_function_resources.py

@@ -149,7 +149,7 @@ def test_sam_function_with_deployment_preference(self, get_resolved_alias_name_m
 
         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, None
+            self.sam_func.logical_id, deploy_preference_dict, None, None, None
         )
 
         aliases = [r.to_dict() for r in resources if r.resource_type == LambdaAlias.resource_type]
@@ -230,7 +230,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, None)
+        preference_collection.add.assert_called_once_with(sam_func.logical_id, deploy_preference_dict, None, None, None)
         preference_collection.get.assert_called_once_with(sam_func.logical_id)
         self.intrinsics_resolver_mock.resolve_parameter_refs.assert_has_calls([call(enabled), call(None)])
         aliases = [r.to_dict() for r in resources if r.resource_type == LambdaAlias.resource_type]
@@ -330,7 +330,7 @@ def test_sam_function_with_deployment_preference_intrinsic_ref_enabled_boolean_p
 
         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, None
+            self.sam_func.logical_id, deploy_preference_dict, None, None, None
         )
         self.intrinsics_resolver_mock.resolve_parameter_refs.assert_any_call(enabled)
 
@@ -451,7 +451,7 @@ def test_sam_function_with_deployment_preference_passthrough_condition_through_p
 
         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, "Condition1"
+            self.sam_func.logical_id, deploy_preference_dict, "Condition1", None, None
         )
 
         aliases = [r.to_dict() for r in resources if r.resource_type == LambdaAlias.resource_type]
@@ -502,7 +502,7 @@ def test_sam_function_with_deployment_preference_passthrough_condition_through_f
 
         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, "Condition1"
+            self.sam_func.logical_id, deploy_preference_dict, "Condition1", None, None
         )
 
         aliases = [r.to_dict() for r in resources if r.resource_type == LambdaAlias.resource_type]