(self, request: Request, results: TestResult)
| 59 | return request |
| 60 | |
| 61 | def add_post_hook(self, request: Request, results: TestResult) -> Request: |
| 62 | if hasattr(self, "post_process"): |
| 63 | cb = request.callback |
| 64 | assert cb is not None |
| 65 | |
| 66 | @wraps(cb) |
| 67 | def wrapper(response: Response, **cb_kwargs: Any) -> list[Any]: |
| 68 | cb_result = cb(response, **cb_kwargs) |
| 69 | if isinstance(cb_result, (AsyncGenerator, CoroutineType)): |
| 70 | raise TypeError("Contracts don't support async callbacks") |
| 71 | output = list(cast("Iterable[Any]", iterate_spider_output(cb_result))) |
| 72 | try: |
| 73 | results.startTest(self.testcase_post) |
| 74 | self.post_process(output) |
| 75 | results.stopTest(self.testcase_post) |
| 76 | except AssertionError: |
| 77 | results.addFailure(self.testcase_post, sys.exc_info()) |
| 78 | except Exception: |
| 79 | results.addError(self.testcase_post, sys.exc_info()) |
| 80 | else: |
| 81 | results.addSuccess(self.testcase_post) |
| 82 | return output |
| 83 | |
| 84 | request.callback = wrapper |
| 85 | |
| 86 | return request |
| 87 | |
| 88 | def adjust_request_args(self, args: dict[str, Any]) -> dict[str, Any]: |
| 89 | return args |
no outgoing calls