feat: Add support for FindInMap in serverless app plugin

samtranslator/intrinsics/actions.py

@@ -503,3 +503,50 @@ def _get_resolved_dictionary(self, input_dict, key, resolved_value, remaining):
             input_dict[key] = [resolved_value] + remaining
 
         return input_dict
+
+
+class FindInMapAction(Action):
+    """
+    This action can't be used along with other actions.
+    """
+    intrinsic_name = "Fn::FindInMap"
+
+    def resolve_parameter_refs(self, input_dict, parameters):
+        """
+        Recursively resolves "Fn::FindInMap"references that are present in the mappings and returns the value.
+        If it is not in mappings, this method simply returns the input unchanged.
+
+        :param input_dict: Dictionary representing the FindInMap function. Must contain only one key and it
+                           should be "Fn::FindInMap".
+
+        :param parameters: Dictionary of mappings from the SAM template
+        """
+        if not self.can_handle(input_dict):
+            return input_dict
+
+        value = input_dict[self.intrinsic_name]
+
+        if not isinstance(value, list) or len(value) != 3:
+            return input_dict
+
+        map_name = self.resolve_parameter_refs(value[0], parameters)
+        top_level_key = self.resolve_parameter_refs(value[1], parameters)
+        second_level_key = self.resolve_parameter_refs(value[2], parameters)
+
+        if not isinstance(map_name, basestring) or \
+                not isinstance(top_level_key, basestring) or \
+                not isinstance(second_level_key, basestring):
+            return input_dict
+
+        if map_name not in parameters or \
+                top_level_key not in parameters[map_name] or \
+                second_level_key not in parameters[map_name][top_level_key]:
+            return input_dict
+
+        return parameters[map_name][top_level_key][second_level_key]
+
+    def resolve_resource_refs(self, input_dict, supported_resource_refs):
+        return input_dict
+
+    def resolve_resource_id_refs(self, input_dict, supported_resource_id_refs):
+        return input_dict

samtranslator/plugins/application/serverless_app_plugin.py

@@ -2,12 +2,15 @@
 from botocore.exceptions import ClientError, EndpointConnectionError
 import logging
 from time import sleep, time
+import copy
 
 from samtranslator.model.exceptions import InvalidResourceException
 from samtranslator.plugins import BasePlugin
 from samtranslator.plugins.exceptions import InvalidPluginException
 from samtranslator.public.sdk.resource import SamResourceType
 from samtranslator.public.sdk.template import SamTemplate
+from samtranslator.intrinsics.resolver import IntrinsicsResolver
+from samtranslator.intrinsics.actions import FindInMapAction
 
 
 class ServerlessAppPlugin(BasePlugin):
@@ -35,7 +38,7 @@ class ServerlessAppPlugin(BasePlugin):
     LOCATION_KEY = 'Location'
     TEMPLATE_URL_KEY = 'TemplateUrl'
 
-    def __init__(self, sar_client=None, wait_for_template_active_status=False, validate_only=False):
+    def __init__(self, sar_client=None, wait_for_template_active_status=False, validate_only=False, parameters={}):
         """
         Initialize the plugin.
 
@@ -50,6 +53,7 @@ def __init__(self, sar_client=None, wait_for_template_active_status=False, valid
         self._sar_client = sar_client
         self._wait_for_template_active_status = wait_for_template_active_status
         self._validate_only = validate_only
+        self._parameters = parameters
 
         # make sure the flag combination makes sense
         if self._validate_only is True and self._wait_for_template_active_status is True:
@@ -68,6 +72,7 @@ def on_before_transform_template(self, template_dict):
         :return: Nothing
         """
         template = SamTemplate(template_dict)
+        intrinsic_resolvers = self._get_intrinsic_resolvers(template_dict.get('Mappings', {}))
 
         service_call = None
         if self._validate_only:
@@ -79,8 +84,11 @@ def on_before_transform_template(self, template_dict):
                 # Handle these cases in the on_before_transform_resource event
                 continue
 
-            app_id = app.properties[self.LOCATION_KEY].get(self.APPLICATION_ID_KEY)
-            semver = app.properties[self.LOCATION_KEY].get(self.SEMANTIC_VERSION_KEY)
+            app_id = self._replace_value(app.properties[self.LOCATION_KEY],
+                                         self.APPLICATION_ID_KEY, intrinsic_resolvers)
+            semver = self._replace_value(app.properties[self.LOCATION_KEY],
+                                         self.SEMANTIC_VERSION_KEY, intrinsic_resolvers)
+
             key = (app_id, semver)
             if key not in self._applications:
                 try:
@@ -92,6 +100,21 @@ def on_before_transform_template(self, template_dict):
                     # Catch all InvalidResourceExceptions, raise those in the before_resource_transform target.
                     self._applications[key] = e
 
+    def _replace_value(self, input_dict, key, intrinsic_resolvers):
+        value = self._resolve_location_value(input_dict.get(key), intrinsic_resolvers)
+        input_dict[key] = value
+        return value
+
+    def _get_intrinsic_resolvers(self, mappings):
+        return [IntrinsicsResolver(self._parameters),
+                IntrinsicsResolver(mappings, {FindInMapAction.intrinsic_name: FindInMapAction()})]
+
+    def _resolve_location_value(self, value, intrinsic_resolvers):
+        resolved_value = copy.deepcopy(value)
+        for intrinsic_resolver in intrinsic_resolvers:
+            resolved_value = intrinsic_resolver.resolve_parameter_refs(resolved_value)
+        return resolved_value
+
     def _can_process_application(self, app):
         """
         Determines whether or not the on_before_transform_template event can process this application

samtranslator/translator/translator.py

@@ -46,10 +46,10 @@ def translate(self, sam_template, parameter_values):
         :returns: a copy of the template with SAM resources replaced with the corresponding CloudFormation, which may \
                 be dumped into a valid CloudFormation JSON or YAML template
         """
-        # Create & Install plugins
-        sam_plugins = prepare_plugins(self.plugins)
         parameter_values = self._add_default_parameter_values(sam_template, parameter_values)
         parameter_values = self._add_pseudo_parameter_values(parameter_values)
+        # Create & Install plugins
+        sam_plugins = prepare_plugins(self.plugins, parameter_values)
 
         self.sam_parser.parse(
             sam_template=sam_template,
@@ -214,12 +214,13 @@ def _add_pseudo_parameter_values(self, parameter_values):
         return updated_parameter_values
 
 
-def prepare_plugins(plugins):
+def prepare_plugins(plugins, parameters={}):
     """
     Creates & returns a plugins object with the given list of plugins installed. In addition to the given plugins,
     we will also install a few "required" plugins that are necessary to provide complete support for SAM template spec.
 
-    :param list of samtranslator.plugins.BasePlugin plugins: List of plugins to install
+    :param plugins: list of samtranslator.plugins.BasePlugin plugins: List of plugins to install
+    :param parameters: Dictionary of parameter values
     :return samtranslator.plugins.SamPlugins: Instance of `SamPlugins`
     """
 
@@ -234,7 +235,7 @@ def prepare_plugins(plugins):
 
     # If a ServerlessAppPlugin does not yet exist, create one and add to the beginning of the required plugins list.
     if not any(isinstance(plugin, ServerlessAppPlugin) for plugin in plugins):
-        required_plugins.insert(0, ServerlessAppPlugin())
+        required_plugins.insert(0, ServerlessAppPlugin(parameters=parameters))
 
     # Execute customer's plugins first before running SAM plugins. It is very important to retain this order because
     # other plugins will be dependent on this ordering.

tests/intrinsics/test_actions.py

@@ -1,6 +1,6 @@
 from unittest import TestCase
 from mock import patch, Mock
-from samtranslator.intrinsics.actions import Action, RefAction, SubAction, GetAttAction
+from samtranslator.intrinsics.actions import Action, RefAction, SubAction, GetAttAction, FindInMapAction
 from samtranslator.intrinsics.resource_refs import SupportedResourceReferences
 
 class TestAction(TestCase):
@@ -932,3 +932,203 @@ def test_return_value_if_cannot_handle(self, can_handle_mock):
         getatt = GetAttAction()
         can_handle_mock.return_value = False # Simulate failure to handle the input. Result should be same as input
         self.assertEqual(expected, getatt.resolve_resource_id_refs(input, self.supported_resource_id_refs))
+
+
+class TestFindInMapCanResolveParameterRefs(TestCase):
+
+    def setUp(self):
+        self.ref = FindInMapAction()
+
+    @patch.object(FindInMapAction, "can_handle")
+    def test_cannot_handle(self, can_handle_mock):
+        input = {
+            "Fn::FindInMap": ["a", "b", "c"]
+        }
+        can_handle_mock.return_value = False
+        output = self.ref.resolve_parameter_refs(input, {})
+
+        self.assertEqual(input, output)
+
+    def test_value_not_list(self):
+        input = {
+            "Fn::FindInMap": "a string"
+        }
+        output = self.ref.resolve_parameter_refs(input, {})
+
+        self.assertEqual(input, output)
+
+    def test_value_not_list_of_length_three(self):
+        input = {
+            "Fn::FindInMap": ["a string"]
+        }
+        output = self.ref.resolve_parameter_refs(input, {})
+
+        self.assertEqual(input, output)
+
+    def test_mapping_not_string(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "value4"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": [["MapA"], "TopKey2", "SecondKey1"]
+        }
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(input, output)
+
+    def test_top_level_key_not_string(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "value4"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": ["MapA", ["TopKey2"], "SecondKey1"]
+        }
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(input, output)
+
+    def test_second_level_key_not_string(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "value4"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": ["MapA", "TopKey1", ["SecondKey2"]]
+        }
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(input, output)
+
+    def test_mapping_not_found(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "value4"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": ["MapB", "TopKey2", "SecondKey1"]
+        }
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(input, output)
+
+    def test_top_level_key_not_found(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "value4"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": ["MapA", "TopKey3", "SecondKey1"]
+        }
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(input, output)
+
+    def test_second_level_key_not_found(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "value4"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": ["MapA", "TopKey1", "SecondKey1"]
+        }
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(input, output)
+
+    def test_one_level_find_in_mappings(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "value4"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": ["MapA", "TopKey2", "SecondKey1"]
+        }
+        expected = "value4"
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(expected, output)
+
+    def test_nested_find_in_mappings(self):
+        mappings = {
+            "MapA":{
+                "TopKey1": {
+                   "SecondKey2": "value3"
+               },
+                "TopKey2": {
+                    "SecondKey1": "TopKey1"
+                }
+            }
+        }
+        input = {
+            "Fn::FindInMap": ["MapA", {"Fn::FindInMap": ["MapA", "TopKey2", "SecondKey1"]}, "SecondKey2"]
+        }
+        expected = "value3"
+        output = self.ref.resolve_parameter_refs(input, mappings)
+
+        self.assertEqual(expected, output)
+
+
+class TestFindInMapCanResolveResourceRefs(TestCase):
+
+    def test_must_do_nothing(self):
+        input = "foo"
+        expected = "foo"
+        supported_resource_refs = Mock()
+        self.assertEqual(expected, FindInMapAction().resolve_resource_refs(input, supported_resource_refs))
+
+        supported_resource_refs.assert_not_called()
+
+
+class TestFindInMapCanResolveResourceIdRefs(TestCase):
+
+    def test_must_do_nothing(self):
+        input = "foo"
+        expected = "foo"
+        supported_resource_refs = Mock()
+        self.assertEqual(expected, FindInMapAction().resolve_resource_id_refs(input, supported_resource_refs))
+
+        supported_resource_refs.assert_not_called()