fix: apply resource conditions to DeploymentPreference resources

samtranslator/model/preferences/deployment_preference.py

@@ -75,8 +75,6 @@ def from_dict(cls, logical_id, deployment_preference_dict, condition=None):
         role = deployment_preference_dict.get("Role", None)
         trigger_configurations = deployment_preference_dict.get("TriggerConfigurations", None)
         passthrough_condition = deployment_preference_dict.get("PassthroughCondition", False)
-        # FIXME: We should support not only true/false, but also yes/no, on/off? See https://yaml.org/type/bool.html 
-        passthrough_condition = True if passthrough_condition in [True, "true", "True"] else False
 
         return DeploymentPreference(
             deployment_type,

samtranslator/model/sam_resources.py

@@ -1,5 +1,7 @@
 """ SAM macro definitions """
 import copy
+from typing import Union
+from samtranslator.intrinsics.resolver import IntrinsicsResolver
 
 import samtranslator.model.eventsources
 import samtranslator.model.eventsources.pull
@@ -853,11 +855,12 @@ def _validate_deployment_preference_and_add_update_policy(
             self.DeploymentPreference["Type"] = preference_type
 
         if "PassthroughCondition" in self.DeploymentPreference:
-            # resolve intrinsics and mappings for PassthroughCondition
-            passthrough_condition = self.DeploymentPreference["PassthroughCondition"]
-            passthrough_condition = intrinsics_resolver.resolve_parameter_refs(passthrough_condition)
-            passthrough_condition = mappings_resolver.resolve_parameter_refs(passthrough_condition)
-            self.DeploymentPreference["PassthroughCondition"] = passthrough_condition
+            self.DeploymentPreference["PassthroughCondition"] = self._resolve_property_to_boolean(
+                self.DeploymentPreference["PassthroughCondition"],
+                "PassthroughCondition",
+                intrinsics_resolver,
+                mappings_resolver,
+            )
         elif feature_toggle:
             self.DeploymentPreference["PassthroughCondition"] = feature_toggle.is_enabled(
                 "deployment_preference_condition_fix"
@@ -886,6 +889,40 @@ def _validate_deployment_preference_and_add_update_policy(
                 "UpdatePolicy", deployment_preference_collection.update_policy(self.logical_id).to_dict()
             )
 
+    def _resolve_property_to_boolean(
+        self,
+        property_value: Union[bool, str, dict],
+        property_name: str,
+        intrinsics_resolver: IntrinsicsResolver,
+        mappings_resolver: IntrinsicsResolver,
+    ) -> bool:
+        """
+        Resolves intrinsics, if any, and/or converts string in a given property to boolean.
+        Raises InvalidResourceException if can't resolve intrinsic or can't resolve string to boolean
+
+        :param property_value: property value to resolve
+        :param property_name: name/key of property to resolve
+        :param intrinsics_resolver: resolves intrinsics
+        :param mappings_resolver: resolves FindInMap
+        :return bool: resolved boolean value
+        """
+        processed_property_value = intrinsics_resolver.resolve_parameter_refs(property_value)
+        processed_property_value = mappings_resolver.resolve_parameter_refs(processed_property_value)
+
+        # FIXME: We should support not only true/false, but also yes/no, on/off? See https://yaml.org/type/bool.html
+        if processed_property_value in [True, "true", "True"]:
+            return True
+        elif processed_property_value in [False, "false", "False"]:
+            return False
+        elif is_intrinsic(processed_property_value):  # couldn't resolve intrinsic
+            raise InvalidResourceException(
+                self.logical_id,
+                f"Unsupported intrinsic: the only intrinsic functions supported for "
+                f"property {property_name} are FindInMap and parameter Refs.",
+            )
+        else:
+            raise InvalidResourceException(self.logical_id, f"Invalid value for property {property_name}.")
+
     def _construct_function_url(self, lambda_function, lambda_alias):
         """
         This method is used to construct a lambda url resource

tests/translator/input/error_function_with_deployment_preference_passthrough_condition_with_invalid_values.yaml

@@ -0,0 +1,72 @@
+# Tests unsupported intrinsic and invalid type in the PassthroughCondition property
+
+Mappings:
+  HelloWorldMap:
+    hello:
+      key1: true
+      key2: false
+    world:
+      key1: false
+      key2: true
+
+Parameters:
+  FnName:
+    Type: String
+  ProvisionedConcurrency:
+    Type: String
+    Default: 10
+  EnableAliasProvisionedConcurrency:
+    Type: String
+    AllowedValues:
+      - true
+      - false
+    Default: true
+  DefaultTrueParam:
+    Type: String
+    Default: "true"
+  DefaultFalseParam:
+    Type: String
+    Default: "false"
+  HelloWorldParam:
+    Type: String
+    Default: "hello"
+
+Conditions:
+  AliasProvisionedConcurrencyEnabled: !Equals [!Ref EnableAliasProvisionedConcurrency, true]
+  FunctionCondition: !Equals [true, true]
+
+Resources:
+  InvalidType:
+    Type: 'AWS::Serverless::Function'
+    Condition: FunctionCondition
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        PassthroughCondition: ["hello"]
+      ProvisionedConcurrencyConfig: !If
+        - AliasProvisionedConcurrencyEnabled
+        - ProvisionedConcurrentExecutions: !Ref ProvisionedConcurrency
+        - !Ref 'AWS::NoValue'
+
+  UnsupportedIntrinsic:
+    Type: 'AWS::Serverless::Function'
+    Condition: FunctionCondition
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        PassthroughCondition: !If
+          - AliasProvisionedConcurrencyEnabled
+          - true
+          - false 
+      ProvisionedConcurrencyConfig: !If
+        - AliasProvisionedConcurrencyEnabled
+        - ProvisionedConcurrentExecutions: !Ref ProvisionedConcurrency
+        - !Ref 'AWS::NoValue'

tests/translator/input/function_with_deployment_preference_passthrough_condition_with_supported_intrinsics.yaml

@@ -0,0 +1,110 @@
+# Tests supported intrinsics in the PassthroughCondition property
+
+Mappings:
+  HelloWorldMap:
+    hello:
+      key1: true
+      key2: false
+    world:
+      key1: false
+      key2: true
+
+Parameters:
+  FnName:
+    Type: String
+  ProvisionedConcurrency:
+    Type: String
+    Default: 10
+  EnableAliasProvisionedConcurrency:
+    Type: String
+    AllowedValues:
+      - true
+      - false
+    Default: true
+  DefaultTrueParam:
+    Type: String
+    Default: "true"
+  DefaultFalseParam:
+    Type: String
+    Default: "false"
+  HelloParam:
+    Type: String
+    Default: "hello"
+  WorldParam:
+    Type: String
+    Default: "world"
+
+Conditions:
+  AliasProvisionedConcurrencyEnabled: !Equals [!Ref EnableAliasProvisionedConcurrency, true]
+  FunctionCondition: !Equals [true, true]
+
+Resources:
+  TrueRef:
+    Type: 'AWS::Serverless::Function'
+    Condition: FunctionCondition
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        PassthroughCondition: !Ref DefaultTrueParam
+      ProvisionedConcurrencyConfig: !If
+        - AliasProvisionedConcurrencyEnabled
+        - ProvisionedConcurrentExecutions: !Ref ProvisionedConcurrency
+        - !Ref 'AWS::NoValue'
+
+  FalseRef:
+    Type: 'AWS::Serverless::Function'
+    Condition: FunctionCondition
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        PassthroughCondition: !Ref DefaultFalseParam 
+      ProvisionedConcurrencyConfig: !If
+        - AliasProvisionedConcurrencyEnabled
+        - ProvisionedConcurrentExecutions: !Ref ProvisionedConcurrency
+        - !Ref 'AWS::NoValue'
+
+  TrueFindInMap:
+    Type: 'AWS::Serverless::Function'
+    Condition: FunctionCondition
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        PassthroughCondition: !FindInMap
+          - HelloWorldMap
+          - !Ref HelloParam
+          - key1
+      ProvisionedConcurrencyConfig: !If
+        - AliasProvisionedConcurrencyEnabled
+        - ProvisionedConcurrentExecutions: !Ref ProvisionedConcurrency
+        - !Ref 'AWS::NoValue'
+
+  FalseFindInMap:
+    Type: 'AWS::Serverless::Function'
+    Condition: FunctionCondition
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: hello.handler
+      Runtime: python2.7
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: Linear10PercentEvery3Minutes
+        PassthroughCondition: !FindInMap
+          - HelloWorldMap
+          - !Ref WorldParam 
+          - key1
+      ProvisionedConcurrencyConfig: !If
+        - AliasProvisionedConcurrencyEnabled
+        - ProvisionedConcurrentExecutions: !Ref ProvisionedConcurrency
+        - !Ref 'AWS::NoValue'

tests/translator/model/preferences/test_deployment_preference.py

@@ -17,33 +17,50 @@ def setUp(self):
             "TriggerName": "TestTrigger",
         }
         self.condition = "condition"
-        self.expected_deployment_preference = DeploymentPreference(
-            self.deployment_type,
+
+    def test_from_dict_with_intrinsic_function_type(self):
+
+        type = {"Ref": "SomeType"}
+        expected_deployment_preference = DeploymentPreference(
+            type,
             self.pre_traffic_hook,
             self.post_traffic_hook,
             self.alarms,
             True,
             self.role,
             self.trigger_configurations,
-            self.condition,
+            None,
         )
 
-    def test_from_dict_with_intrinsic_function_type(self):
+        deployment_preference_yaml_dict = dict()
+        deployment_preference_yaml_dict["Type"] = type
+        deployment_preference_yaml_dict["Hooks"] = {
+            "PreTraffic": self.pre_traffic_hook,
+            "PostTraffic": self.post_traffic_hook,
+        }
+        deployment_preference_yaml_dict["Alarms"] = self.alarms
+        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, self.condition
+        )
 
-        type = {"Ref": "SomeType"}
+        self.assertEqual(expected_deployment_preference, deployment_preference_from_yaml_dict)
+
+    def test_from_dict(self):
         expected_deployment_preference = DeploymentPreference(
-            type,
+            self.deployment_type,
             self.pre_traffic_hook,
             self.post_traffic_hook,
             self.alarms,
             True,
             self.role,
             self.trigger_configurations,
-            self.condition,
+            None,
         )
 
         deployment_preference_yaml_dict = dict()
-        deployment_preference_yaml_dict["Type"] = type
+        deployment_preference_yaml_dict["Type"] = self.deployment_type
         deployment_preference_yaml_dict["Hooks"] = {
             "PreTraffic": self.pre_traffic_hook,
             "PostTraffic": self.post_traffic_hook,
@@ -57,7 +74,18 @@ def test_from_dict_with_intrinsic_function_type(self):
 
         self.assertEqual(expected_deployment_preference, deployment_preference_from_yaml_dict)
 
-    def test_from_dict(self):
+    def test_from_dict_with_passthrough_condition(self):
+        expected_deployment_preference = DeploymentPreference(
+            self.deployment_type,
+            self.pre_traffic_hook,
+            self.post_traffic_hook,
+            self.alarms,
+            True,
+            self.role,
+            self.trigger_configurations,
+            self.condition,
+        )
+
         deployment_preference_yaml_dict = dict()
         deployment_preference_yaml_dict["Type"] = self.deployment_type
         deployment_preference_yaml_dict["Hooks"] = {
@@ -67,11 +95,12 @@ def test_from_dict(self):
         deployment_preference_yaml_dict["Alarms"] = self.alarms
         deployment_preference_yaml_dict["Role"] = self.role
         deployment_preference_yaml_dict["TriggerConfigurations"] = self.trigger_configurations
+        deployment_preference_yaml_dict["PassthroughCondition"] = True
         deployment_preference_from_yaml_dict = DeploymentPreference.from_dict(
             "logical_id", deployment_preference_yaml_dict, self.condition
         )
 
-        self.assertEqual(self.expected_deployment_preference, deployment_preference_from_yaml_dict)
+        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)