RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Full typing for scrapy/contracts.

This commit is contained in:
Andrey Rakhmatullin 2024-05-19 15:45:51 +05:00
parent 631fc65fad
commit b99526b740
2 changed files with 60 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
scrapy/contracts/__init__.py
View File

@ -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__}"

19
scrapy/contracts/default.py
View File

@ -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)]