Verify profile variables have been replaced (#264)

Author: hoffa

samtranslator/model/connector_profiles/profile.py

@@ -13,6 +13,16 @@ def get_profile(source_type: str, dest_type: str):
     return PROFILE["Permissions"].get(source_type, {}).get(dest_type)
 
 
+def verify_profile_variables_replaced(obj: Any) -> None:
+    """
+    Verifies all profile variables have been replaced; throws ValueError if not.
+    """
+    s = json.dumps(obj)
+    matches = re.findall(r"%{[\w\.]+}", s)
+    if matches:
+        raise ValueError(f"The following variables have not been replaced: {matches}")
+
+
 def profile_replace(obj: Any, replacements: Dict[str, Any]):
     """
     This function is used to recursively replace all keys in 'replacements' found

samtranslator/model/sam_resources.py

@@ -13,6 +13,7 @@
     ConnectorProfile,
     profile_replace,
     get_profile,
+    verify_profile_variables_replaced,
 )
 
 import samtranslator.model.eventsources
@@ -1694,6 +1695,8 @@ def to_cloudformation(self, **kwargs) -> List:
         except ValueError as e:
             raise InvalidResourceException(self.logical_id, str(e))
 
+        verify_profile_variables_replaced(profile_properties)
+
         generated_resources: List[Resource] = []
         if profile_type == "AWS_IAM_ROLE_MANAGED_POLICY":
             generated_resources.append(

tests/model/connector_profiles/test_profile.py

@@ -1,6 +1,8 @@
 from unittest import TestCase
 
-from samtranslator.model.connector_profiles.profile import profile_replace
+from parameterized import parameterized
+
+from samtranslator.model.connector_profiles.profile import profile_replace, verify_profile_variables_replaced
 
 
 class TestProfile(TestCase):
@@ -215,3 +217,21 @@ def test_profile_replace_dict_input(self):
                 }
             },
         )
+
+    def test_verify_replaced(self):
+        verify_profile_variables_replaced({"Foo": {"Bar": "${AllGood}something What"}})
+        verify_profile_variables_replaced({"Foo": {"Bar": "${All.Good}something %{What"}})
+        verify_profile_variables_replaced({"Foo": {"${AllGood}": "something %{What"}})
+
+    @parameterized.expand(
+        [
+            ({"Foo": {"Bar": "%{NotGood}something What"}}, "%{NotGood}"),
+            ({"Foo": {"Bar": "%{Not.Good}something What"}}, "%{Not.Good}"),
+            ({"Foo": {"%{NotGood}": "something What"}}, "%{NotGood}"),
+            ({"Foo": {"%{NotGood}": "something %{What.No}"}}, "['%{NotGood}', '%{What.No}']"),
+        ]
+    )
+    def test_verify_not_replaced(self, profile, error_includes):
+        with self.assertRaises(ValueError) as ctx:
+            verify_profile_variables_replaced(profile)
+        self.assertIn(error_includes, str(ctx.exception))