Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/connexion/validators/abstract.py: 29%

1""" 

2This module defines a Validator interface with base functionality that can be subclassed 

3for custom validators provided to the RequestValidationMiddleware. 

4""" 

5 

6import copy 

7import json 

8import typing as t 

9 

10from starlette.datastructures import Headers, MutableHeaders 

11from starlette.types import Receive, Scope, Send 

12 

13from connexion.exceptions import BadRequestProblem 

14 

15 

16class AbstractRequestBodyValidator: 

17 """ 

18 Validator interface with base functionality that can be subclassed for custom validators. 

19 

20 .. note: Validators load the whole body into memory, which can be a problem for large payloads. 

21 """ 

22 

23 MUTABLE_VALIDATION = False 

24 """ 

25 Whether mutations to the body during validation should be transmitted via the receive channel. 

26 Note that this does not apply to the substitution of a missing body with the default body, which always 

27 updates the receive channel. 

28 """ 

29 MAX_MESSAGE_LENGTH = 256000 

30 """Maximum message length that will be sent via the receive channel for mutated bodies.""" 

31 

32 def __init__( 

33 self, 

34 *, 

35 schema: dict, 

36 required: bool = False, 

37 nullable: bool = False, 

38 encoding: str, 

39 strict_validation: bool, 

40 **kwargs, 

41 ): 

42 """ 

43 :param schema: Schema of operation to validate 

44 :param required: Whether RequestBody is required 

45 :param nullable: Whether RequestBody is nullable 

46 :param encoding: Encoding of body (passed via Content-Type header) 

47 :param kwargs: Additional arguments for subclasses 

48 :param strict_validation: Whether to allow parameters not defined in the spec 

49 """ 

50 self._schema = schema 

51 self._nullable = nullable 

52 self._required = required 

53 self._encoding = encoding 

54 self._strict_validation = strict_validation 

55 

56 async def _parse( 

57 self, stream: t.AsyncGenerator[bytes, None], scope: Scope 

58 ) -> t.Any: 

59 """Parse the incoming stream.""" 

60 

61 def _validate(self, body: t.Any) -> t.Optional[dict]: 

62 """ 

63 Validate the parsed body. 

64 

65 :raises: :class:`connexion.exceptions.BadRequestProblem` 

66 """ 

67 

68 def _insert_body( 

69 self, receive: Receive, *, body: t.Any, scope: Scope 

70 ) -> t.Tuple[Receive, Scope]: 

71 """ 

72 Insert messages transmitting the body at the start of the `receive` channel. 

73 

74 This method updates the provided `scope` in place with the right `Content-Length` header. 

75 """ 

76 if body is None: 

77 return receive, scope 

78 

79 bytes_body = json.dumps(body).encode(self._encoding) 

80 

81 # Update the content-length header 

82 new_scope = scope.copy() 

83 new_scope["headers"] = copy.deepcopy(scope["headers"]) 

84 headers = MutableHeaders(scope=new_scope) 

85 headers["content-length"] = str(len(bytes_body)) 

86 

87 # Wrap in new receive channel 

88 messages = ( 

89 { 

90 "type": "http.request", 

91 "body": bytes_body[i : i + self.MAX_MESSAGE_LENGTH], 

92 "more_body": i + self.MAX_MESSAGE_LENGTH < len(bytes_body), 

93 } 

94 for i in range(0, len(bytes_body), self.MAX_MESSAGE_LENGTH) 

95 ) 

96 

97 receive = self._insert_messages(receive, messages=messages) 

98 

99 return receive, new_scope 

100 

101 @staticmethod 

102 def _insert_messages( 

103 receive: Receive, *, messages: t.Iterable[t.MutableMapping[str, t.Any]] 

104 ) -> Receive: 

105 """Insert messages at the start of the `receive` channel.""" 

106 # Ensure that messages is an iterator so each message is replayed once. 

107 message_iterator = iter(messages) 

108 

109 async def receive_() -> t.MutableMapping[str, t.Any]: 

110 try: 

111 return next(message_iterator) 

112 except StopIteration: 

113 return await receive() 

114 

115 return receive_ 

116 

117 async def wrap_receive( 

118 self, receive: Receive, *, scope: Scope 

119 ) -> t.Tuple[Receive, Scope]: 

120 """ 

121 Wrap the provided `receive` channel with request body validation. 

122 

123 This method updates the provided `scope` in place with the right `Content-Length` header. 

124 """ 

125 # Handle missing bodies 

126 headers = Headers(scope=scope) 

127 if not int(headers.get("content-length", 0)): 

128 body = self._schema.get("default") 

129 if body is None and self._required: 

130 raise BadRequestProblem("RequestBody is required") 

131 # The default body is encoded as a `receive` channel to mimic an incoming body 

132 receive, scope = self._insert_body(receive, body=body, scope=scope) 

133 

134 # The receive channel is converted to a stream for convenient access 

135 messages = [] 

136 

137 async def stream() -> t.AsyncGenerator[bytes, None]: 

138 more_body = True 

139 while more_body: 

140 message = await receive() 

141 messages.append(message) 

142 more_body = message.get("more_body", False) 

143 yield message.get("body", b"") 

144 yield b"" 

145 

146 # The body is parsed and validated 

147 body = await self._parse(stream(), scope=scope) 

148 if not (body is None and self._nullable): 

149 self._validate(body) 

150 

151 # If MUTABLE_VALIDATION is enabled, include any changes made during validation in the messages to send 

152 if self.MUTABLE_VALIDATION: 

153 # Include changes made during validation 

154 receive, scope = self._insert_body(receive, body=body, scope=scope) 

155 else: 

156 # Serialize original messages 

157 receive = self._insert_messages(receive, messages=messages) 

158 

159 return receive, scope 

160 

161 

162class AbstractResponseBodyValidator: 

163 """ 

164 Validator interface with base functionality that can be subclassed for custom validators. 

165 

166 .. note: Validators load the whole body into memory, which can be a problem for large payloads. 

167 """ 

168 

169 def __init__( 

170 self, 

171 scope: Scope, 

172 *, 

173 schema: dict, 

174 nullable: bool = False, 

175 encoding: str, 

176 ) -> None: 

177 self._scope = scope 

178 self._schema = schema 

179 self._nullable = nullable 

180 self._encoding = encoding 

181 

182 def _parse(self, stream: t.Generator[bytes, None, None]) -> t.Any: 

183 """Parse the incoming stream.""" 

184 

185 def _validate(self, body: t.Any) -> t.Optional[dict]: 

186 """ 

187 Validate the body. 

188 

189 :raises: :class:`connexion.exceptions.NonConformingResponse` 

190 """ 

191 

192 def wrap_send(self, send: Send) -> Send: 

193 """Wrap the provided send channel with response body validation""" 

194 

195 messages = [] 

196 

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

198 messages.append(message) 

199 

200 if message["type"] == "http.response.start" or message.get( 

201 "more_body", False 

202 ): 

203 return 

204 

205 stream = (message.get("body", b"") for message in messages) 

206 body = self._parse(stream) 

207 

208 if not (body is None and self._nullable): 

209 self._validate(body) 

210 

211 while messages: 

212 await send(messages.pop(0)) 

213 

214 return send_