fix: Raise correct exception when path/method items from get_conditional_contents are not dict (#2683)

Author: aahung

samtranslator/open_api/base_editor.py

@@ -48,18 +48,20 @@ def method_definition_has_integration(method_definition: Dict[str, Any]) -> bool
         :param dict method_definition: method definition dictionary
         :return: True if an integration exists
         """
-
         return bool(method_definition.get(BaseEditor._X_APIGW_INTEGRATION))
 
-    def method_has_integration(self, method: Dict[str, Any]) -> bool:
+    def method_has_integration(self, raw_method_definition: Dict[str, Any], path: str, method: str) -> bool:
         """
         Returns true if the given method contains a valid method definition.
         This uses the get_conditional_contents function to handle conditionals.
 
-        :param dict method: method dictionary
+        :param dict raw_method_definition: raw method dictionary
+        :param str path: path name
+        :param str method: method name
         :return: true if method has one or multiple integrations
         """
-        for method_definition in self.get_conditional_contents(method):
+        for method_definition in self.get_conditional_contents(raw_method_definition):
+            self.validate_method_definition_is_dict(method_definition, path, method)
             if self.method_definition_has_integration(method_definition):
                 return True
         return False
@@ -117,7 +119,7 @@ def has_path(self, path: str, method: Optional[str] = None) -> bool:
         method = self._normalize_method_name(method)
         if method:
             for path_item in self.get_conditional_contents(self.paths.get(path)):
-                if not path_item or method not in path_item:
+                if not isinstance(path_item, dict) or method not in path_item:
                     return False
         return True
 
@@ -136,8 +138,11 @@ def has_integration(self, path: str, method: str) -> bool:
             return False
 
         for path_item in self.get_conditional_contents(self.paths.get(path)):
+            BaseEditor.validate_path_item_is_dict(path_item, path)
             method_definition = path_item.get(method)
-            if not (isinstance(method_definition, dict) and self.method_has_integration(method_definition)):
+            if not (
+                isinstance(method_definition, dict) and self.method_has_integration(method_definition, path, method)
+            ):
                 return False
         # Integration present and non-empty
         return True
@@ -192,7 +197,9 @@ def iter_on_method_definitions_for_path_at_method(
         normalized_method_name = self._normalize_method_name(method_name)
 
         for path_item in self.get_conditional_contents(self.paths.get(path_name)):
+            BaseEditor.validate_path_item_is_dict(path_item, path_name)
             for method_definition in self.get_conditional_contents(path_item.get(normalized_method_name)):
+                BaseEditor.validate_method_definition_is_dict(method_definition, path_name, method_name)
                 if skip_methods_without_apigw_integration and not self.method_definition_has_integration(
                     method_definition
                 ):
@@ -224,6 +231,12 @@ def validate_path_item_is_dict(path_item: Any, path: str) -> None:
             path_item, "Value of '{}' path must be a dictionary according to Swagger spec.".format(path)
         )
 
+    @staticmethod
+    def validate_method_definition_is_dict(method_definition: Optional[Any], path: str, method: str) -> None:
+        BaseEditor.validate_is_dict(
+            method_definition, f"Definition of method '{method}' for path '{path}' should be a map."
+        )
+
     @staticmethod
     def safe_compare_regex_with_string(regex: str, data: Any) -> bool:
         return re.match(regex, str(data)) is not None

samtranslator/open_api/open_api.py

@@ -127,6 +127,7 @@ def add_lambda_integration(  # type: ignore[no-untyped-def]
             integration_uri = make_conditional(condition, integration_uri)
 
         for path_item in self.get_conditional_contents(self.paths.get(path)):
+            BaseEditor.validate_path_item_is_dict(path_item, path)
             # create as Py27Dict and insert key one by one to preserve input order
             if path_item[method] is None:
                 path_item[method] = Py27Dict()
@@ -155,8 +156,10 @@ def iter_on_all_methods_for_path(self, path_name, skip_methods_without_apigw_int
         :yields list of (method name, method definition) tuples
         """
         for path_item in self.get_conditional_contents(self.paths.get(path_name)):
+            BaseEditor.validate_path_item_is_dict(path_item, path_name)
             for method_name, method in path_item.items():
                 for method_definition in self.get_conditional_contents(method):
+                    BaseEditor.validate_method_definition_is_dict(method_definition, path_name, method_name)
                     if skip_methods_without_apigw_integration and not self.method_definition_has_integration(
                         method_definition
                     ):
@@ -253,6 +256,7 @@ def set_path_default_authorizer(
         :param dict authorizers: Dict of Authorizer configurations defined on the related Api.
         """
         for path_item in self.get_conditional_contents(self.paths.get(path)):
+            BaseEditor.validate_path_item_is_dict(path_item, path)
             for method_name, method in path_item.items():
                 normalized_method_name = self._normalize_method_name(method_name)
                 # Excluding parameters section
@@ -270,16 +274,8 @@ def set_path_default_authorizer(
                             ]
                         )
                     for method_definition in self.get_conditional_contents(method):
-                        # check if there is any method_definition given by customer
-                        if not method_definition:
-                            raise InvalidDocumentException(
-                                [
-                                    InvalidTemplateException(
-                                        f"Invalid method definition ({normalized_method_name}) for path: {path}"
-                                    )
-                                ]
-                            )
                         # If no integration given, then we don't need to process this definition (could be AWS::NoValue)
+                        BaseEditor.validate_method_definition_is_dict(method_definition, path, method_name)
                         if not self.method_definition_has_integration(method_definition):
                             continue
                         existing_security = method_definition.get("security")

samtranslator/swagger/swagger.py

@@ -121,6 +121,7 @@ def add_lambda_integration(  # type: ignore[no-untyped-def]
             integration_uri = make_conditional(condition, integration_uri)
 
         for path_item in self.get_conditional_contents(self.paths.get(path)):
+            BaseEditor.validate_path_item_is_dict(path_item, path)
             path_item[method][self._X_APIGW_INTEGRATION] = Py27Dict()
             # insert key one by one to preserce input order
             path_item[method][self._X_APIGW_INTEGRATION]["type"] = "aws_proxy"
@@ -187,6 +188,7 @@ def add_state_machine_integration(  # type: ignore[no-untyped-def]
             integration_uri = make_conditional(condition, integration_uri)
 
         for path_item in self.get_conditional_contents(self.paths.get(path)):
+            BaseEditor.validate_path_item_is_dict(path_item, path)
             # Responses
             integration_responses = Py27Dict()
             # insert key one by one to preserce input order
@@ -233,18 +235,14 @@ def iter_on_all_methods_for_path(self, path_name, skip_methods_without_apigw_int
         :yields list of (method name, method definition) tuples
         """
         for path_item in self.get_conditional_contents(self.paths.get(path_name)):
+            BaseEditor.validate_path_item_is_dict(path_item, path_name)
             for method_name, method in path_item.items():
                 # Excluding non-method sections
                 if method_name in SwaggerEditor._EXCLUDED_PATHS_FIELDS:
                     continue
 
                 for method_definition in self.get_conditional_contents(method):
-                    SwaggerEditor.validate_is_dict(
-                        method_definition,
-                        'Value of "{}" ({}) for path {} is not a valid dictionary.'.format(
-                            method_name, method_definition, path_name
-                        ),
-                    )
+                    BaseEditor.validate_method_definition_is_dict(method_definition, path_name, method_name)
                     if skip_methods_without_apigw_integration and not self.method_definition_has_integration(
                         method_definition
                     ):
@@ -280,6 +278,7 @@ def add_cors(  # type: ignore[no-untyped-def]
         """
 
         for path_item in self.get_conditional_contents(self.paths.get(path)):
+            BaseEditor.validate_path_item_is_dict(path_item, path)
             # Skip if Options is already present
             method = self._normalize_method_name(self._OPTIONS_METHOD)
             if method in path_item:
@@ -290,7 +289,7 @@ def add_cors(  # type: ignore[no-untyped-def]
 
             if not allowed_methods:
                 # AllowMethods is not given. Let's try to generate the list from the given Swagger.
-                allowed_methods = self._make_cors_allowed_methods_for_path_item(path_item)  # type: ignore[no-untyped-call]
+                allowed_methods = self._make_cors_allowed_methods_for_path_item(path_item)
 
                 # APIGW expects the value to be a "string expression". Hence wrap in another quote. Ex: "'GET,POST,DELETE'"
                 allowed_methods = "'{}'".format(allowed_methods)
@@ -412,7 +411,7 @@ def _options_method_response_for_cors(  # type: ignore[no-untyped-def]
         to_return["responses"]["200"]["headers"] = response_headers
         return to_return
 
-    def _make_cors_allowed_methods_for_path_item(self, path_item):  # type: ignore[no-untyped-def]
+    def _make_cors_allowed_methods_for_path_item(self, path_item: Dict[str, Any]) -> str:
         """
         Creates the value for Access-Control-Allow-Methods header for given path item. All HTTP methods defined for this
         path item will be included in the result. If the path item contains "ANY" method, then *all available* HTTP methods will
@@ -993,6 +992,7 @@ def _get_method_path_uri_list(self, path, stage):  # type: ignore[no-untyped-def
         """
         methods = []
         for path_item in self.get_conditional_contents(self.paths.get(path)):
+            BaseEditor.validate_path_item_is_dict(path_item, path)
             methods += list(path_item.keys())
 
         uri_list = []

tests/translator/output/error_http_api_null_method.json

@@ -1,3 +1,3 @@
 {
-  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Invalid method definition (post) for path: /test"
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Definition of method 'post' for path '/test' should be a map."
 }

tests/translator/output/error_invalid_method_definition.json

@@ -1,3 +1,3 @@
 {
-  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Value of \"tags\" (['InvalidMethodDefinition']) for path / is not a valid dictionary."
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Definition of method 'tags' for path '/' should be a map."
 }

tests/translator/output/error_null_method_definition.json

@@ -1,3 +1,3 @@
 {
-  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Value of \"get\" (None) for path / is not a valid dictionary."
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Definition of method 'get' for path '/' should be a map."
 }