(self, request: Request, results: TestResult)
| 33 | self.args: tuple[Any, ...] = args |
| 34 | |
| 35 | def add_pre_hook(self, request: Request, results: TestResult) -> Request: |
| 36 | if hasattr(self, "pre_process"): |
| 37 | cb = request.callback |
| 38 | assert cb is not None |
| 39 | |
| 40 | @wraps(cb) |
| 41 | def wrapper(response: Response, **cb_kwargs: Any) -> list[Any]: |
| 42 | try: |
| 43 | results.startTest(self.testcase_pre) |
| 44 | self.pre_process(response) |
| 45 | results.stopTest(self.testcase_pre) |
| 46 | except AssertionError: |
| 47 | results.addFailure(self.testcase_pre, sys.exc_info()) |
| 48 | except Exception: |
| 49 | results.addError(self.testcase_pre, sys.exc_info()) |
| 50 | else: |
| 51 | results.addSuccess(self.testcase_pre) |
| 52 | cb_result = cb(response, **cb_kwargs) |
| 53 | if isinstance(cb_result, (AsyncGenerator, CoroutineType)): |
| 54 | raise TypeError("Contracts don't support async callbacks") |
| 55 | return list(cast("Iterable[Any]", iterate_spider_output(cb_result))) |
| 56 | |
| 57 | request.callback = wrapper |
| 58 | |
| 59 | return request |
| 60 | |
| 61 | def add_post_hook(self, request: Request, results: TestResult) -> Request: |
| 62 | if hasattr(self, "post_process"): |
no outgoing calls