diff --git a/samtranslator/model/api/api_generator.py b/samtranslator/model/api/api_generator.py index 55ae276fb5..a05094a9aa 100644 --- a/samtranslator/model/api/api_generator.py +++ b/samtranslator/model/api/api_generator.py @@ -957,10 +957,17 @@ def _add_gateway_responses(self): # type: ignore[no-untyped-def] # The dicts below will eventually become part of swagger/openapi definition, thus requires using Py27Dict() gateway_responses = Py27Dict() for response_type, response in self.gateway_responses.items(): + sam_expect(response, self.logical_id, f"GatewayResponses.{response_type}").to_be_a_map() + response_parameters = response.get("ResponseParameters", Py27Dict()) + response_templates = response.get("ResponseTemplates", Py27Dict()) + if response_parameters: + sam_expect( + response_parameters, self.logical_id, f"GatewayResponses.{response_type}.ResponseParameters" + ).to_be_a_map() gateway_responses[response_type] = ApiGatewayResponse( api_logical_id=self.logical_id, - response_parameters=response.get("ResponseParameters", Py27Dict()), - response_templates=response.get("ResponseTemplates", Py27Dict()), + response_parameters=response_parameters, + response_templates=response_templates, status_code=response.get("StatusCode", None), ) @@ -1028,6 +1035,12 @@ def _openapi_postprocess(self, definition_body): # type: ignore[no-untyped-def] del definition_body["securityDefinitions"] if definition_body.get("definitions"): components = definition_body.get("components", Py27Dict()) + # the following line to check if components is None + # is copied from the previous if... + # In the previous line, the default value `Py27Dict()` will be only returned only if `components` + # property is not in definition_body dict, but if it exist, and its value is None, so None will be + # returned and not the default value. That is why the below line is required. + components = components if components else Py27Dict() components["schemas"] = definition_body["definitions"] definition_body["components"] = components del definition_body["definitions"] @@ -1035,9 +1048,18 @@ def _openapi_postprocess(self, definition_body): # type: ignore[no-untyped-def] # adds `schema` for the headers in responses for openapi3 paths = definition_body.get("paths") if paths: + SwaggerEditor.validate_is_dict( + paths, + "Value of paths must be a dictionary according to Swagger spec.", + ) for path, path_item in paths.items(): - SwaggerEditor.validate_path_item_is_dict(path_item, path) # type: ignore[no-untyped-call] + SwaggerEditor.validate_path_item_is_dict(path_item, path) if path_item.get("options"): + SwaggerEditor.validate_is_dict( + path_item.get("options"), + "Value of options method for path {} must be a " + "dictionary according to Swagger spec.".format(path), + ) options = path_item.get("options").copy() for field, field_val in options.items(): # remove unsupported produces and consumes in options for openapi3 @@ -1050,19 +1072,28 @@ def _openapi_postprocess(self, definition_body): # type: ignore[no-untyped-def] "Value of responses in options method for path {} must be a " "dictionary according to Swagger spec.".format(path), ) - if field_val.get("200") and field_val.get("200").get("headers"): - headers = field_val["200"]["headers"] - SwaggerEditor.validate_is_dict( - headers, - "Value of response's headers in options method for path {} must be a " - "dictionary according to Swagger spec.".format(path), - ) - for header, header_val in headers.items(): - new_header_val_with_schema = Py27Dict() - new_header_val_with_schema["schema"] = header_val - definition_body["paths"][path]["options"][field]["200"]["headers"][ - header - ] = new_header_val_with_schema + response_200 = field_val.get("200") + if not response_200: + continue + SwaggerEditor.validate_is_dict( + response_200, + "Value of responses.200 in options method for path {} must be a " + "dictionary according to Swagger spec.".format(path), + ) + response_200_headers = response_200.get("headers") + if not response_200_headers: + continue + SwaggerEditor.validate_is_dict( + response_200_headers, + "Value of response's headers in options method for path {} must be a " + "dictionary according to Swagger spec.".format(path), + ) + for header, header_val in response_200_headers.items(): + new_header_val_with_schema = Py27Dict() + new_header_val_with_schema["schema"] = header_val + definition_body["paths"][path]["options"][field]["200"]["headers"][ + header + ] = new_header_val_with_schema return definition_body diff --git a/samtranslator/model/api/http_api_generator.py b/samtranslator/model/api/http_api_generator.py index 87fe2fb2ad..c87925d507 100644 --- a/samtranslator/model/api/http_api_generator.py +++ b/samtranslator/model/api/http_api_generator.py @@ -656,7 +656,13 @@ def _add_description(self) -> None: self.logical_id, "Description works only with inline OpenApi specified in the 'DefinitionBody' property.", ) - if self.definition_body.get("info", {}).get("description"): + info = self.definition_body.get("info", {}) + if not isinstance(info, dict): + raise InvalidResourceException( + self.logical_id, + "'info' in OpenApi definition body must be a map.", + ) + if info.get("description"): raise InvalidResourceException( self.logical_id, "Unable to set Description because it is already defined within inline OpenAPI specified in the " @@ -677,7 +683,14 @@ def _add_title(self) -> None: "Name works only with inline OpenApi specified in the 'DefinitionBody' property.", ) - if self.definition_body.get("info", {}).get("title") != OpenApiEditor._DEFAULT_OPENAPI_TITLE: + info = self.definition_body.get("info", {}) + if not isinstance(info, dict): + raise InvalidResourceException( + self.logical_id, + "'info' in OpenApi definition body must be a map.", + ) + + if info.get("title") != OpenApiEditor._DEFAULT_OPENAPI_TITLE: raise InvalidResourceException( self.logical_id, "Unable to set Name because it is already defined within inline OpenAPI specified in the " diff --git a/samtranslator/model/apigateway.py b/samtranslator/model/apigateway.py index bdd8249fbe..305a935f94 100644 --- a/samtranslator/model/apigateway.py +++ b/samtranslator/model/apigateway.py @@ -6,6 +6,7 @@ from samtranslator.model.exceptions import InvalidResourceException from samtranslator.model.types import is_type, one_of, is_str, list_of from samtranslator.model.intrinsics import ref, fnSub +from samtranslator.schema.common import PassThrough from samtranslator.translator import logical_id_generator from samtranslator.translator.arn_generator import ArnGenerator from samtranslator.utils.py27hash_fix import Py27Dict, Py27UniStr @@ -123,10 +124,11 @@ def __init__( self, api_logical_id: str, response_parameters: Optional[Dict[str, Any]] = None, - response_templates: Optional[Dict[str, Any]] = None, + response_templates: Optional[PassThrough] = None, status_code: Optional[str] = None, ) -> None: if response_parameters: + # response_parameters has been validated in ApiGenerator._add_gateway_responses() for response_parameter_key in response_parameters.keys(): if response_parameter_key not in ApiGatewayResponse.ResponseParameterProperties: raise InvalidResourceException( diff --git a/samtranslator/swagger/swagger.py b/samtranslator/swagger/swagger.py index 38812ac3b6..869360a861 100644 --- a/samtranslator/swagger/swagger.py +++ b/samtranslator/swagger/swagger.py @@ -67,7 +67,7 @@ def __init__(self, doc): # type: ignore[no-untyped-def] # so we don't need to validate wherever we use them. for path in self.iter_on_path(): for path_item in self.get_conditional_contents(self.paths.get(path)): # type: ignore[no-untyped-call] - SwaggerEditor.validate_path_item_is_dict(path_item, path) # type: ignore[no-untyped-call] + SwaggerEditor.validate_path_item_is_dict(path_item, path) def get_conditional_contents(self, item): # type: ignore[no-untyped-def] """ @@ -1368,7 +1368,7 @@ def validate_is_dict(obj: Any, exception_message: str) -> None: raise InvalidDocumentException([InvalidTemplateException(exception_message)]) @staticmethod - def validate_path_item_is_dict(path_item, path): # type: ignore[no-untyped-def] + def validate_path_item_is_dict(path_item: Any, path: str) -> None: """ Throws exception if path_item is not a dict