From b99526b740890e63f1d05074b3e358a9ae59b77f Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Sun, 19 May 2024 15:45:51 +0500 Subject: [PATCH] Full typing for scrapy/contracts. --- scrapy/contracts/__init__.py | 75 ++++++++++++++++++++++++------------ scrapy/contracts/default.py | 19 ++++----- 2 files changed, 60 insertions(+), 34 deletions(-) diff --git a/scrapy/contracts/__init__.py b/scrapy/contracts/__init__.py index d46eb7c51..b300b8457 100644 --- a/scrapy/contracts/__init__.py +++ b/scrapy/contracts/__init__.py @@ -3,10 +3,23 @@ import sys from functools import wraps from inspect import getmembers from types import CoroutineType -from typing import AsyncGenerator, Dict, Optional, Type -from unittest import TestCase +from typing import ( + Any, + AsyncGenerator, + Callable, + Dict, + Iterable, + List, + Optional, + Tuple, + Type, +) +from unittest import TestCase, TestResult -from scrapy.http import Request +from twisted.python.failure import Failure + +from scrapy import Spider +from scrapy.http import Request, Response from scrapy.utils.python import get_spec from scrapy.utils.spider import iterate_spider_output @@ -15,18 +28,20 @@ class Contract: """Abstract class for contracts""" request_cls: Optional[Type[Request]] = None + name: str - def __init__(self, method, *args): + def __init__(self, method: Callable, *args: Any): self.testcase_pre = _create_testcase(method, f"@{self.name} pre-hook") self.testcase_post = _create_testcase(method, f"@{self.name} post-hook") - self.args = args + self.args: Tuple[Any, ...] = args - def add_pre_hook(self, request, results): + def add_pre_hook(self, request: Request, results: TestResult) -> Request: if hasattr(self, "pre_process"): cb = request.callback + assert cb is not None @wraps(cb) - def wrapper(response, **cb_kwargs): + def wrapper(response: Response, **cb_kwargs: Any) -> List[Any]: try: results.startTest(self.testcase_pre) self.pre_process(response) @@ -49,12 +64,13 @@ class Contract: return request - def add_post_hook(self, request, results): + def add_post_hook(self, request: Request, results: TestResult) -> Request: if hasattr(self, "post_process"): cb = request.callback + assert cb is not None @wraps(cb) - def wrapper(response, **cb_kwargs): + def wrapper(response: Response, **cb_kwargs: Any) -> List[Any]: cb_result = cb(response, **cb_kwargs) if isinstance(cb_result, (AsyncGenerator, CoroutineType)): raise TypeError("Contracts don't support async callbacks") @@ -76,18 +92,18 @@ class Contract: return request - def adjust_request_args(self, args): + def adjust_request_args(self, args: Dict[str, Any]) -> Dict[str, Any]: return args class ContractsManager: - contracts: Dict[str, Contract] = {} + contracts: Dict[str, Type[Contract]] = {} - def __init__(self, contracts): + def __init__(self, contracts: Iterable[Type[Contract]]): for contract in contracts: self.contracts[contract.name] = contract - def tested_methods_from_spidercls(self, spidercls): + def tested_methods_from_spidercls(self, spidercls: Type[Spider]) -> List[str]: is_method = re.compile(r"^\s*@", re.MULTILINE).search methods = [] for key, value in getmembers(spidercls): @@ -96,21 +112,26 @@ class ContractsManager: return methods - def extract_contracts(self, method): - contracts = [] + def extract_contracts(self, method: Callable) -> List[Contract]: + contracts: List[Contract] = [] + assert method.__doc__ is not None for line in method.__doc__.split("\n"): line = line.strip() if line.startswith("@"): - name, args = re.match(r"@(\w+)\s*(.*)", line).groups() + m = re.match(r"@(\w+)\s*(.*)", line) + assert m is not None + name, args = m.groups() args = re.split(r"\s+", args) contracts.append(self.contracts[name](method, *args)) return contracts - def from_spider(self, spider, results): - requests = [] + def from_spider( + self, spider: Spider, results: TestResult + ) -> List[Optional[Request]]: + requests: List[Optional[Request]] = [] for method in self.tested_methods_from_spidercls(type(spider)): bound_method = spider.__getattribute__(method) try: @@ -121,7 +142,7 @@ class ContractsManager: return requests - def from_method(self, method, results): + def from_method(self, method: Callable, results: TestResult) -> Optional[Request]: contracts = self.extract_contracts(method) if contracts: request_cls = Request @@ -154,14 +175,18 @@ class ContractsManager: self._clean_req(request, method, results) return request + return None - def _clean_req(self, request, method, results): + def _clean_req( + self, request: Request, method: Callable, results: TestResult + ) -> None: """stop the request from returning objects and records any errors""" cb = request.callback + assert cb is not None @wraps(cb) - def cb_wrapper(response, **cb_kwargs): + def cb_wrapper(response: Response, **cb_kwargs: Any) -> None: try: output = cb(response, **cb_kwargs) output = list(iterate_spider_output(output)) @@ -169,7 +194,7 @@ class ContractsManager: case = _create_testcase(method, "callback") results.addError(case, sys.exc_info()) - def eb_wrapper(failure): + def eb_wrapper(failure: Failure) -> None: case = _create_testcase(method, "errback") exc_info = failure.type, failure.value, failure.getTracebackObject() results.addError(case, exc_info) @@ -178,11 +203,11 @@ class ContractsManager: request.errback = eb_wrapper -def _create_testcase(method, desc): - spider = method.__self__.name +def _create_testcase(method: Callable, desc: str) -> TestCase: + spider = method.__self__.name # type: ignore[attr-defined] class ContractTestCase(TestCase): - def __str__(_self): + def __str__(_self) -> str: return f"[{spider}] {method.__name__} ({desc})" name = f"{spider}_{method.__name__}" diff --git a/scrapy/contracts/default.py b/scrapy/contracts/default.py index eac702cef..71ca4168a 100644 --- a/scrapy/contracts/default.py +++ b/scrapy/contracts/default.py @@ -1,4 +1,5 @@ import json +from typing import Any, Callable, Dict, List, Optional from itemadapter import ItemAdapter, is_item @@ -15,7 +16,7 @@ class UrlContract(Contract): name = "url" - def adjust_request_args(self, args): + def adjust_request_args(self, args: Dict[str, Any]) -> Dict[str, Any]: args["url"] = self.args[0] return args @@ -29,7 +30,7 @@ class CallbackKeywordArgumentsContract(Contract): name = "cb_kwargs" - def adjust_request_args(self, args): + def adjust_request_args(self, args: Dict[str, Any]) -> Dict[str, Any]: args["cb_kwargs"] = json.loads(" ".join(self.args)) return args @@ -48,14 +49,14 @@ class ReturnsContract(Contract): """ name = "returns" - object_type_verifiers = { + object_type_verifiers: Dict[Optional[str], Callable[[Any], bool]] = { "request": lambda x: isinstance(x, Request), "requests": lambda x: isinstance(x, Request), "item": is_item, "items": is_item, } - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any): super().__init__(*args, **kwargs) if len(self.args) not in [1, 2, 3]: @@ -66,16 +67,16 @@ class ReturnsContract(Contract): self.obj_type_verifier = self.object_type_verifiers[self.obj_name] try: - self.min_bound = int(self.args[1]) + self.min_bound: float = int(self.args[1]) except IndexError: self.min_bound = 1 try: - self.max_bound = int(self.args[2]) + self.max_bound: float = int(self.args[2]) except IndexError: self.max_bound = float("inf") - def post_process(self, output): + def post_process(self, output: List[Any]) -> None: occurrences = 0 for x in output: if self.obj_type_verifier(x): @@ -85,7 +86,7 @@ class ReturnsContract(Contract): if not assertion: if self.min_bound == self.max_bound: - expected = self.min_bound + expected = str(self.min_bound) else: expected = f"{self.min_bound}..{self.max_bound}" @@ -101,7 +102,7 @@ class ScrapesContract(Contract): name = "scrapes" - def post_process(self, output): + def post_process(self, output: List[Any]) -> None: for x in output: if is_item(x): missing = [arg for arg in self.args if arg not in ItemAdapter(x)]