Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/connexion/middleware/response_validation.py: 32%

1""" 

2Validation Middleware. 

3""" 

4 

5import logging 

6import typing as t 

7 

8from starlette.types import ASGIApp, Receive, Scope, Send 

9 

10from connexion import utils 

11from connexion.datastructures import MediaTypeDict 

12from connexion.exceptions import NonConformingResponseHeaders 

13from connexion.middleware.abstract import RoutedAPI, RoutedMiddleware 

14from connexion.operations import AbstractOperation 

15from connexion.validators import VALIDATOR_MAP 

16 

17logger = logging.getLogger("connexion.middleware.validation") 

18 

19 

20class ResponseValidationOperation: 

21 def __init__( 

22 self, 

23 next_app: ASGIApp, 

24 *, 

25 operation: AbstractOperation, 

26 validator_map: t.Optional[dict] = None, 

27 ) -> None: 

28 self.next_app = next_app 

29 self._operation = operation 

30 self._validator_map = VALIDATOR_MAP.copy() 

31 self._validator_map.update(validator_map or {}) 

32 

33 def extract_content_type( 

34 self, headers: t.List[t.Tuple[bytes, bytes]] 

35 ) -> t.Tuple[str, str]: 

36 """Extract the mime type and encoding from the content type headers. 

37 

38 :param headers: Headers from ASGI scope 

39 

40 :return: A tuple of mime type, encoding 

41 """ 

42 content_type = utils.extract_content_type(headers) 

43 mime_type, encoding = utils.split_content_type(content_type) 

44 if mime_type is None: 

45 # Content-type header is not required. Take a best guess. 

46 try: 

47 mime_type = self._operation.produces[0] 

48 except IndexError: 

49 mime_type = "application/octet-stream" 

50 if encoding is None: 

51 encoding = "utf-8" 

52 

53 return mime_type, encoding 

54 

55 def validate_mime_type(self, mime_type: str) -> None: 

56 """Validate the mime type against the spec if it defines which mime types are produced. 

57 

58 :param mime_type: mime type from content type header 

59 """ 

60 if not self._operation.produces: 

61 return 

62 

63 media_type_dict = MediaTypeDict( 

64 [(p.lower(), None) for p in self._operation.produces] 

65 ) 

66 if mime_type.lower() not in media_type_dict: 

67 raise NonConformingResponseHeaders( 

68 detail=f"Invalid Response Content-type ({mime_type}), " 

69 f"expected {self._operation.produces}", 

70 ) 

71 

72 @staticmethod 

73 def validate_required_headers( 

74 headers: t.List[tuple], response_definition: dict 

75 ) -> None: 

76 required_header_keys = { 

77 k.lower() 

78 for (k, v) in response_definition.get("headers", {}).items() 

79 if v.get("required", False) 

80 } 

81 header_keys = set(header[0].decode("latin-1").lower() for header in headers) 

82 missing_keys = required_header_keys - header_keys 

83 if missing_keys: 

84 pretty_list = ", ".join(missing_keys) 

85 msg = ( 

86 "Keys in response header don't match response specification. Difference: {}" 

87 ).format(pretty_list) 

88 raise NonConformingResponseHeaders(detail=msg) 

89 

90 async def __call__(self, scope: Scope, receive: Receive, send: Send): 

91 async def wrapped_send(message: t.MutableMapping[str, t.Any]) -> None: 

92 nonlocal send 

93 

94 if message["type"] == "http.response.start": 

95 headers = message["headers"] 

96 

97 mime_type, encoding = self.extract_content_type(headers) 

98 if message["status"] < 400: 

99 self.validate_mime_type(mime_type) 

100 

101 status = str(message["status"]) 

102 response_definition = self._operation.response_definition( 

103 status, mime_type 

104 ) 

105 self.validate_required_headers(headers, response_definition) 

106 

107 # Validate body 

108 try: 

109 body_validator = self._validator_map["response"][mime_type] # type: ignore 

110 except KeyError: 

111 logger.info( 

112 f"Skipping validation. No validator registered for content type: " 

113 f"{mime_type}." 

114 ) 

115 else: 

116 validator = body_validator( 

117 scope, 

118 schema=self._operation.response_schema(status, mime_type), 

119 nullable=utils.is_nullable( 

120 self._operation.response_definition(status, mime_type) 

121 ), 

122 encoding=encoding, 

123 ) 

124 send = validator.wrap_send(send) 

125 

126 return await send(message) 

127 

128 await self.next_app(scope, receive, wrapped_send) 

129 

130 

131class ResponseValidationAPI(RoutedAPI[ResponseValidationOperation]): 

132 """Validation API.""" 

133 

134 def __init__( 

135 self, 

136 *args, 

137 validator_map=None, 

138 validate_responses=False, 

139 **kwargs, 

140 ): 

141 super().__init__(*args, **kwargs) 

142 self.validator_map = validator_map 

143 self.validate_responses = validate_responses 

144 self.add_paths() 

145 

146 def make_operation( 

147 self, operation: AbstractOperation 

148 ) -> ResponseValidationOperation: 

149 if self.validate_responses: 

150 return ResponseValidationOperation( 

151 self.next_app, 

152 operation=operation, 

153 validator_map=self.validator_map, 

154 ) 

155 else: 

156 return self.next_app # type: ignore 

157 

158 

159class ResponseValidationMiddleware(RoutedMiddleware[ResponseValidationAPI]): 

160 """Middleware for validating requests according to the API contract.""" 

161 

162 api_cls = ResponseValidationAPI