Works
World
", "text/html", ), ) - self.putChild( + put_child( + self, b"enc-gb18030", Data(b"gb18030 encoding
", "text/html; charset=gb18030"), ) - self.putChild(b"redirect", Redirect(b"/redirected")) - self.putChild( - b"redirect-no-meta-refresh", NoMetaRefreshRedirect(b"/redirected") + put_child(self, b"redirect", Redirect(b"/redirected")) + put_child( + self, b"redirect-no-meta-refresh", NoMetaRefreshRedirect(b"/redirected") ) - self.putChild(b"redirected", Data(b"Redirected here", "text/plain")) + put_child(self, b"redirected", Data(b"Redirected here", "text/plain")) numbers = [str(x).encode("utf8") for x in range(2**18)] - self.putChild(b"numbers", Data(b"".join(numbers), "text/plain")) - self.putChild(b"wait", ForeverTakingResource()) - self.putChild(b"hang-after-headers", ForeverTakingResource(write=True)) - self.putChild(b"host", HostHeaderResource()) - self.putChild(b"client-ip", ClientIPResource()) - self.putChild(b"broken", BrokenDownloadResource()) - self.putChild(b"chunked", ChunkedResource()) - self.putChild(b"broken-chunked", BrokenChunkedResource()) - self.putChild(b"contentlength", ContentLengthHeaderResource()) - self.putChild(b"nocontenttype", EmptyContentTypeHeaderResource()) - self.putChild(b"largechunkedfile", LargeChunkedFileResource()) - self.putChild(b"compress", Compress()) - self.putChild(b"duplicate-header", DuplicateHeaderResource()) - self.putChild(b"response-headers", ResponseHeadersResource()) - self.putChild(b"set-cookie", SetCookie()) - self.putChild(b"uri", UriResource()) + put_child(self, b"numbers", Data(b"".join(numbers), "text/plain")) + put_child(self, b"wait", ForeverTakingResource()) + put_child(self, b"hang-after-headers", ForeverTakingResource(write=True)) + put_child(self, b"host", HostHeaderResource()) + put_child(self, b"client-ip", ClientIPResource()) + put_child(self, b"broken", BrokenDownloadResource()) + put_child(self, b"chunked", ChunkedResource()) + put_child(self, b"broken-chunked", BrokenChunkedResource()) + put_child(self, b"contentlength", ContentLengthHeaderResource()) + put_child(self, b"nocontenttype", EmptyContentTypeHeaderResource()) + put_child(self, b"largechunkedfile", LargeChunkedFileResource()) + put_child(self, b"compress", Compress()) + put_child(self, b"duplicate-header", DuplicateHeaderResource()) + put_child(self, b"response-headers", ResponseHeadersResource()) + put_child(self, b"set-cookie", SetCookie()) + put_child(self, b"uri", UriResource()) - def getChild(self, path, request): + def getChild(self, path: bytes, request: Request) -> Root: return self - def render(self, request): + def render(self, request: Request) -> bytes: return b"Scrapy mock HTTP server\n" diff --git a/tests/mockserver/http_base.py b/tests/mockserver/http_base.py index 343c79781..5bc1252d6 100644 --- a/tests/mockserver/http_base.py +++ b/tests/mockserver/http_base.py @@ -17,6 +17,7 @@ from .utils import ssl_context_factory if TYPE_CHECKING: from collections.abc import Callable + from types import TracebackType from twisted.web import resource @@ -60,7 +61,12 @@ class BaseMockServer(ABC): self.https_port = https_parsed.port return self - def __exit__(self, exc_type, exc_value, traceback) -> None: + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: if self.proc: self.proc.kill() self.proc.communicate() @@ -135,7 +141,7 @@ def main_factory( context_factory = ssl_context_factory(**context_factory_kw) https_port = reactor.listenSSL(0, factory, context_factory) - def print_listening(): + def print_listening() -> None: if listen_http: http_host = http_port.getHost() http_address = f"http://{http_host.host}:{http_host.port}" diff --git a/tests/mockserver/http_resources.py b/tests/mockserver/http_resources.py index 98ac6cf6a..cb028bc10 100644 --- a/tests/mockserver/http_resources.py +++ b/tests/mockserver/http_resources.py @@ -3,7 +3,7 @@ from __future__ import annotations import gzip import json import random -from typing import TYPE_CHECKING, ParamSpec, TypeVar +from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar from urllib.parse import urlencode from twisted.internet.task import deferLater @@ -14,17 +14,24 @@ from twisted.web.util import Redirect, redirectTo from scrapy.utils.python import to_bytes, to_unicode if TYPE_CHECKING: - from collections.abc import Callable + from collections.abc import Callable, Sequence from twisted.internet.defer import Deferred - from twisted.web.http import Request + from twisted.python.failure import Failure + from twisted.web.http import Request as HTTPRequest + from twisted.web.server import Request _T = TypeVar("_T") _P = ParamSpec("_P") -def getarg(request, name, default=None, type_=None): +def getarg( + request: Request, + name: bytes, + default: Any = None, + type_: Callable[[bytes], Any] | None = None, +) -> Any: if name in request.args: value = request.args[name][0] if type_ is not None: @@ -33,73 +40,91 @@ def getarg(request, name, default=None, type_=None): return default -def close_connection(request): +def close_connection(request: Request) -> None: # We have to force a disconnection for HTTP/1.1 clients. Otherwise # client keeps the connection open waiting for more data. request.channel.loseConnection() request.finish() +def put_child(parent: resource.Resource, path: bytes, child: resource.Resource) -> None: + # zope.interface has no type hints, so mypy cannot tell that Resource + # instances provide the IResource interface that putChild() expects. + parent.putChild(path, child) # type: ignore[arg-type] + + +class BaseResource(resource.Resource): + """Base class for mockserver resources, with type hints.""" + + # Only needed to give subclasses a typed __init__ to call. + def __init__(self) -> None: # pylint: disable=useless-parent-delegation + super().__init__() # type: ignore[no-untyped-call] + + # most of the following resources are copied from twisted.web.test.test_webclient -class ForeverTakingResource(resource.Resource): +class ForeverTakingResource(BaseResource): """ L{ForeverTakingResource} is a resource which never finishes responding to requests. """ - def __init__(self, write=False): - resource.Resource.__init__(self) + def __init__(self, write: bool = False): + super().__init__() self._write = write - def render(self, request): + def render(self, request: Request) -> int: if self._write: request.write(b"some bytes") return server.NOT_DONE_YET -class HostHeaderResource(resource.Resource): +class HostHeaderResource(BaseResource): """ A testing resource which renders itself as the value of the host header from the request. """ - def render(self, request): - return request.requestHeaders.getRawHeaders(b"host")[0] + def render(self, request: Request) -> bytes: + headers = request.requestHeaders.getRawHeaders(b"host") + assert headers + return headers[0] -class ClientIPResource(resource.Resource): +class ClientIPResource(BaseResource): """ A testing resource which renders itself as the request client IP address. """ - def render(self, request): + def render(self, request: Request) -> bytes: client_address = request.getClientAddress() if client_address is None or client_address.host is None: return b"" return to_bytes(client_address.host) -class PayloadResource(resource.Resource): +class PayloadResource(BaseResource): """ A testing resource which renders itself as the contents of the request body as long as the request body is 100 bytes long, otherwise which renders itself as C{"ERROR"}. """ - def render(self, request): - data = request.content.read() - contentLength = request.requestHeaders.getRawHeaders(b"content-length")[0] - if len(data) != 100 or int(contentLength) != 100: + def render(self, request: Request) -> bytes: + assert request.content + data: bytes = request.content.read() + content_length = request.requestHeaders.getRawHeaders(b"content-length") + assert content_length + if len(data) != 100 or int(content_length[0]) != 100: return b"ERROR" return data -class LeafResource(resource.Resource): +class LeafResource(BaseResource): isLeaf = True def deferRequest( self, - request: Request, + request: HTTPRequest, delay: float, f: Callable[_P, _T], *a: _P.args, @@ -107,7 +132,7 @@ class LeafResource(resource.Resource): ) -> Deferred[_T]: from twisted.internet import reactor - def _cancelrequest(_): + def _cancelrequest(_: Failure) -> None: # silence CancelledError d.addErrback(lambda _: None) d.cancel() @@ -118,12 +143,13 @@ class LeafResource(resource.Resource): class Follow(LeafResource): - def render(self, request): + def render(self, request: Request) -> int: total = getarg(request, b"total", 100, type_=int) show = getarg(request, b"show", 1, type_=int) order = getarg(request, b"order", b"desc") maxlatency = getarg(request, b"maxlatency", 0, type_=float) n = getarg(request, b"n", total, type_=int) + nlist: Sequence[int] if order == b"rand": nlist = [random.randint(1, total) for _ in range(show)] else: # order == "desc" @@ -133,7 +159,7 @@ class Follow(LeafResource): self.deferRequest(request, lag, self.renderRequest, request, nlist) return NOT_DONE_YET - def renderRequest(self, request, nlist): + def renderRequest(self, request: Request, nlist: Sequence[int]) -> None: s = """ """ args = request.args.copy() for nl in nlist: @@ -146,45 +172,47 @@ class Follow(LeafResource): class Delay(LeafResource): - def render_GET(self, request): + def render_GET(self, request: Request) -> int: n = getarg(request, b"n", 1, type_=float) b = getarg(request, b"b", 1, type_=int) if b: # send headers now and delay body - request.write("") + request.write(b"") self.deferRequest(request, n, self._delayedRender, request, n) return NOT_DONE_YET - def _delayedRender(self, request, n): + def _delayedRender(self, request: Request, n: float) -> None: request.write(to_bytes(f"Response delayed for {n:.3f} seconds\n")) request.finish() class Status(LeafResource): - def render_GET(self, request): + def render_GET(self, request: Request) -> bytes: n = getarg(request, b"n", 200, type_=int) request.setResponseCode(n) return b"" class Raw(LeafResource): - def render_GET(self, request): + def render_GET(self, request: Request) -> int: request.startedWriting = 1 self.deferRequest(request, 0, self._delayedRender, request) return NOT_DONE_YET render_POST = render_GET - def _delayedRender(self, request): + def _delayedRender(self, request: Request) -> None: raw = getarg(request, b"raw", b"HTTP 1.1 200 OK\n") request.startedWriting = 1 request.write(raw) + assert request.channel.transport is not None request.channel.transport.loseConnection() request.finish() class Echo(LeafResource): - def render_GET(self, request): + def render_GET(self, request: Request) -> bytes: + assert request.content output = { "headers": { to_unicode(k): [to_unicode(v) for v in vs] @@ -198,27 +226,29 @@ class Echo(LeafResource): class RedirectTo(LeafResource): - def render(self, request): + def render(self, request: Request) -> bytes: goto = getarg(request, b"goto", b"/") # we force the body content, otherwise Twisted redirectTo() # returns HTML with int: request.setHeader(b"Content-Length", b"1024") self.deferRequest(request, 0, self._delayedRender, request) return NOT_DONE_YET - def _delayedRender(self, request): + def _delayedRender(self, request: Request) -> None: request.write(b"partial content\n") request.finish() class Drop(Partial): - def _delayedRender(self, request): + def _delayedRender(self, request: Request) -> None: abort = getarg(request, b"abort", 0, type_=int) request.write(b"this connection will be dropped\n") tr = request.channel.transport @@ -233,8 +263,10 @@ class Drop(Partial): class ArbitraryLengthPayloadResource(LeafResource): - def render(self, request): - return request.content.read() + def render(self, request: Request) -> bytes: + assert request.content + data: bytes = request.content.read() + return data class NoMetaRefreshRedirect(Redirect): @@ -245,21 +277,23 @@ class NoMetaRefreshRedirect(Redirect): ) -class ContentLengthHeaderResource(resource.Resource): +class ContentLengthHeaderResource(BaseResource): """ A testing resource which renders itself as the value of the Content-Length header from the request. """ - def render(self, request): - return request.requestHeaders.getRawHeaders(b"content-length")[0] + def render(self, request: Request) -> bytes: + headers = request.requestHeaders.getRawHeaders(b"content-length") + assert headers + return headers[0] -class ChunkedResource(resource.Resource): - def render(self, request): +class ChunkedResource(BaseResource): + def render(self, request: Request) -> int: from twisted.internet import reactor - def response(): + def response() -> None: request.write(b"chunked ") request.write(b"content\n") request.finish() @@ -268,11 +302,11 @@ class ChunkedResource(resource.Resource): return server.NOT_DONE_YET -class BrokenChunkedResource(resource.Resource): - def render(self, request): +class BrokenChunkedResource(BaseResource): + def render(self, request: Request) -> int: from twisted.internet import reactor - def response(): + def response() -> None: request.write(b"chunked ") request.write(b"content\n") # Disable terminating chunk on finish. @@ -283,11 +317,11 @@ class BrokenChunkedResource(resource.Resource): return server.NOT_DONE_YET -class BrokenDownloadResource(resource.Resource): - def render(self, request): +class BrokenDownloadResource(BaseResource): + def render(self, request: Request) -> int: from twisted.internet import reactor - def response(): + def response() -> None: request.setHeader(b"Content-Length", b"20") request.write(b"partial") close_connection(request) @@ -296,22 +330,24 @@ class BrokenDownloadResource(resource.Resource): return server.NOT_DONE_YET -class EmptyContentTypeHeaderResource(resource.Resource): +class EmptyContentTypeHeaderResource(BaseResource): """ A testing resource which renders itself as the value of request body without content-type header in response. """ - def render(self, request): + def render(self, request: Request) -> bytes: + assert request.content request.setHeader("content-type", "") - return request.content.read() + data: bytes = request.content.read() + return data -class LargeChunkedFileResource(resource.Resource): - def render(self, request): +class LargeChunkedFileResource(BaseResource): + def render(self, request: Request) -> int: from twisted.internet import reactor - def response(): + def response() -> None: for _ in range(1024): request.write(b"x" * 1024) request.finish() @@ -320,43 +356,45 @@ class LargeChunkedFileResource(resource.Resource): return server.NOT_DONE_YET -class DuplicateHeaderResource(resource.Resource): - def render(self, request): +class DuplicateHeaderResource(BaseResource): + def render(self, request: Request) -> bytes: request.responseHeaders.setRawHeaders(b"Set-Cookie", [b"a=b", b"c=d"]) return b"" -class UriResource(resource.Resource): +class UriResource(BaseResource): """Return the full uri that was requested""" - def getChild(self, path, request): + def getChild(self, path: bytes, request: Request) -> resource.Resource: return self - def render(self, request): + def render(self, request: Request) -> bytes | int: # Note: this is an ugly hack for CONNECT request timeout test. # Returning some data here fail SSL/TLS handshake # ToDo: implement proper HTTPS proxy tests, not faking them. if request.method != b"CONNECT": return request.uri + assert request.transport is not None request.transport.write(b"HTTP/1.1 200 Connection established\r\n\r\n") return NOT_DONE_YET -class ResponseHeadersResource(resource.Resource): +class ResponseHeadersResource(BaseResource): """Return a response with headers set from the JSON request body""" - def render(self, request): + def render(self, request: Request) -> bytes: + assert request.content body = json.loads(request.content.read().decode()) for header_name, header_value in body.items(): request.responseHeaders.setRawHeaders(header_name, [header_value]) return json.dumps(body).encode("utf-8") -class Compress(resource.Resource): +class Compress(BaseResource): """Compress the data sent in the request url params and set Content-Encoding header""" - def render(self, request): - data = request.args.get(b"data")[0] + def render(self, request: Request) -> bytes: + data = request.args[b"data"][0] accept_encoding_header = request.getHeader(b"accept-encoding") @@ -370,10 +408,10 @@ class Compress(resource.Resource): return b"Did not receive a valid accept-encoding header" -class SetCookie(resource.Resource): +class SetCookie(BaseResource): """Return a response with a Set-Cookie header for each request url parameter""" - def render(self, request): + def render(self, request: Request) -> bytes: for cookie_name, cookie_values in request.args.items(): for cookie_value in cookie_values: cookie = (cookie_name.decode() + "=" + cookie_value.decode()).encode() diff --git a/tests/mockserver/simple_https.py b/tests/mockserver/simple_https.py index fdea666e1..2a6cb6dd8 100644 --- a/tests/mockserver/simple_https.py +++ b/tests/mockserver/simple_https.py @@ -2,18 +2,23 @@ from __future__ import annotations -from twisted.web import resource +from typing import TYPE_CHECKING + from twisted.web.static import Data from .http_base import BaseMockServer, main_factory +from .http_resources import BaseResource, put_child + +if TYPE_CHECKING: + from twisted.web.server import Request -class Root(resource.Resource): - def __init__(self): - resource.Resource.__init__(self) - self.putChild(b"file", Data(b"0123456789", "text/plain")) +class Root(BaseResource): + def __init__(self) -> None: + super().__init__() + put_child(self, b"file", Data(b"0123456789", "text/plain")) - def getChild(self, path, request): + def getChild(self, path: bytes, request: Request) -> Root: return self @@ -29,7 +34,7 @@ class SimpleMockServer(BaseMockServer): cipher_string: str | None = None, tls_min_version: str | None = None, tls_max_version: str | None = None, - ): + ) -> None: super().__init__() self.keyfile = keyfile self.certfile = certfile diff --git a/tests/test_core_downloader.py b/tests/test_core_downloader.py index a06f30f15..7da4ad4f2 100644 --- a/tests/test_core_downloader.py +++ b/tests/test_core_downloader.py @@ -16,7 +16,7 @@ from twisted.web import server, static from twisted.web.client import Agent, BrowserLikePolicyForHTTPS, readBody from twisted.web.client import Response as TxResponse -from scrapy import Request +from scrapy import Request, Spider from scrapy.core.downloader import Downloader, Slot, tls from scrapy.core.downloader.contextfactory import ( _load_context_factory_from_settings, @@ -33,14 +33,17 @@ from scrapy.utils.misc import build_from_crawler from scrapy.utils.python import to_bytes from scrapy.utils.spider import DefaultSpider from scrapy.utils.test import get_crawler -from tests.mockserver.http_resources import PayloadResource +from tests.mockserver.http_resources import PayloadResource, put_child from tests.mockserver.utils import ssl_context_factory from tests.utils.decorators import coroutine_test if TYPE_CHECKING: + from twisted.internet.interfaces import IListeningPort from twisted.python.failure import Failure from twisted.web.iweb import IBodyProducer + from scrapy.http import Response + class TestSlot: def test_repr(self): @@ -54,7 +57,7 @@ class TestContextFactoryBase: async def server_url(self, tmp_path): (tmp_path / "file").write_bytes(b"0123456789") r = static.File(str(tmp_path)) - r.putChild(b"payload", PayloadResource()) + put_child(r, b"payload", PayloadResource()) site = server.Site(r, timeout=None) port = self._listen(site) portno = port.getHost().port @@ -63,7 +66,7 @@ class TestContextFactoryBase: await port.stopListening() - def _listen(self, site): + def _listen(self, site: server.Site) -> IListeningPort: from twisted.internet import reactor return reactor.listenSSL( @@ -319,7 +322,10 @@ def test_needs_backout(concurrency: int, active: int, expected: bool) -> None: @coroutine_test async def test_fetch_deprecated_spider_arg(): class CustomDownloader(Downloader): - def fetch(self, request, spider): # pylint: disable=signature-differs + # requiring the spider argument is what triggers the deprecation + def fetch( # type: ignore[override] # pylint: disable=signature-differs + self, request: Request, spider: Spider + ) -> Deferred[Response | Request]: return super().fetch(request, spider) crawler = get_crawler(DefaultSpider, {"DOWNLOADER": CustomDownloader}) diff --git a/tests/test_crawler.py b/tests/test_crawler.py index 9763eb82e..e7b48acae 100644 --- a/tests/test_crawler.py +++ b/tests/test_crawler.py @@ -74,6 +74,39 @@ class TestCrawler: assert not settings.frozen assert crawler.settings.frozen + @pytest.mark.parametrize( + ("attr", "setting"), + [ + ("download_delay", "DOWNLOAD_DELAY"), + ("max_concurrent_requests", "CONCURRENT_REQUESTS_PER_DOMAIN"), + ], + ) + def test_deprecated_spider_attr(self, attr: str, setting: str) -> None: + crawler = get_raw_crawler(type("_Spider", (DefaultSpider,), {attr: 2})) + with pytest.warns( + ScrapyDeprecationWarning, + match=f"The {attr!r} spider attribute is deprecated. Use the {setting} ", + ): + crawler._apply_settings() + assert crawler.settings.getint(setting) == 2 + + @pytest.mark.parametrize( + ("attr", "setting"), + [ + ("download_delay", "DOWNLOAD_DELAY"), + ("max_concurrent_requests", "CONCURRENT_REQUESTS_PER_DOMAIN"), + ], + ) + def test_deprecated_spider_attr_ignored(self, attr: str, setting: str) -> None: + crawler = get_raw_crawler(type("_Spider", (DefaultSpider,), {attr: 2})) + crawler.settings.set(setting, 3, priority="spider") + with pytest.warns( + ScrapyDeprecationWarning, + match=f"The {attr!r} spider attribute is deprecated. It is also being ", + ): + crawler._apply_settings() + assert crawler.settings.getint(setting) == 3 + def test_crawler_accepts_dict(self) -> None: crawler = get_crawler(DefaultSpider, {"foo": "bar"}) assert crawler.settings["foo"] == "bar" diff --git a/tests/test_downloader_handler_twisted_ftp.py b/tests/test_downloader_handler_twisted_ftp.py index 489b70e74..14de97b21 100644 --- a/tests/test_downloader_handler_twisted_ftp.py +++ b/tests/test_downloader_handler_twisted_ftp.py @@ -156,14 +156,17 @@ class TestFTP(TestFTPBase): for filename, content in self.test_files: (userdir / filename).write_bytes(content) - def _get_factory(self, root): + def _get_factory(self, root: Path) -> FTPFactory: from twisted.protocols.ftp import FTPFactory, FTPRealm realm = FTPRealm(anonymousRoot=str(root), userHome=str(root)) - p = portal.Portal(realm) + # zope.interface has no type hints, so mypy cannot tell that these + # objects provide the interfaces that Portal expects. + p = portal.Portal(realm) # type: ignore[arg-type] users_checker = checkers.InMemoryUsernamePasswordDatabaseDontUse() - users_checker.addUser(self.username, self.password) - p.registerChecker(users_checker, credentials.IUsernamePassword) + # the FTP protocol authenticates with str credentials + users_checker.addUser(self.username, self.password) # type: ignore[arg-type] + p.registerChecker(users_checker, credentials.IUsernamePassword) # type: ignore[arg-type] return FTPFactory(portal=p) @deferred_f_from_coro_f @@ -192,12 +195,17 @@ class TestAnonymousFTP(TestFTPBase): for filename, content in self.test_files: (root / filename).write_bytes(content) - def _get_factory(self, tmp_path): + def _get_factory(self, tmp_path: Path) -> FTPFactory: from twisted.protocols.ftp import FTPFactory, FTPRealm realm = FTPRealm(anonymousRoot=str(tmp_path)) - p = portal.Portal(realm) - p.registerChecker(checkers.AllowAnonymousAccess(), credentials.IAnonymous) + # zope.interface has no type hints, so mypy cannot tell that these + # objects provide the interfaces that Portal expects. + p = portal.Portal(realm) # type: ignore[arg-type] + p.registerChecker( + checkers.AllowAnonymousAccess(), # type: ignore[arg-type] + credentials.IAnonymous, + ) return FTPFactory(portal=p, userAnonymous=self.username) diff --git a/tests/test_engine_download.py b/tests/test_engine_download.py index 962808d96..09b998f6a 100644 --- a/tests/test_engine_download.py +++ b/tests/test_engine_download.py @@ -116,6 +116,18 @@ class TestEngineDownloadAsync: engine._slot.add_request.assert_called_once_with(request) engine._slot.remove_request.assert_called_once_with(request) + @coroutine_test + async def test_download_async_fetch_needs_spider(self, engine): + engine._downloader_fetch_needs_spider = True + request = Request("http://example.com") + response = Response("http://example.com", body=b"test body") + engine.spider = Mock() + engine.downloader.fetch.return_value = defer.succeed(response) + + result = await self._download(engine, request) + assert result == response + engine.downloader.fetch.assert_called_once_with(request, engine.spider) + @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") class TestEngineDownload(TestEngineDownloadAsync): diff --git a/tests/test_extension_throttle.py b/tests/test_extension_throttle.py index 4874f284a..2c718d95f 100644 --- a/tests/test_extension_throttle.py +++ b/tests/test_extension_throttle.py @@ -3,7 +3,7 @@ from unittest.mock import Mock import pytest -from scrapy import Request, Spider +from scrapy import Request from scrapy.exceptions import NotConfigured from scrapy.extensions.throttle import AutoThrottle from scrapy.http.response import Response @@ -25,6 +25,13 @@ def get_crawler(settings=None, spidercls=None): return _get_crawler(settings_dict=settings, spidercls=spidercls) +def _mock_downloader(crawler): + """Give *crawler* a mock engine, whose downloader AutoThrottle reads.""" + crawler.engine = Mock() + crawler.engine.downloader.slots = {} + return crawler.engine.downloader + + @pytest.mark.parametrize( ("value", "expected"), [ @@ -60,29 +67,21 @@ def test_target_concurrency_invalid(value): @pytest.mark.parametrize( - ("spider", "setting", "expected"), + ("setting", "expected"), [ - (UNSET, UNSET, DOWNLOAD_DELAY), - (1.0, UNSET, 1.0), - (UNSET, 1.0, 1.0), - (1.0, 2.0, 1.0), - (3.0, 2.0, 3.0), + (UNSET, DOWNLOAD_DELAY), + (1.0, 1.0), ], ) -def test_mindelay_definition(spider, setting, expected): +def test_mindelay_definition(setting, expected): settings = {} if setting is not UNSET: settings["DOWNLOAD_DELAY"] = setting - class _TestSpider(Spider): - name = "test" - - if spider is not UNSET: - _TestSpider.download_delay = spider - - crawler = get_crawler(settings, _TestSpider) + crawler = get_crawler(settings) at = build_from_crawler(AutoThrottle, crawler) - at._spider_opened(_TestSpider()) + _mock_downloader(crawler) + at._spider_opened(DefaultSpider()) assert at.mindelay == expected @@ -99,58 +98,43 @@ def test_maxdelay_definition(value, expected): settings["AUTOTHROTTLE_MAX_DELAY"] = value crawler = get_crawler(settings) at = build_from_crawler(AutoThrottle, crawler) + _mock_downloader(crawler) at._spider_opened(DefaultSpider()) assert at.maxdelay == expected @pytest.mark.parametrize( - ("min_spider", "min_setting", "start_setting", "expected"), + ("min_setting", "start_setting", "expected"), [ - (UNSET, UNSET, UNSET, AUTOTHROTTLE_START_DELAY), - (AUTOTHROTTLE_START_DELAY - 1.0, UNSET, UNSET, AUTOTHROTTLE_START_DELAY), - (AUTOTHROTTLE_START_DELAY + 1.0, UNSET, UNSET, AUTOTHROTTLE_START_DELAY + 1.0), - (UNSET, AUTOTHROTTLE_START_DELAY - 1.0, UNSET, AUTOTHROTTLE_START_DELAY), - (UNSET, AUTOTHROTTLE_START_DELAY + 1.0, UNSET, AUTOTHROTTLE_START_DELAY + 1.0), - (UNSET, UNSET, AUTOTHROTTLE_START_DELAY - 1.0, AUTOTHROTTLE_START_DELAY - 1.0), - (UNSET, UNSET, AUTOTHROTTLE_START_DELAY + 1.0, AUTOTHROTTLE_START_DELAY + 1.0), - ( - AUTOTHROTTLE_START_DELAY + 1.0, - AUTOTHROTTLE_START_DELAY + 2.0, - UNSET, - AUTOTHROTTLE_START_DELAY + 1.0, - ), + (UNSET, UNSET, AUTOTHROTTLE_START_DELAY), + (AUTOTHROTTLE_START_DELAY - 1.0, UNSET, AUTOTHROTTLE_START_DELAY), + (AUTOTHROTTLE_START_DELAY + 1.0, UNSET, AUTOTHROTTLE_START_DELAY + 1.0), + (UNSET, AUTOTHROTTLE_START_DELAY - 1.0, AUTOTHROTTLE_START_DELAY - 1.0), + (UNSET, AUTOTHROTTLE_START_DELAY + 1.0, AUTOTHROTTLE_START_DELAY + 1.0), ( AUTOTHROTTLE_START_DELAY + 2.0, - UNSET, AUTOTHROTTLE_START_DELAY + 1.0, AUTOTHROTTLE_START_DELAY + 2.0, ), ( AUTOTHROTTLE_START_DELAY + 1.0, - UNSET, AUTOTHROTTLE_START_DELAY + 2.0, AUTOTHROTTLE_START_DELAY + 2.0, ), ], ) -def test_startdelay_definition(min_spider, min_setting, start_setting, expected): +def test_startdelay_definition(min_setting, start_setting, expected): settings = {} if min_setting is not UNSET: settings["DOWNLOAD_DELAY"] = min_setting if start_setting is not UNSET: settings["AUTOTHROTTLE_START_DELAY"] = start_setting - class _TestSpider(Spider): - name = "test" - - if min_spider is not UNSET: - _TestSpider.download_delay = min_spider - - crawler = get_crawler(settings, _TestSpider) + crawler = get_crawler(settings) at = build_from_crawler(AutoThrottle, crawler) - spider = _TestSpider() - at._spider_opened(spider) - assert spider.download_delay == expected + downloader = _mock_downloader(crawler) + at._spider_opened(DefaultSpider()) + assert downloader._delay == expected @pytest.mark.parametrize( @@ -174,15 +158,13 @@ def test_startdelay_definition(min_spider, min_setting, start_setting, expected) def test_skipped(meta, slot): crawler = get_crawler() at = build_from_crawler(AutoThrottle, crawler) + downloader = _mock_downloader(crawler) spider = DefaultSpider() at._spider_opened(spider) request = Request("https://example.com", meta=meta) - crawler.engine = Mock() - crawler.engine.downloader = Mock() - crawler.engine.downloader.slots = {} if slot is not None: - crawler.engine.downloader.slots[slot] = object() + downloader.slots[slot] = object() at._adjust_delay = None # Raise exception if called. at._response_downloaded(None, request, spider) @@ -204,18 +186,16 @@ def test_adjustment(download_latency, target_concurrency, slot_delay, expected): settings = {"AUTOTHROTTLE_TARGET_CONCURRENCY": target_concurrency} crawler = get_crawler(settings) at = build_from_crawler(AutoThrottle, crawler) + downloader = _mock_downloader(crawler) spider = DefaultSpider() at._spider_opened(spider) meta = {"download_latency": download_latency, "download_slot": "foo"} request = Request("https://example.com", meta=meta) response = Response(request.url) - crawler.engine = Mock() - crawler.engine.downloader = Mock() - crawler.engine.downloader.slots = {} slot = Mock() slot.delay = slot_delay - crawler.engine.downloader.slots["foo"] = slot + downloader.slots["foo"] = slot at._response_downloaded(response, request, spider) @@ -240,18 +220,16 @@ def test_adjustment_limits(mindelay, maxdelay, expected): } crawler = get_crawler(settings) at = build_from_crawler(AutoThrottle, crawler) + downloader = _mock_downloader(crawler) spider = DefaultSpider() at._spider_opened(spider) meta = {"download_latency": download_latency, "download_slot": "foo"} request = Request("https://example.com", meta=meta) response = Response(request.url) - crawler.engine = Mock() - crawler.engine.downloader = Mock() - crawler.engine.downloader.slots = {} slot = Mock() slot.delay = slot_delay - crawler.engine.downloader.slots["foo"] = slot + downloader.slots["foo"] = slot at._response_downloaded(response, request, spider) @@ -272,18 +250,16 @@ def test_adjustment_bad_response( settings = {"AUTOTHROTTLE_TARGET_CONCURRENCY": target_concurrency} crawler = get_crawler(settings) at = build_from_crawler(AutoThrottle, crawler) + downloader = _mock_downloader(crawler) spider = DefaultSpider() at._spider_opened(spider) meta = {"download_latency": download_latency, "download_slot": "foo"} request = Request("https://example.com", meta=meta) response = Response(request.url, status=400) - crawler.engine = Mock() - crawler.engine.downloader = Mock() - crawler.engine.downloader.slots = {} slot = Mock() slot.delay = slot_delay - crawler.engine.downloader.slots["foo"] = slot + downloader.slots["foo"] = slot at._response_downloaded(response, request, spider) @@ -294,19 +270,17 @@ def test_debug(caplog): settings = {"AUTOTHROTTLE_DEBUG": True} crawler = get_crawler(settings) at = build_from_crawler(AutoThrottle, crawler) + downloader = _mock_downloader(crawler) spider = DefaultSpider() at._spider_opened(spider) meta = {"download_latency": 1.0, "download_slot": "foo"} request = Request("https://example.com", meta=meta) response = Response(request.url, body=b"foo") - crawler.engine = Mock() - crawler.engine.downloader = Mock() - crawler.engine.downloader.slots = {} slot = Mock() slot.delay = 2.0 slot.transferring = (None, None) - crawler.engine.downloader.slots["foo"] = slot + downloader.slots["foo"] = slot caplog.clear() with caplog.at_level(INFO): @@ -324,19 +298,17 @@ def test_debug(caplog): def test_debug_disabled(caplog): crawler = get_crawler() at = build_from_crawler(AutoThrottle, crawler) + downloader = _mock_downloader(crawler) spider = DefaultSpider() at._spider_opened(spider) meta = {"download_latency": 1.0, "download_slot": "foo"} request = Request("https://example.com", meta=meta) response = Response(request.url, body=b"foo") - crawler.engine = Mock() - crawler.engine.downloader = Mock() - crawler.engine.downloader.slots = {} slot = Mock() slot.delay = 2.0 slot.transferring = (None, None) - crawler.engine.downloader.slots["foo"] = slot + downloader.slots["foo"] = slot caplog.clear() with caplog.at_level(INFO): diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py index 3c1347fd3..b8586d1ca 100644 --- a/tests/test_http2_client_protocol.py +++ b/tests/test_http2_client_protocol.py @@ -30,7 +30,7 @@ from scrapy.utils.defer import ( deferred_from_coro, maybe_deferred_to_future, ) -from tests.mockserver.http_resources import LeafResource, Status +from tests.mockserver.http_resources import LeafResource, Status, put_child from tests.mockserver.utils import ssl_context_factory if TYPE_CHECKING: @@ -199,18 +199,18 @@ class TestHttps2ClientProtocol: @pytest.fixture def site(self, tmp_path): r = File(str(tmp_path)) - r.putChild(b"get-data-html-small", GetDataHtmlSmall()) - r.putChild(b"get-data-html-large", GetDataHtmlLarge()) + put_child(r, b"get-data-html-small", GetDataHtmlSmall()) + put_child(r, b"get-data-html-large", GetDataHtmlLarge()) - r.putChild(b"post-data-json-small", PostDataJsonSmall()) - r.putChild(b"post-data-json-large", PostDataJsonLarge()) + put_child(r, b"post-data-json-small", PostDataJsonSmall()) + put_child(r, b"post-data-json-large", PostDataJsonLarge()) - r.putChild(b"dataloss", Dataloss()) - r.putChild(b"no-content-length-header", NoContentLengthHeader()) - r.putChild(b"status", Status()) - r.putChild(b"query-params", QueryParams()) - r.putChild(b"timeout", TimeoutResponse()) - r.putChild(b"request-headers", RequestHeaders()) + put_child(r, b"dataloss", Dataloss()) + put_child(r, b"no-content-length-header", NoContentLengthHeader()) + put_child(r, b"status", Status()) + put_child(r, b"query-params", QueryParams()) + put_child(r, b"timeout", TimeoutResponse()) + put_child(r, b"request-headers", RequestHeaders()) return Site(r, timeout=None) @async_yield_fixture # type: ignore[untyped-decorator] 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: