From 298c9e610ec29dea378a706e925ad0d897007410 Mon Sep 17 00:00:00 2001 From: Adrian Date: Mon, 3 Aug 2026 14:16:23 +0200 Subject: [PATCH] Type tests related to requests and responses (#7864) --- pyproject.toml | 11 ---- scrapy/http/request/form.py | 8 +-- tests/test_http_headers.py | 29 +++++----- tests/test_http_request.py | 3 +- tests/test_http_request_form.py | 75 +++++++++++++++---------- tests/test_http_response_text.py | 15 +++-- tests/test_request_attribute_binding.py | 5 ++ tests/test_request_cb_kwargs.py | 26 ++++++--- tests/test_request_dict.py | 58 +++++++++++-------- tests/test_request_left.py | 43 +++++++------- tests/utils/bases/http_request.py | 59 ++++++++++--------- tests/utils/bases/http_response.py | 48 ++++++++-------- 12 files changed, 214 insertions(+), 166 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 11e971a35..609c70708 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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 diff --git a/scrapy/http/request/form.py b/scrapy/http/request/form.py index f1a8dbf3b..12745292b 100644 --- a/scrapy/http/request/form.py +++ b/scrapy/http/request/form.py @@ -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 diff --git a/tests/test_http_headers.py b/tests/test_http_headers.py index aff3562e3..e7ed17615 100644 --- a/tests/test_http_headers.py +++ b/tests/test_http_headers.py @@ -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] diff --git a/tests/test_http_request.py b/tests/test_http_request.py index e58ae8f39..b9cec93a1 100644 --- a/tests/test_http_request.py +++ b/tests/test_http_request.py @@ -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( diff --git a/tests/test_http_request_form.py b/tests/test_http_request_form.py index 5e965e8dc..cb18a0b03 100644 --- a/tests/test_http_request_form.py +++ b/tests/test_http_request_form.py @@ -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): """ ) - 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): """ ) 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 diff --git a/tests/test_http_response_text.py b/tests/test_http_response_text.py index 04315ad89..efa63e049 100644 --- a/tests/test_http_response_text.py +++ b/tests/test_http_response_text.py @@ -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", diff --git a/tests/test_request_attribute_binding.py b/tests/test_request_attribute_binding.py index a624d2097..a2a4e8fdb 100644 --- a/tests/test_request_attribute_binding.py +++ b/tests/test_request_attribute_binding.py @@ -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 diff --git a/tests/test_request_cb_kwargs.py b/tests/test_request_cb_kwargs.py index b88893b2b..6c26aa878 100644 --- a/tests/test_request_cb_kwargs.py +++ b/tests/test_request_cb_kwargs.py @@ -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: diff --git a/tests/test_request_dict.py b/tests/test_request_dict.py index 78ff18b15..c7596e45d 100644 --- a/tests/test_request_dict.py +++ b/tests/test_request_dict.py @@ -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: 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: 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 diff --git a/tests/test_request_left.py b/tests/test_request_left.py index 726e0573a..46a16ad1e 100644 --- a/tests/test_request_left.py +++ b/tests/test_request_left.py @@ -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 diff --git a/tests/utils/bases/http_request.py b/tests/utils/bases/http_request.py index 3a1e588ef..2b712f3ab 100644 --- a/tests/utils/bases/http_request.py +++ b/tests/utils/bases/http_request.py @@ -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): diff --git a/tests/utils/bases/http_response.py b/tests/utils/bases/http_response.py index 2fbf6527a..78e14e7b0 100644 --- a/tests/utils/bases/http_response.py +++ b/tests/utils/bases/http_response.py @@ -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