(self, method: str, path: str, **kwargs)
| 215 | ) |
| 216 | |
| 217 | def _execute(self, method: str, path: str, **kwargs) -> TestResponse: |
| 218 | request = self._build_request(method, path, **kwargs) |
| 219 | |
| 220 | # ---- before middlewares (global) ---------------------------------- |
| 221 | for mw_fn in self._global_before: |
| 222 | result = self._call(mw_fn, request) |
| 223 | if result is not None: |
| 224 | if isinstance(result, Response): |
| 225 | return self._to_test_response(result) |
| 226 | request = result |
| 227 | |
| 228 | # ---- before middleware (route-specific) --------------------------- |
| 229 | mw_table = self._mw_before.get(method) |
| 230 | if mw_table is not None: |
| 231 | mw_fn, mw_params = mw_table.match(path) |
| 232 | if mw_fn is not None: |
| 233 | request.path_params = mw_params |
| 234 | result = self._call(mw_fn, request) |
| 235 | if result is not None: |
| 236 | if isinstance(result, Response): |
| 237 | return self._to_test_response(result) |
| 238 | request = result |
| 239 | |
| 240 | # ---- route match -------------------------------------------------- |
| 241 | route_table = self._http_routes.get(method) |
| 242 | if route_table is None: |
| 243 | return TestResponse(status_code=404, headers=Headers({}), _body=b"Not Found") |
| 244 | |
| 245 | fn_info, path_params = route_table.match(path) |
| 246 | if fn_info is None: |
| 247 | return TestResponse(status_code=404, headers=Headers({}), _body=b"Not Found") |
| 248 | |
| 249 | request.path_params = path_params |
| 250 | |
| 251 | # ---- execute handler ---------------------------------------------- |
| 252 | response = self._call(fn_info, request) |
| 253 | |
| 254 | if not isinstance(response, Response): |
| 255 | if isinstance(response, (dict, list)): |
| 256 | body = json.dumps(response).encode("utf-8") |
| 257 | elif isinstance(response, bytes): |
| 258 | body = response |
| 259 | else: |
| 260 | body = str(response).encode("utf-8") |
| 261 | return TestResponse( |
| 262 | status_code=200, |
| 263 | headers=Headers({}), |
| 264 | _body=body, |
| 265 | ) |
| 266 | |
| 267 | # ---- merge global response headers -------------------------------- |
| 268 | excluded = self.app.excluded_response_headers_paths or [] |
| 269 | if path not in excluded and not self.app.response_headers.is_empty(): |
| 270 | if isinstance(response.headers, dict): |
| 271 | response.headers = Headers(response.headers) |
| 272 | resp_headers = response.headers |
| 273 | global_dict = self.app.response_headers.get_headers() |
| 274 | for key, values in global_dict.items(): |
no test coverage detected