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

Type tests related to requests and responses (#7864)

This commit is contained in:
Adrian 2026-08-03 14:16:23 +02:00 committed by GitHub
parent cde7af87fa
commit 298c9e610e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 214 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
pyproject.toml
View File

@ -136,11 +136,6 @@ module = [
"tests.test_feedexport_storages",
"tests.test_feedexport_uri_params",
"tests.test_http2_client_protocol",
"tests.test_http_headers",
"tests.test_http_request",
"tests.test_http_request_form",
"tests.test_http_response",
"tests.test_http_response_text",
"tests.test_item",
"tests.test_linkextractors",
"tests.test_loader",
@ -152,10 +147,6 @@ module = [
"tests.test_pipeline_media",
"tests.test_pipelines",
"tests.test_pqueues",
"tests.test_request_attribute_binding",
"tests.test_request_cb_kwargs",
"tests.test_request_dict",
"tests.test_request_left",
"tests.test_scheduler_base",
"tests.test_settings",
"tests.test_spider",
@ -166,8 +157,6 @@ module = [
"tests.test_squeues",
"tests.test_squeues_request",
"tests.test_stats",
"tests.utils.bases.http_request",
"tests.utils.bases.http_response",
"tests.utils.bases.spider",
]
check_untyped_defs = false

8
scrapy/http/request/form.py
View File

@ -7,7 +7,7 @@ See documentation in docs/topics/request-response.rst
from __future__ import annotations
from collections.abc import Iterable
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias, cast
from urllib.parse import urlencode, urljoin, urlsplit, urlunsplit
from warnings import warn
@ -34,7 +34,7 @@ if TYPE_CHECKING:
FormdataVType: TypeAlias = str | Iterable[str]
FormdataKVType: TypeAlias = tuple[str, FormdataVType]
FormdataType: TypeAlias = dict[str, FormdataVType] | list[FormdataKVType] | None
FormdataType: TypeAlias = Mapping[str, FormdataVType] | Iterable[FormdataKVType] | None
class FormRequest(Request):
@ -100,7 +100,7 @@ class FormRequest(Request):
super().__init__(*args, **kwargs)
if formdata:
items = formdata.items() if isinstance(formdata, dict) else formdata
items = formdata.items() if isinstance(formdata, Mapping) else formdata
form_query_str = _urlencode(items, self.encoding)
if self.method == "POST":
self.headers.setdefault(
@ -248,7 +248,7 @@ def _get_inputs(
if clickable and clickable[0] not in formdata and clickable[0] is not None:
values.append(clickable)
formdata_items = formdata.items() if isinstance(formdata, dict) else formdata
formdata_items = formdata.items() if isinstance(formdata, Mapping) else formdata
values.extend((k, v) for k, v in formdata_items if v is not None)
return values

29
tests/test_http_headers.py
View File

@ -6,9 +6,6 @@ from scrapy.http import Headers
class TestHeaders:
def assertSortedEqual(self, first, second, msg=None):
assert sorted(first) == sorted(second), msg
def test_basics(self):
h = Headers({"Content-Type": "text/html", "Content-Length": 1234})
assert h["Content-Type"]
@ -39,7 +36,7 @@ class TestHeaders:
assert h["X-Forwarded-For"] == b"ip2"
assert h.get("X-Forwarded-For") == b"ip2"
assert h.getlist("X-Forwarded-For") == [b"ip1", b"ip2"]
assert h.getlist("X-Forwarded-For") is not hlist
assert h.getlist("X-Forwarded-For") is not hlist # type: ignore[comparison-overlap]
def test_multivalue_for_one_header(self):
h = Headers((("a", "b"), ("a", "c")))
@ -49,19 +46,19 @@ class TestHeaders:
def test_encode_utf8(self):
h = Headers({"key": "\xa3"}, encoding="utf-8")
key, val = dict(h).popitem()
key, val = dict(h.items()).popitem()
assert isinstance(key, bytes), key
assert isinstance(val[0], bytes), val[0]
assert val[0] == b"\xc2\xa3"
def test_encode_latin1(self):
h = Headers({"key": "\xa3"}, encoding="latin1")
_, val = dict(h).popitem()
_, val = dict(h.items()).popitem()
assert val[0] == b"\xa3"
def test_encode_multiple(self):
h = Headers({"key": ["\xa3"]}, encoding="utf-8")
_, val = dict(h).popitem()
_, val = dict(h.items()).popitem()
assert val[0] == b"\xc2\xa3"
def test_delete_and_contains(self):
@ -75,7 +72,7 @@ class TestHeaders:
h = Headers()
hlist = ["ip1", "ip2"]
olist = h.setdefault("X-Forwarded-For", hlist)
assert h.getlist("X-Forwarded-For") is not hlist
assert h.getlist("X-Forwarded-For") is not hlist # type: ignore[comparison-overlap]
assert h.getlist("X-Forwarded-For") is olist
h = Headers()
@ -87,16 +84,16 @@ class TestHeaders:
idict = {"Content-Type": "text/html", "X-Forwarded-For": ["ip1", "ip2"]}
h = Headers(idict)
assert dict(h) == {
assert dict(h.items()) == {
b"Content-Type": [b"text/html"],
b"X-Forwarded-For": [b"ip1", b"ip2"],
}
self.assertSortedEqual(h.keys(), [b"X-Forwarded-For", b"Content-Type"])
self.assertSortedEqual(
h.items(),
[(b"X-Forwarded-For", [b"ip1", b"ip2"]), (b"Content-Type", [b"text/html"])],
)
self.assertSortedEqual(h.values(), [b"ip2", b"text/html"])
assert sorted(h.keys()) == [b"Content-Type", b"X-Forwarded-For"]
assert sorted(h.items()) == [
(b"Content-Type", [b"text/html"]),
(b"X-Forwarded-For", [b"ip1", b"ip2"]),
]
assert set(h.values()) == {b"ip2", b"text/html"}
def test_update(self):
h = Headers()
@ -162,4 +159,4 @@ class TestHeaders:
with pytest.raises(TypeError, match="Unsupported value type"):
Headers().setdefault("foo", object())
with pytest.raises(TypeError, match="Unsupported value type"):
Headers().setlist("foo", [object()])
Headers().setlist("foo", [object()]) # type: ignore[list-item]

3
tests/test_http_request.py
View File

@ -1,4 +1,5 @@
import xmlrpc.client
from typing import Any
import pytest
@ -17,7 +18,7 @@ class TestXmlRpcRequest(TestRequestBase):
default_method = "POST"
default_headers = {b"Content-Type": [b"text/xml"]}
def _test_request(self, **kwargs):
def _test_request(self, **kwargs: Any) -> None:
r = self.request_class("http://scrapytest.org/rpc2", **kwargs)
assert r.headers[b"Content-Type"] == b"text/xml"
assert r.body == to_bytes(

75
tests/test_http_request_form.py
View File

@ -2,6 +2,7 @@ from __future__ import annotations
import re
import warnings
from typing import TYPE_CHECKING, Any
from urllib.parse import parse_qs, unquote_to_bytes
import pytest
@ -12,20 +13,32 @@ from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.python import to_unicode
from tests.utils.bases.http_request import TestRequestBase
if TYPE_CHECKING:
from scrapy import Request
def _buildresponse(body, **kwargs):
def _buildresponse(body: bytes | str, **kwargs: Any) -> HtmlResponse:
kwargs.setdefault("body", body)
kwargs.setdefault("url", "http://example.com")
kwargs.setdefault("encoding", "utf-8")
return HtmlResponse(**kwargs)
def _qs(req, encoding="utf-8", to_unicode=False):
qs = req.body if req.method == "POST" else req.url.partition("?")[2]
uqs = unquote_to_bytes(qs)
if to_unicode:
uqs = uqs.decode(encoding)
return parse_qs(uqs, True)
def _query_string(req: Request) -> bytes:
return req.body if req.method == "POST" else req.url.partition("?")[2].encode()
def _qs(req: Request) -> dict[bytes, list[bytes]]:
return parse_qs(unquote_to_bytes(_query_string(req)), True)
def _qs_unicode(req: Request, encoding: str = "utf-8") -> dict[str, list[str]]:
qs = unquote_to_bytes(_query_string(req)).decode(encoding)
return parse_qs(qs, True)
def _assert_query_equal(first: bytes, second: bytes) -> None:
assert sorted(to_unicode(first).split("&")) == sorted(to_unicode(second).split("&"))
# FormRequest.from_response() is deprecated in favor of form2request, so the
@ -34,11 +47,6 @@ def _qs(req, encoding="utf-8", to_unicode=False):
class TestFormRequest(TestRequestBase):
request_class = FormRequest
def assertQueryEqual(self, first, second, msg=None):
first = to_unicode(first).split("&")
second = to_unicode(second).split("&")
assert sorted(first) == sorted(second), msg
def test_init_not_deprecated(self):
# Building a request directly from form data is not deprecated.
with warnings.catch_warnings():
@ -75,20 +83,22 @@ class TestFormRequest(TestRequestBase):
assert fs[b"b"] == [b"2"]
assert fs.get(b"c") is None
data = {"a": "1", "b": "2"}
mapping = {"a": "1", "b": "2"}
fs = _qs(
self.request_class("http://www.example.com/", method="GET", formdata=data)
self.request_class(
"http://www.example.com/", method="GET", formdata=mapping
)
)
assert fs[b"a"] == [b"1"]
assert fs[b"b"] == [b"2"]
def test_default_encoding_bytes(self):
# using default encoding (utf-8)
data = {b"one": b"two", b"price": b"\xc2\xa3 100"}
data: dict[Any, Any] = {b"one": b"two", b"price": b"\xc2\xa3 100"}
r2 = self.request_class("http://www.example.com", formdata=data)
assert r2.method == "POST"
assert r2.encoding == "utf-8"
self.assertQueryEqual(r2.body, b"price=%C2%A3+100&one=two")
_assert_query_equal(r2.body, b"price=%C2%A3+100&one=two")
assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded"
def test_default_encoding_textual_data(self):
@ -97,26 +107,26 @@ class TestFormRequest(TestRequestBase):
r2 = self.request_class("http://www.example.com", formdata=data)
assert r2.method == "POST"
assert r2.encoding == "utf-8"
self.assertQueryEqual(r2.body, b"price=%C2%A3+100&%C2%B5+one=two")
_assert_query_equal(r2.body, b"price=%C2%A3+100&%C2%B5+one=two")
assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded"
def test_default_encoding_mixed_data(self):
# using default encoding (utf-8)
data = {"\u00b5one": b"two", b"price\xc2\xa3": "\u00a3 100"}
data: dict[Any, Any] = {"\u00b5one": b"two", b"price\xc2\xa3": "\u00a3 100"}
r2 = self.request_class("http://www.example.com", formdata=data)
assert r2.method == "POST"
assert r2.encoding == "utf-8"
self.assertQueryEqual(r2.body, b"%C2%B5one=two&price%C2%A3=%C2%A3+100")
_assert_query_equal(r2.body, b"%C2%B5one=two&price%C2%A3=%C2%A3+100")
assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded"
def test_custom_encoding_bytes(self):
data = {b"\xb5 one": b"two", b"price": b"\xa3 100"}
data: dict[Any, Any] = {b"\xb5 one": b"two", b"price": b"\xa3 100"}
r2 = self.request_class(
"http://www.example.com", formdata=data, encoding="latin1"
)
assert r2.method == "POST"
assert r2.encoding == "latin1"
self.assertQueryEqual(r2.body, b"price=%A3+100&%B5+one=two")
_assert_query_equal(r2.body, b"price=%A3+100&%B5+one=two")
assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded"
def test_custom_encoding_textual_data(self):
@ -131,7 +141,7 @@ class TestFormRequest(TestRequestBase):
# using multiples values for a single key
data = {"price": "\xa3 100", "colours": ["red", "blue", "green"]}
r3 = self.request_class("http://www.example.com", formdata=data)
self.assertQueryEqual(
_assert_query_equal(
r3.body, b"colours=red&colours=blue&colours=green&price=%C2%A3+100"
)
@ -173,7 +183,7 @@ class TestFormRequest(TestRequestBase):
assert req.method == "POST"
assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded"
assert req.url == "http://www.example.com/this/post.php"
fs = _qs(req, to_unicode=True)
fs = _qs_unicode(req)
assert set(fs["test £"]) == {"val1", "val2"}
assert set(fs["one"]) == {"two", "three"}
assert fs["test2"] == ["xxx µ"]
@ -196,7 +206,7 @@ class TestFormRequest(TestRequestBase):
assert req.method == "POST"
assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded"
assert req.url == "http://www.example.com/this/post.php"
fs = _qs(req, to_unicode=True, encoding="latin1")
fs = _qs_unicode(req, encoding="latin1")
assert set(fs["test £"]) == {"val1", "val2"}
assert set(fs["one"]) == {"two", "three"}
assert fs["test2"] == ["xxx µ"]
@ -218,7 +228,7 @@ class TestFormRequest(TestRequestBase):
assert req.method == "POST"
assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded"
assert req.url == "http://www.example.com/this/post.php"
fs = _qs(req, to_unicode=True)
fs = _qs_unicode(req)
assert set(fs["test £"]) == {"val1", "val2"}
assert set(fs["one"]) == {"two", "three"}
assert fs["test2"] == ["xxx µ"]
@ -305,7 +315,10 @@ class TestFormRequest(TestRequestBase):
<input type="hidden" name="two" value="3">
</form>"""
)
req = self.request_class.from_response(response, formdata={"two": None})
req = self.request_class.from_response(
response,
formdata={"two": None}, # type: ignore[arg-type]
)
fs = _qs(req)
assert fs[b"one"] == [b"1"]
assert b"two" not in fs
@ -450,7 +463,7 @@ class TestFormRequest(TestRequestBase):
req = self.request_class.from_response(
response, clickdata={"name": "price in \u00a3"}
)
fs = _qs(req, to_unicode=True)
fs = _qs_unicode(req)
assert fs["price in \u00a3"]
def test_from_response_unicode_clickdata_latin1(self):
@ -466,7 +479,7 @@ class TestFormRequest(TestRequestBase):
req = self.request_class.from_response(
response, clickdata={"name": "price in \u00a5"}
)
fs = _qs(req, to_unicode=True, encoding="latin1")
fs = _qs_unicode(req, encoding="latin1")
assert fs["price in \u00a5"]
def test_from_response_multiple_forms_clickdata(self):
@ -737,7 +750,7 @@ class TestFormRequest(TestRequestBase):
</form>"""
)
req = self.request_class.from_response(res)
fs = _qs(req, to_unicode=True)
fs = _qs_unicode(req)
assert fs == {"i1": ["i1v2"], "i2": ["i2v1"], "i4": ["i4v2", "i4v3"]}
def test_from_response_radio(self):
@ -1022,7 +1035,7 @@ class TestFormRequest(TestRequestBase):
with pytest.raises(
ValueError, match="formdata should be a dict or iterable of tuples"
):
FormRequest.from_response(response, formdata=123)
FormRequest.from_response(response, formdata=123) # type: ignore[arg-type]
def test_form_response_with_custom_invalid_formdata_value_error(self):
"""Test that a ValueError is raised for fault-inducing iterable formdata input"""
@ -1037,7 +1050,7 @@ class TestFormRequest(TestRequestBase):
with pytest.raises(
ValueError, match="formdata should be a dict or iterable of tuples"
):
FormRequest.from_response(response, formdata=("a",))
FormRequest.from_response(response, formdata=("a",)) # type: ignore[arg-type]
def test_get_form_with_xpath_no_form_parent(self):
"""Test that _get_from raised a ValueError when an XPath selects an element

15
tests/test_http_response_text.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import codecs
from typing import cast
from unittest import mock
import pytest
@ -14,6 +15,12 @@ from tests.utils.bases.http_response import TestResponseBase
class TestTextResponse(TestResponseBase):
response_class = TextResponse
def _links_response(self) -> TextResponse:
return cast("TextResponse", super()._links_response())
def _links_response_no_href(self) -> TextResponse:
return cast("TextResponse", super()._links_response_no_href())
def test_follow_None_encoding(self):
# unlike the base Response, TextResponse.follow() falls back to the
# response encoding when encoding is None instead of raising
@ -21,7 +28,7 @@ class TestTextResponse(TestResponseBase):
req = r.follow("foo", encoding=None)
assert req.encoding == "cp1252"
def test_replace(self):
def test_replace(self) -> None:
super().test_replace()
r1 = self.response_class(
"http://www.example.com", body="hello", encoding="cp852"
@ -344,7 +351,7 @@ class TestTextResponse(TestResponseBase):
def test_follow_selector_list(self):
resp = self._links_response()
with pytest.raises(ValueError, match="SelectorList"):
resp.follow(resp.css("a"))
resp.follow(resp.css("a")) # type: ignore[arg-type]
def test_follow_selector_invalid(self):
resp = self._links_response()
@ -616,7 +623,7 @@ class CustomResponse(TextResponse):
class TestCustomResponse(TestTextResponse):
response_class = CustomResponse
def test_copy(self):
def test_copy(self) -> None:
super().test_copy()
r1 = self.response_class(
url="https://example.org",
@ -632,7 +639,7 @@ class TestCustomResponse(TestTextResponse):
assert r1.lost == "lost"
assert r2.lost is None
def test_replace(self):
def test_replace(self) -> None:
super().test_replace()
r1 = self.response_class(
url="https://example.org",

5
tests/test_request_attribute_binding.py
View File

@ -85,6 +85,7 @@ class TestCrawl:
url = self.mockserver.url("/status?n=200")
crawler = get_crawler(SingleRequestSpider)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
assert isinstance(crawler.spider, SingleRequestSpider)
response = crawler.spider.meta["responses"][0]
assert response.request.url == url
@ -94,6 +95,7 @@ class TestCrawl:
url = self.mockserver.url(f"/status?n={status}")
crawler = get_crawler(SingleRequestSpider)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
assert isinstance(crawler.spider, SingleRequestSpider)
failure = crawler.spider.meta["failure"]
response = failure.value.response
assert failure.request.url == url
@ -111,6 +113,7 @@ class TestCrawl:
},
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
assert isinstance(crawler.spider, SingleRequestSpider)
failure = crawler.spider.meta["failure"]
assert failure.request.url == url
assert isinstance(failure.value, ZeroDivisionError)
@ -178,6 +181,7 @@ class TestCrawl:
},
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
assert isinstance(crawler.spider, SingleRequestSpider)
response = crawler.spider.meta["responses"][0]
assert response.body == b"Caught ZeroDivisionError"
assert response.request.url == OVERRIDDEN_URL
@ -201,6 +205,7 @@ class TestCrawl:
},
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
assert isinstance(crawler.spider, SingleRequestSpider)
response = crawler.spider.meta["responses"][0]
assert response.body == b"Caught ZeroDivisionError"
assert response.request.url == url

26
tests/test_request_cb_kwargs.py
View File

@ -49,6 +49,7 @@ class InjectArgumentsSpiderMiddleware:
async for element in result:
if (
isinstance(element, Request)
and element.callback
and element.callback.__name__ == "parse_spider_mw_2"
):
element.cb_kwargs["from_process_spider_output"] = True
@ -68,7 +69,12 @@ class KeywordArgumentsSpider(MockServerSpider):
checks: list[bool] = []
def _inc_checks(self, count: int = 1) -> None:
assert self.crawler.stats
self.crawler.stats.inc_value("boolean_checks", count)
async def start(self):
assert self.mockserver
data = {"key": "value", "number": 123, "callback": "some_callback"}
yield Request(self.mockserver.url("/first"), self.parse_first, cb_kwargs=data)
yield Request(
@ -89,9 +95,10 @@ class KeywordArgumentsSpider(MockServerSpider):
yield Request(self.mockserver.url("/spider_mw"), self.parse_spider_mw)
def parse_first(self, response, key, number):
assert self.mockserver
self.checks.append(key == "value")
self.checks.append(number == 123)
self.crawler.stats.inc_value("boolean_checks", 2)
self._inc_checks(2)
yield response.follow(
self.mockserver.url("/two"),
self.parse_second,
@ -100,28 +107,28 @@ class KeywordArgumentsSpider(MockServerSpider):
def parse_second(self, response, new_key):
self.checks.append(new_key == "new_value")
self.crawler.stats.inc_value("boolean_checks")
self._inc_checks()
def parse_general(self, response, **kwargs):
if response.url.endswith("/general_with"):
self.checks.append(kwargs["key"] == "value")
self.checks.append(kwargs["number"] == 123)
self.checks.append(kwargs["callback"] == "some_callback")
self.crawler.stats.inc_value("boolean_checks", 3)
self._inc_checks(3)
elif response.url.endswith("/general_without"):
self.checks.append(kwargs == {})
self.crawler.stats.inc_value("boolean_checks")
self._inc_checks()
def parse_no_kwargs(self, response):
self.checks.append(response.url.endswith("/no_kwargs"))
self.crawler.stats.inc_value("boolean_checks")
self._inc_checks()
def parse_default(self, response, key, number=None, default=99):
self.checks.append(response.url.endswith("/default"))
self.checks.append(key == "value")
self.checks.append(number == 123)
self.checks.append(default == 99)
self.crawler.stats.inc_value("boolean_checks", 4)
self._inc_checks(4)
def parse_takes_less(self, response, key, callback):
"""
@ -140,17 +147,18 @@ class KeywordArgumentsSpider(MockServerSpider):
):
self.checks.append(bool(from_process_request))
self.checks.append(bool(from_process_response))
self.crawler.stats.inc_value("boolean_checks", 2)
self._inc_checks(2)
def parse_spider_mw(self, response, from_process_spider_input, from_process_start):
assert self.mockserver
self.checks.append(bool(from_process_spider_input))
self.checks.append(bool(from_process_start))
self.crawler.stats.inc_value("boolean_checks", 2)
self._inc_checks(2)
return Request(self.mockserver.url("/spider_mw_2"), self.parse_spider_mw_2)
def parse_spider_mw_2(self, response, from_process_spider_output):
self.checks.append(bool(from_process_spider_output))
self.crawler.stats.inc_value("boolean_checks", 1)
self._inc_checks()
class TestCallbackKeywordArguments:

58
tests/test_request_dict.py
View File

@ -1,7 +1,10 @@
from typing import Any
import pytest
from twisted.python.failure import Failure
from scrapy import Request, Spider
from scrapy.http import JsonRequest
from scrapy.http import JsonRequest, Response
from scrapy.utils.request import request_from_dict
@ -10,7 +13,7 @@ class CustomRequest(Request):
class TestRequestSerialization:
def setup_method(self):
def setup_method(self) -> None:
self.spider = MethodsSpider()
def test_basic(self):
@ -42,12 +45,14 @@ class TestRequestSerialization:
r = Request("http://www.example.com", body=b"\xc2\xa3")
self._assert_serializes_ok(r)
def _assert_serializes_ok(self, request, spider=None):
def _assert_serializes_ok(
self, request: Request, spider: Spider | None = None
) -> None:
d = request.to_dict(spider=spider)
request2 = request_from_dict(d, spider=spider)
self._assert_same_request(request, request2)
def _assert_same_request(self, r1, r2):
def _assert_same_request(self, r1: Request, r2: Request) -> None:
assert r1.__class__ == r2.__class__
assert r1.url == r2.url
assert r1.callback == r2.callback
@ -64,6 +69,7 @@ class TestRequestSerialization:
assert r1.dont_filter == r2.dont_filter
assert r1.flags == r2.flags
if isinstance(r1, JsonRequest):
assert isinstance(r2, JsonRequest)
assert r1.dumps_kwargs == r2.dumps_kwargs
def test_request_class(self):
@ -83,8 +89,8 @@ class TestRequestSerialization:
def test_reference_callback_serialization(self):
r = Request(
"http://www.example.com",
callback=self.spider.parse_item_reference,
errback=self.spider.handle_error_reference,
callback=self.spider.parse_item_reference, # type: ignore[arg-type,misc]
errback=self.spider.handle_error_reference, # type: ignore[arg-type,misc]
)
self._assert_serializes_ok(r, spider=self.spider)
request_dict = r.to_dict(spider=self.spider)
@ -94,8 +100,8 @@ class TestRequestSerialization:
def test_private_reference_callback_serialization(self):
r = Request(
"http://www.example.com",
callback=self.spider._MethodsSpider__parse_item_reference,
errback=self.spider._MethodsSpider__handle_error_reference,
callback=self.spider._MethodsSpider__parse_item_reference, # type: ignore[attr-defined]
errback=self.spider._MethodsSpider__handle_error_reference, # type: ignore[attr-defined]
)
self._assert_serializes_ok(r, spider=self.spider)
request_dict = r.to_dict(spider=self.spider)
@ -105,7 +111,7 @@ class TestRequestSerialization:
def test_private_callback_serialization(self):
r = Request(
"http://www.example.com",
callback=self.spider._MethodsSpider__parse_item_private,
callback=self.spider._MethodsSpider__parse_item_private, # type: ignore[attr-defined]
errback=self.spider.handle_error,
)
self._assert_serializes_ok(r, spider=self.spider)
@ -113,7 +119,7 @@ class TestRequestSerialization:
def test_mixin_private_callback_serialization(self):
r = Request(
"http://www.example.com",
callback=self.spider._SpiderMixin__mixin_callback,
callback=self.spider._SpiderMixin__mixin_callback, # type: ignore[attr-defined]
errback=self.spider.handle_error,
)
self._assert_serializes_ok(r, spider=self.spider)
@ -127,7 +133,7 @@ class TestRequestSerialization:
self._assert_serializes_ok(r, spider=self.spider)
def test_unserializable_callback1(self):
r = Request("http://www.example.com", callback=lambda x: x)
r = Request("http://www.example.com", callback=lambda x: x) # type: ignore[misc]
with pytest.raises(
ValueError, match="is not an instance method in: <MethodsSpider"
):
@ -144,12 +150,12 @@ class TestRequestSerialization:
class MySpider(Spider):
name = "my_spider"
def parse(self, response):
def parse(self, response: Response) -> None:
pass
spider = MySpider()
r = Request("http://www.example.com", callback=spider.parse)
spider.parse = None
spider.parse = None # type: ignore[method-assign,assignment]
with pytest.raises(ValueError, match="is not an instance method in: <MySpider"):
r.to_dict(spider=spider)
@ -157,7 +163,7 @@ class TestRequestSerialization:
"""Callback method is not available in the spider passed to from_dict"""
spider = SpiderDelegation()
r = Request("http://www.example.com", callback=spider.delegated_callback)
d = r.to_dict(spider=spider)
d = r.to_dict(spider=spider) # type: ignore[arg-type]
with pytest.raises(
ValueError, match="Method 'delegated_callback' not found in: <Spider"
):
@ -165,28 +171,30 @@ class TestRequestSerialization:
class SpiderMixin:
def __mixin_callback(self, response): # pylint: disable=unused-private-member
def __mixin_callback( # pylint: disable=unused-private-member
self, response: Response
) -> None:
pass
class SpiderDelegation:
def delegated_callback(self, response):
def delegated_callback(self, response: Response) -> None:
pass
def parse_item(response):
def parse_item(response: Response) -> None:
pass
def handle_error(failure):
def handle_error(failure: Failure) -> None:
pass
def private_parse_item(response):
def private_parse_item(response: Response) -> None:
pass
def private_handle_error(failure):
def private_handle_error(failure: Failure) -> None:
pass
@ -197,15 +205,17 @@ class MethodsSpider(Spider, SpiderMixin):
__parse_item_reference = private_parse_item
__handle_error_reference = private_handle_error
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.delegated_callback = SpiderDelegation().delegated_callback
def parse_item(self, response):
def parse_item(self, response: Response) -> None:
pass
def handle_error(self, failure):
def handle_error(self, failure: Failure) -> None:
pass
def __parse_item_private(self, response): # pylint: disable=unused-private-member
def __parse_item_private( # pylint: disable=unused-private-member
self, response: Response
) -> None:
pass

43
tests/test_request_left.py
View File

@ -1,57 +1,62 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from scrapy.signals import request_left_downloader
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.utils.decorators import inline_callbacks_test
if TYPE_CHECKING:
from scrapy import Request
from scrapy.crawler import Crawler
from tests.mockserver.http import MockServer
class SignalCatcherSpider(Spider):
name = "signal_catcher"
def __init__(self, crawler, url, *args, **kwargs):
def __init__(self, crawler: Crawler, url: str, *args: Any, **kwargs: Any):
super().__init__(*args, **kwargs)
crawler.signals.connect(self.on_request_left, signal=request_left_downloader)
self.caught_times = 0
self.start_urls = [url]
@classmethod
def from_crawler(cls, crawler, *args, **kwargs):
def from_crawler(
cls, crawler: Crawler, *args: Any, **kwargs: Any
) -> SignalCatcherSpider:
return cls(crawler, *args, **kwargs)
def on_request_left(self, request, spider):
def on_request_left(self, request: Request, spider: Spider) -> None:
self.caught_times += 1
class TestCatching:
@classmethod
def setup_class(cls):
cls.mockserver = MockServer()
cls.mockserver.__enter__()
@classmethod
def teardown_class(cls):
cls.mockserver.__exit__(None, None, None)
@inline_callbacks_test
def test_success(self):
def test_success(self, mockserver: MockServer):
crawler = get_crawler(SignalCatcherSpider)
yield crawler.crawl(self.mockserver.url("/status?n=200"))
yield crawler.crawl(mockserver.url("/status?n=200"))
assert isinstance(crawler.spider, SignalCatcherSpider)
assert crawler.spider.caught_times == 1
@inline_callbacks_test
def test_timeout(self):
def test_timeout(self, mockserver: MockServer):
crawler = get_crawler(SignalCatcherSpider, {"DOWNLOAD_TIMEOUT": 0.1})
yield crawler.crawl(self.mockserver.url("/delay?n=0.2"))
yield crawler.crawl(mockserver.url("/delay?n=0.2"))
assert isinstance(crawler.spider, SignalCatcherSpider)
assert crawler.spider.caught_times == 1
@inline_callbacks_test
def test_disconnect(self):
def test_disconnect(self, mockserver: MockServer):
crawler = get_crawler(SignalCatcherSpider)
yield crawler.crawl(self.mockserver.url("/drop"))
yield crawler.crawl(mockserver.url("/drop"))
assert isinstance(crawler.spider, SignalCatcherSpider)
assert crawler.spider.caught_times == 1
@inline_callbacks_test
def test_noconnect(self):
crawler = get_crawler(SignalCatcherSpider)
yield crawler.crawl("http://thereisdefinetelynosuchdomain.com")
assert isinstance(crawler.spider, SignalCatcherSpider)
assert crawler.spider.caught_times == 1

59
tests/utils/bases/http_request.py
View File

@ -3,8 +3,9 @@ from abc import ABC, abstractmethod
from typing import Any
import pytest
from twisted.python.failure import Failure
from scrapy.http import Headers, Request
from scrapy.http import Headers, Request, Response
from scrapy.http.request import NO_CALLBACK
from scrapy.utils.request import request_to_curl
@ -22,15 +23,15 @@ class TestRequestBase(ABC):
def test_init(self):
# Request requires url in the __init__ method
with pytest.raises(TypeError):
self.request_class()
self.request_class() # type: ignore[call-arg]
# url argument must be basestring
with pytest.raises(TypeError):
self.request_class(123)
self.request_class(123) # type: ignore[arg-type]
# priority argument must be an integer
with pytest.raises(TypeError, match="Request priority not an integer"):
self.request_class("http://www.example.com", priority="1")
self.request_class("http://www.example.com", priority="1") # type: ignore[arg-type]
r = self.request_class("http://www.example.com")
assert isinstance(r.url, str)
@ -205,14 +206,17 @@ class TestRequestBase(ABC):
def test_copy(self):
"""Test Request copy"""
def somecallback():
def somecallback(response: Response) -> None:
pass
def someerrback(failure: Failure) -> None:
pass
r1 = self.request_class(
"http://www.example.com",
flags=["f1", "f2"],
callback=somecallback,
errback=somecallback,
errback=someerrback,
)
r1.meta["foo"] = "bar"
r1.cb_kwargs["key"] = "value"
@ -220,7 +224,7 @@ class TestRequestBase(ABC):
# make sure callbaclks are copied
assert r1.callback is somecallback
assert r1.errback is somecallback
assert r1.errback is someerrback
assert r2.callback is r1.callback
assert r2.errback is r1.errback
@ -251,7 +255,7 @@ class TestRequestBase(ABC):
def test_copy_inherited_classes(self):
"""Test Request children copies preserve their class"""
class CustomRequest(self.request_class):
class CustomRequest(self.request_class): # type: ignore[misc,name-defined]
pass
r1 = CustomRequest("http://www.example.com")
@ -283,7 +287,9 @@ class TestRequestBase(ABC):
assert r4.dont_filter is False
# the cls argument allows changing the resulting class
custom_request_cls = type("CustomRequest", (self.request_class,), {})
custom_request_cls: type[Request] = type(
"CustomRequest", (self.request_class,), {}
)
r5 = r1.replace(cls=custom_request_cls)
assert isinstance(r5, custom_request_cls)
assert r5.url == r1.url
@ -295,33 +301,36 @@ class TestRequestBase(ABC):
def test_immutable_attributes(self):
r = self.request_class("http://example.com")
with pytest.raises(AttributeError):
r.url = "http://example2.com"
r.url = "http://example2.com" # type: ignore[misc]
with pytest.raises(AttributeError):
r.body = "xxx"
r.body = "xxx" # type: ignore[misc,assignment]
def test_callback_and_errback(self):
def a_function():
def a_callback(response: Response) -> None:
pass
def an_errback(failure: Failure) -> None:
pass
r1 = self.request_class("http://example.com")
assert r1.callback is None
assert r1.errback is None
r2 = self.request_class("http://example.com", callback=a_function)
assert r2.callback is a_function
r2 = self.request_class("http://example.com", callback=a_callback)
assert r2.callback is a_callback
assert r2.errback is None
r3 = self.request_class("http://example.com", errback=a_function)
r3 = self.request_class("http://example.com", errback=an_errback)
assert r3.callback is None
assert r3.errback is a_function
assert r3.errback is an_errback
r4 = self.request_class(
url="http://example.com",
callback=a_function,
errback=a_function,
callback=a_callback,
errback=an_errback,
)
assert r4.callback is a_function
assert r4.errback is a_function
assert r4.callback is a_callback
assert r4.errback is an_errback
r5 = self.request_class(
url="http://example.com",
@ -329,18 +338,18 @@ class TestRequestBase(ABC):
errback=NO_CALLBACK,
)
assert r5.callback is NO_CALLBACK
assert r5.errback is NO_CALLBACK
assert r5.errback is NO_CALLBACK # type: ignore[comparison-overlap]
def test_callback_and_errback_type(self):
with pytest.raises(TypeError):
self.request_class("http://example.com", callback="a_function")
self.request_class("http://example.com", callback="a_function") # type: ignore[arg-type]
with pytest.raises(TypeError):
self.request_class("http://example.com", errback="a_function")
self.request_class("http://example.com", errback="a_function") # type: ignore[arg-type]
with pytest.raises(TypeError):
self.request_class(
url="http://example.com",
callback="a_function",
errback="a_function",
callback="a_function", # type: ignore[arg-type]
errback="a_function", # type: ignore[arg-type]
)
def test_setters(self):

48
tests/utils/bases/http_response.py
View File

@ -7,7 +7,7 @@ import pytest
from w3lib.encoding import resolve_encoding
from scrapy.exceptions import NotSupported
from scrapy.http import Headers, Request, Response
from scrapy.http import Headers, Request, Response, TextResponse
from scrapy.link import Link
from scrapy.utils._deps_compat import W3LIB_STRIPS_URLS
from tests import get_testdata
@ -15,6 +15,8 @@ from tests import get_testdata
if TYPE_CHECKING:
from collections.abc import Iterable
from parsel import Selector
class TestResponseBase(ABC):
@property
@ -25,14 +27,14 @@ class TestResponseBase(ABC):
def test_init(self):
# Response requires url in the constructor
with pytest.raises(TypeError):
self.response_class()
self.response_class() # type: ignore[call-arg]
assert isinstance(
self.response_class("http://example.com/"), self.response_class
)
with pytest.raises(TypeError):
self.response_class(b"http://example.com")
self.response_class(b"http://example.com") # type: ignore[arg-type]
with pytest.raises(TypeError):
self.response_class(url="http://example.com", body={})
self.response_class(url="http://example.com", body={}) # type: ignore[arg-type]
# body can be str or None
assert isinstance(
self.response_class("http://example.com/", body=b""),
@ -67,12 +69,12 @@ class TestResponseBase(ABC):
r = self.response_class("http://www.example.com", status=301)
assert r.status == 301
r = self.response_class("http://www.example.com", status="301")
r = self.response_class("http://www.example.com", status="301") # type: ignore[arg-type]
assert r.status == 301
with pytest.raises(ValueError, match=r"invalid literal for int\(\)"):
self.response_class("http://example.com", status="lala200")
self.response_class("http://example.com", status="lala200") # type: ignore[arg-type]
def test_copy(self):
def test_copy(self) -> None:
"""Test Response copy"""
r1 = self.response_class("http://www.example.com", body=b"Some body")
@ -121,7 +123,7 @@ class TestResponseBase(ABC):
def test_copy_inherited_classes(self):
"""Test Response children copies preserve their class"""
class CustomResponse(self.response_class):
class CustomResponse(self.response_class): # type: ignore[misc,name-defined]
pass
r1 = CustomResponse("http://www.example.com")
@ -129,7 +131,7 @@ class TestResponseBase(ABC):
assert isinstance(r2, CustomResponse)
def test_replace(self):
def test_replace(self) -> None:
"""Test Response.replace() method"""
hdrs = Headers({"key": "value"})
r1 = self.response_class("http://www.example.com")
@ -146,7 +148,9 @@ class TestResponseBase(ABC):
assert r4.body == b""
assert not r4.flags
def _assert_response_values(self, response, encoding, body):
def _assert_response_values(
self, response: TextResponse, encoding: str, body: str | bytes
) -> None:
if isinstance(body, str):
body_unicode = body
body_bytes = body.encode(encoding)
@ -160,15 +164,15 @@ class TestResponseBase(ABC):
assert response.body == body_bytes
assert response.text == body_unicode
def _assert_response_encoding(self, response, encoding):
def _assert_response_encoding(self, response: TextResponse, encoding: str) -> None:
assert response.encoding == resolve_encoding(encoding)
def test_immutable_attributes(self):
r = self.response_class("http://example.com")
with pytest.raises(AttributeError):
r.url = "http://example2.com"
r.url = "http://example2.com" # type: ignore[misc]
with pytest.raises(AttributeError):
r.body = "xxx"
r.body = "xxx" # type: ignore[misc,assignment]
def test_setter_mutable_lazy_loading(self):
"""Mutable attributes are set internally to None only until they are
@ -256,7 +260,7 @@ class TestResponseBase(ABC):
def test_follow_None_url(self):
r = self.response_class("http://example.com")
with pytest.raises(ValueError, match="url can't be None"):
r.follow(None)
r.follow(None) # type: ignore[arg-type]
def test_follow_None_encoding(self):
r = self.response_class("http://example.com")
@ -325,20 +329,20 @@ class TestResponseBase(ABC):
r = self.response_class("http://example.com")
if self.response_class == Response:
with pytest.raises(TypeError):
list(r.follow_all(urls=None))
list(r.follow_all(urls=None)) # type: ignore[arg-type]
with pytest.raises(TypeError):
list(r.follow_all(urls=12345))
list(r.follow_all(urls=12345)) # type: ignore[arg-type]
with pytest.raises(ValueError, match="url can't be None"):
list(r.follow_all(urls=[None]))
list(r.follow_all(urls=[None])) # type: ignore[list-item]
else:
with pytest.raises(
ValueError, match="Please supply exactly one of the following arguments"
):
list(r.follow_all(urls=None))
list(r.follow_all(urls=None)) # type: ignore[arg-type]
with pytest.raises(TypeError):
list(r.follow_all(urls=12345))
list(r.follow_all(urls=12345)) # type: ignore[arg-type]
with pytest.raises(ValueError, match="url can't be None"):
list(r.follow_all(urls=[None]))
list(r.follow_all(urls=[None])) # type: ignore[list-item]
@pytest.mark.xfail(
not W3LIB_STRIPS_URLS,
@ -384,14 +388,14 @@ class TestResponseBase(ABC):
def _assert_followed_url(
self,
follow_obj: str | Link,
follow_obj: str | Link | Selector,
target_url: str,
response: Response | None = None,
encoding: str | None = None,
) -> None:
if response is None:
response = self._links_response()
req = response.follow(follow_obj)
req = response.follow(follow_obj) # type: ignore[arg-type]
assert req.url == target_url
if encoding is not None:
assert req.encoding == encoding