| 255 | |
| 256 | |
| 257 | class ResponseObject(ResponseObjectBase): |
| 258 | def __getattr__(self, key): |
| 259 | if key in ["json", "content", "body"]: |
| 260 | try: |
| 261 | value = self.resp_obj.json() |
| 262 | except ValueError: |
| 263 | value = self.resp_obj.content |
| 264 | elif key == "cookies": |
| 265 | value = self.resp_obj.cookies.get_dict() |
| 266 | else: |
| 267 | try: |
| 268 | value = getattr(self.resp_obj, key) |
| 269 | except AttributeError: |
| 270 | err_msg = "ResponseObject does not have attribute: {}".format(key) |
| 271 | logger.error(err_msg) |
| 272 | raise exceptions.ParamsError(err_msg) |
| 273 | |
| 274 | self.__dict__[key] = value |
| 275 | return value |
| 276 | |
| 277 | def _search_jmespath(self, expr: Text) -> Any: |
| 278 | resp_obj_meta = { |
| 279 | "status_code": self.status_code, |
| 280 | "headers": self.headers, |
| 281 | "cookies": self.cookies, |
| 282 | "body": self.body, |
| 283 | } |
| 284 | if not expr.startswith(tuple(resp_obj_meta.keys())): |
| 285 | if hasattr(self.resp_obj,expr): |
| 286 | return getattr(self.resp_obj,expr) |
| 287 | else: |
| 288 | return expr |
| 289 | |
| 290 | try: |
| 291 | check_value = jmespath.search(expr, resp_obj_meta) |
| 292 | except JMESPathError as ex: |
| 293 | logger.error( |
| 294 | f"failed to search with jmespath\n" |
| 295 | f"expression: {expr}\n" |
| 296 | f"data: {resp_obj_meta}\n" |
| 297 | f"exception: {ex}" |
| 298 | ) |
| 299 | raise |
| 300 | |
| 301 | return check_value |
| 302 | |
| 303 | |
| 304 | class ThriftResponseObject(ResponseObjectBase): |
no outgoing calls