"""Base classes for HTTP download handler tests.""" from __future__ import annotations import gzip import json import logging import os import re import sys from abc import ABC, abstractmethod from contextlib import asynccontextmanager from http import HTTPStatus from ipaddress import IPv4Address from socket import gethostbyname from typing import TYPE_CHECKING, Any, ClassVar from urllib.parse import urlparse import pytest from cryptography.x509 import load_der_x509_certificate from twisted.internet.ssl import Certificate from twisted.python.failure import Failure from scrapy.exceptions import ( CannotResolveHostError, DownloadCancelledError, DownloadConnectionRefusedError, DownloadFailedError, DownloadTimeoutError, ResponseDataLossError, ScrapyDeprecationWarning, StopDownload, UnsupportedURLSchemeError, ) from scrapy.http import Headers, HtmlResponse, Request, Response, TextResponse from scrapy.utils._deps_compat import ( PYOPENSSL_X509_DEPRECATED, TWISTED_TLS_LIMITS_OFFBY1, ) from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future from scrapy.utils.misc import build_from_crawler from scrapy.utils.spider import DefaultSpider from scrapy.utils.test import get_crawler from tests import NON_EXISTING_RESOLVABLE from tests.mockserver.mitm_proxy import wrong_credentials from tests.mockserver.proxy_echo import ProxyEchoMockServer from tests.mockserver.simple_https import SimpleMockServer from tests.spiders import ( BytesReceivedCallbackSpider, BytesReceivedErrbackSpider, HeadersReceivedCallbackSpider, HeadersReceivedErrbackSpider, SimpleSpider, SingleRequestSpider, ) from tests.utils.decorators import coroutine_test if TYPE_CHECKING: from collections.abc import AsyncGenerator, Generator from scrapy.core.downloader.handlers import DownloadHandlerProtocol from tests.mockserver.http import MockServer class TestHttpBase(ABC): is_secure: bool = False http2: bool = False # whether the handler supports per-request bindaddress handler_supports_bindaddress_meta: bool = True # RFC 9113 ยง8.1.1 explicitly says that a Content-Length mismatch is a # stream error (of type PROTOCOL_ERROR) so the client will send # RST_STREAM. Some libraries do only this while e.g. h2 also closes the # connection (see handling of ProtocolError in # h2.connection.H2Connection.receive_data()), thus closing all streams that # were using it, and we handle this as a normal exception. handler_supports_http2_dataloss: bool = True # default headers added by the underlying library that cannot be suppressed always_present_req_headers: ClassVar[frozenset[str]] = frozenset() default_handler_settings: ClassVar[dict[str, Any]] = {} @property @abstractmethod def download_handler_cls(self) -> type[DownloadHandlerProtocol]: raise NotImplementedError @asynccontextmanager async def get_dh( self, settings_dict: dict[str, Any] | None = None ) -> AsyncGenerator[DownloadHandlerProtocol]: settings_dict = { **self.default_handler_settings, **(settings_dict or {}), } crawler = get_crawler(DefaultSpider, settings_dict) crawler.spider = crawler._create_spider() dh = build_from_crawler(self.download_handler_cls, crawler) try: yield dh finally: await dh.close() @coroutine_test async def test_unsupported_scheme(self) -> None: request = Request("unsupp://unsupported.scheme") async with self.get_dh() as download_handler: with pytest.raises(UnsupportedURLSchemeError): await download_handler.download_request(request) @coroutine_test async def test_download(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == b"Works" @coroutine_test async def test_download_head(self, mockserver: MockServer) -> None: request = Request( mockserver.url("/text", is_secure=self.is_secure), method="HEAD" ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == b"" @pytest.mark.parametrize( "http_status", [ pytest.param(http_status, id=f"status={http_status.value}") for http_status in HTTPStatus if http_status.value == 200 or http_status.value // 100 in (4, 5) ], ) @coroutine_test async def test_download_has_correct_http_status_code( self, mockserver: MockServer, http_status: HTTPStatus ) -> None: request = Request( mockserver.url(f"/status?n={http_status.value}", is_secure=self.is_secure) ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == http_status.value @coroutine_test async def test_server_receives_correct_request_headers( self, mockserver: MockServer ) -> None: request_headers = { # common request headers "Accept": "text/html", "Accept-Charset": "utf-8", "Accept-Datetime": "Thu, 31 May 2007 20:35:00 GMT", "Accept-Encoding": "gzip, deflate", # custom headers "X-Custom-Header": "Custom Value", } request = Request( mockserver.url("/echo", is_secure=self.is_secure), headers=request_headers, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == HTTPStatus.OK body = json.loads(response.body.decode("utf-8")) assert "headers" in body for header_name, header_value in request_headers.items(): assert header_name in body["headers"] assert body["headers"][header_name] == [header_value] @coroutine_test async def test_request_header_none(self, mockserver: MockServer) -> None: """Adding a header with None as the value should not send that header.""" request_headers = { "Cookie": None, "X-Custom-Header": None, } request = Request( mockserver.url("/echo", is_secure=self.is_secure), headers=request_headers, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == HTTPStatus.OK body = json.loads(response.body.decode("utf-8")) assert "headers" in body for header_name in request_headers: assert header_name not in body["headers"] @pytest.mark.parametrize( "request_headers", [ {"X-Custom-Header": ["foo", "bar"]}, [("X-Custom-Header", "foo"), ("X-Custom-Header", "bar")], ], ) @coroutine_test async def test_request_header_duplicate( self, mockserver: MockServer, request_headers: Any ) -> None: """All values for a header should be sent.""" request = Request( mockserver.url("/echo", is_secure=self.is_secure), headers=request_headers, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == HTTPStatus.OK body = json.loads(response.body.decode("utf-8")) assert "headers" in body assert body["headers"]["X-Custom-Header"] == ["foo", "bar"] @coroutine_test async def test_server_receives_no_extra_headers( self, mockserver: MockServer ) -> None: """Test that the handler doesn't add headers to the request.""" request = Request(mockserver.url("/echo", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == HTTPStatus.OK body = json.loads(response.body.decode("utf-8")) assert "headers" in body received_headers = set(body["headers"].keys()) allowed_headers = { "Connection", "Content-Length", "Host", } | self.always_present_req_headers extra_headers = received_headers - allowed_headers assert not extra_headers, body["headers"] @coroutine_test async def test_server_receives_correct_request_body( self, mockserver: MockServer ) -> None: request_body = { "message": "It works!", } request = Request( mockserver.url("/echo", is_secure=self.is_secure), body=json.dumps(request_body), ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == HTTPStatus.OK body = json.loads(response.body.decode("utf-8")) assert json.loads(body["body"]) == request_body @coroutine_test async def test_download_has_correct_response_headers( self, mockserver: MockServer ) -> None: # these headers will be set on the response in the resource and returned response_headers = { # common response headers "Access-Control-Allow-Origin": "*", "Allow": "Get, Head", "Age": "12", "Cache-Control": "max-age=3600", "Content-Encoding": "gzip", "Content-MD5": "Q2hlY2sgSW50ZWdyaXR5IQ==", "Content-Type": "text/html; charset=utf-8", "Date": "Tue, 15 Nov 1994 08:12:31 GMT", "Pragma": "no-cache", "Retry-After": "120", "Set-Cookie": "CookieName=CookieValue; Max-Age=3600; Version=1", "WWW-Authenticate": "Basic", # custom headers "X-Custom-Header": "Custom Header Value", } request = Request( mockserver.url("/response-headers", is_secure=self.is_secure), headers={"content-type": "application/json"}, body=json.dumps(response_headers), ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 for header_name, header_value in response_headers.items(): assert header_name in response.headers, ( f"Response was missing expected header {header_name}" ) assert response.headers.getlist(header_name) == [ header_value.encode(encoding="utf-8") ] @coroutine_test async def test_download_no_extra_response_headers( self, mockserver: MockServer ) -> None: """Test that the handler doesn't add headers to the response.""" request = Request( mockserver.url("/response-headers", is_secure=self.is_secure), headers={"content-type": "application/json"}, body=json.dumps({}), ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 received_headers = set(response.headers.keys()) allowed_headers = { b"Content-Length", b"Content-Type", b"Date", b"Server", } extra_headers = received_headers - allowed_headers assert not extra_headers, response.headers @coroutine_test async def test_redirect_status(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/redirect", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 302 assert response.headers["Location"] == b"/redirected" @coroutine_test async def test_redirect_status_head(self, mockserver: MockServer) -> None: request = Request( mockserver.url("/redirect", is_secure=self.is_secure), method="HEAD" ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 302 assert response.headers["Location"] == b"/redirected" @coroutine_test async def test_timeout_download_from_spider_nodata_rcvd( self, mockserver: MockServer, reactor_pytest: str ) -> None: if reactor_pytest == "asyncio" and sys.platform == "win32": # https://twistedmatrix.com/trac/ticket/10279 pytest.skip( "This test produces DirtyReactorAggregateError on Windows with asyncio" ) # client connects but no data is received meta = {"download_timeout": 0.5} request = Request(mockserver.url("/wait", is_secure=self.is_secure), meta=meta) async with self.get_dh() as download_handler: d = deferred_from_coro(download_handler.download_request(request)) with pytest.raises(DownloadTimeoutError): await maybe_deferred_to_future(d) @coroutine_test async def test_timeout_download_from_spider_server_hangs( self, mockserver: MockServer, reactor_pytest: str ) -> None: if reactor_pytest == "asyncio" and sys.platform == "win32": # https://twistedmatrix.com/trac/ticket/10279 pytest.skip( "This test produces DirtyReactorAggregateError on Windows with asyncio" ) # client connects, server send headers and some body bytes but hangs meta = {"download_timeout": 0.5} request = Request( mockserver.url("/hang-after-headers", is_secure=self.is_secure), meta=meta ) async with self.get_dh() as download_handler: d = deferred_from_coro(download_handler.download_request(request)) with pytest.raises(DownloadTimeoutError): await maybe_deferred_to_future(d) @pytest.mark.parametrize("send_header", [True, False]) @coroutine_test async def test_host_header(self, send_header: bool, mockserver: MockServer) -> None: host_port = f"{mockserver.host}:{mockserver.port(is_secure=self.is_secure)}" request = Request( mockserver.url("/host", is_secure=self.is_secure), headers={"Host": host_port} if send_header else {}, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == host_port.encode() if send_header: assert request.headers.get("Host") == host_port.encode() else: assert not request.headers @coroutine_test async def test_content_length_zero_bodyless_post_request_headers( self, mockserver: MockServer ) -> None: """Tests if "Content-Length: 0" is sent for bodyless POST requests. This is not strictly required by HTTP RFCs but can cause trouble for some web servers. See: https://github.com/scrapy/scrapy/issues/823 https://issues.apache.org/jira/browse/TS-2902 https://github.com/kennethreitz/requests/issues/405 https://bugs.python.org/issue14721 """ request = Request( mockserver.url("/contentlength", is_secure=self.is_secure), method="POST" ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == b"0" @coroutine_test async def test_content_length_zero_bodyless_post_only_one( self, mockserver: MockServer ) -> None: request = Request( mockserver.url("/echo", is_secure=self.is_secure), method="POST" ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) headers = Headers(json.loads(response.text)["headers"]) contentlengths = headers.getlist("Content-Length") assert len(contentlengths) == 1 assert contentlengths == [b"0"] @coroutine_test async def test_payload(self, mockserver: MockServer) -> None: body = b"1" * 100 # PayloadResource requires body length to be 100 request = Request( mockserver.url("/payload", is_secure=self.is_secure), method="POST", body=body, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == body @coroutine_test async def test_response_header_content_length(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.headers[b"content-length"] == b"5" @pytest.mark.parametrize( ("filename", "body", "response_class"), [ ("foo.html", b"", HtmlResponse), ("foo", b"\n.", HtmlResponse), ], ) @coroutine_test async def test_response_class( self, filename: str, body: bytes, response_class: type[Response], mockserver: MockServer, ) -> None: request = Request( mockserver.url(f"/{filename}", is_secure=self.is_secure), body=body ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert type(response) is response_class # pylint: disable=unidiomatic-typecheck @coroutine_test async def test_get_duplicate_header(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/duplicate-header", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.headers.getlist(b"Set-Cookie") == [b"a=b", b"c=d"] @coroutine_test async def test_download_is_not_automatically_gzip_decoded( self, mockserver: MockServer ) -> None: """Test download handler does not automatically decode content using the scheme provided in Content-Encoding header""" data = "compress-me" # send a request to mock resource that gzip encodes the "data" url parameter request = Request( mockserver.url(f"/compress?data={data}", is_secure=self.is_secure), headers={ "accept-encoding": "gzip", }, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 # check that the Content-Encoding header is gzip content_encoding = response.headers[b"Content-Encoding"] assert content_encoding == b"gzip" # check that the response is still encoded # by checking for the magic number that is always included at the start of a gzip encoding # see https://datatracker.ietf.org/doc/html/rfc1952#page-5 section 2.3.1 GZIP_MAGIC = b"\x1f\x8b" assert response.body[:2] == GZIP_MAGIC, "Response body was not in gzip format" # check that a gzip decoding matches the data sent in the request expected_decoding = bytes(data, encoding="utf-8") assert gzip.decompress(response.body) == expected_decoding @coroutine_test async def test_no_cookie_processing_or_persistence( self, mockserver: MockServer ) -> None: cookie_name = "foo" cookie_value = "bar" # check that cookies are not modified request = Request( mockserver.url( f"/set-cookie?{cookie_name}={cookie_value}", is_secure=self.is_secure ) ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 set_cookie = response.headers.get(b"Set-Cookie") assert set_cookie == f"{cookie_name}={cookie_value}".encode() # check that cookies are not sent in the next request request = Request(mockserver.url("/echo", is_secure=self.is_secure)) response = await download_handler.download_request(request) assert response.status == 200 headers = Headers(json.loads(response.text)["headers"]) assert "Cookie" not in headers assert "cookie" not in headers @coroutine_test async def test_download_latency(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh() as download_handler: await download_handler.download_request(request) assert "download_latency" in request.meta latency = request.meta["download_latency"] if sys.version_info < (3, 13) and sys.platform == "win32": # time.monotonic() resolution is too low here: # https://docs.python.org/3/whatsnew/3.13.html#time assert latency >= 0 else: assert latency > 0 @coroutine_test async def test_download_without_maxsize_limit(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == b"Works" @coroutine_test async def test_response_class_choosing_request( self, mockserver: MockServer ) -> None: """Tests choosing of correct response type in case of Content-Type is empty but body contains text. """ body = b"Some plain text\ndata with tabs\t and null bytes\0" request = Request( mockserver.url("/nocontenttype", is_secure=self.is_secure), body=body ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert type(response) is TextResponse # pylint: disable=unidiomatic-typecheck @coroutine_test async def test_download_with_maxsize( self, caplog: pytest.LogCaptureFixture, mockserver: MockServer ) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) # 5 is minimal size for this request and the limit is only counted on # response body. (regardless of headers) async with self.get_dh({"DOWNLOAD_MAXSIZE": 5}) as download_handler: response = await download_handler.download_request(request) assert response.body == b"Works" caplog.clear() msg = "Expected to receive 5 bytes which is larger than download max size (4)" async with self.get_dh({"DOWNLOAD_MAXSIZE": 4}) as download_handler: with pytest.raises(DownloadCancelledError, match=re.escape(msg)): await download_handler.download_request(request) assert msg in caplog.text @coroutine_test async def test_download_with_maxsize_very_large_file( self, mockserver: MockServer, caplog: pytest.LogCaptureFixture ) -> None: request = Request(mockserver.url("/largechunkedfile", is_secure=self.is_secure)) async with self.get_dh({"DOWNLOAD_MAXSIZE": 1_500}) as download_handler: with pytest.raises(DownloadCancelledError): await download_handler.download_request(request) assert re.search( r"Received \d+ bytes which is larger than download max size \(1500\)", caplog.text, ) @coroutine_test async def test_download_with_maxsize_per_req(self, mockserver: MockServer) -> None: meta = {"download_maxsize": 2} request = Request(mockserver.url("/text", is_secure=self.is_secure), meta=meta) async with self.get_dh() as download_handler: with pytest.raises(DownloadCancelledError): await download_handler.download_request(request) @coroutine_test async def test_download_with_small_maxsize_via_setting( self, mockserver: MockServer ) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh({"DOWNLOAD_MAXSIZE": 2}) as download_handler: with pytest.raises(DownloadCancelledError): await download_handler.download_request(request) @coroutine_test async def test_download_with_large_maxsize_via_setting( self, mockserver: MockServer ) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh({"DOWNLOAD_MAXSIZE": 100}) as download_handler: response = await download_handler.download_request(request) assert response.body == b"Works" @coroutine_test async def test_download_with_warnsize( self, caplog: pytest.LogCaptureFixture, mockserver: MockServer ) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh({"DOWNLOAD_WARNSIZE": 4}) as download_handler: response = await download_handler.download_request(request) assert response.body == b"Works" assert ( "Expected to receive 5 bytes which is larger than download warn size (4)" in caplog.text ) @coroutine_test async def test_download_with_warnsize_no_content_length( self, caplog: pytest.LogCaptureFixture, mockserver: MockServer ) -> None: request = Request( mockserver.url("/delay?n=0.1", is_secure=self.is_secure), ) async with self.get_dh({"DOWNLOAD_WARNSIZE": 10}) as download_handler: response = await download_handler.download_request(request) assert response.body == b"Response delayed for 0.100 seconds\n" assert ( "Received 35 bytes which is larger than download warn size (10)" in caplog.text ) @coroutine_test async def test_download_chunked_content(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/chunked", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == b"chunked content\n" @coroutine_test async def test_download_cause_data_loss(self, mockserver: MockServer) -> None: if self.http2 and not self.handler_supports_http2_dataloss: pytest.skip("This handler doesn't support dataloss on HTTP/2") request = Request(mockserver.url("/broken", is_secure=self.is_secure)) async with self.get_dh() as download_handler: with pytest.raises(ResponseDataLossError): await download_handler.download_request(request) @coroutine_test async def test_download_cause_data_loss_double_warning( self, caplog: pytest.LogCaptureFixture, mockserver: MockServer ) -> None: if self.http2 and not self.handler_supports_http2_dataloss: pytest.skip("This handler doesn't support dataloss on HTTP/2") request = Request(mockserver.url("/broken", is_secure=self.is_secure)) async with self.get_dh() as download_handler: with pytest.raises(ResponseDataLossError): await download_handler.download_request(request) assert "Got data loss" in caplog.text caplog.clear() with pytest.raises(ResponseDataLossError): await download_handler.download_request(request) # no repeated warning assert "Got data loss" not in caplog.text @coroutine_test async def test_download_allow_data_loss_broken( self, mockserver: MockServer ) -> None: if self.http2 and not self.handler_supports_http2_dataloss: pytest.skip("This handler doesn't support dataloss on HTTP/2") request = Request( mockserver.url("/broken", is_secure=self.is_secure), meta={"download_fail_on_dataloss": False}, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.flags == ["dataloss"] assert response.text == "partial" @coroutine_test async def test_download_allow_data_loss_broken_chunked( self, mockserver: MockServer ) -> None: if self.http2: pytest.skip("Chunked encoding is specific to HTTP/1.1") request = Request( mockserver.url("/broken-chunked", is_secure=self.is_secure), meta={"download_fail_on_dataloss": False}, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.flags == ["dataloss"] assert response.text == "chunked content\n" @coroutine_test async def test_download_allow_data_loss_via_setting( self, mockserver: MockServer ) -> None: if self.http2 and not self.handler_supports_http2_dataloss: pytest.skip("This handler doesn't support dataloss on HTTP/2") request = Request(mockserver.url("/broken", is_secure=self.is_secure)) async with self.get_dh( {"DOWNLOAD_FAIL_ON_DATALOSS": False} ) as download_handler: response = await download_handler.download_request(request) assert response.flags == ["dataloss"] @coroutine_test async def test_download_conn_failed(self) -> None: # copy of TestCrawl.test_retry_conn_failed() scheme = "https" if self.is_secure else "http" request = Request(f"{scheme}://localhost:65432/") async with self.get_dh() as download_handler: with pytest.raises(DownloadConnectionRefusedError): await download_handler.download_request(request) @coroutine_test async def test_download_conn_lost(self, mockserver: MockServer) -> None: # copy of TestCrawl.test_retry_conn_lost() request = Request(mockserver.url("/drop?abort=0", is_secure=self.is_secure)) async with self.get_dh() as download_handler: with pytest.raises(ResponseDataLossError): await download_handler.download_request(request) @coroutine_test async def test_download_conn_aborted(self, mockserver: MockServer) -> None: # copy of TestCrawl.test_retry_conn_aborted() if self.http2: # it may be possible to write a separate resource that does something # suitable on HTTP/2 without sending Content-Length pytest.skip( "On HTTP/2 this triggers a Content-Length mismatch error instead." ) request = Request(mockserver.url("/drop?abort=1", is_secure=self.is_secure)) async with self.get_dh() as download_handler: with pytest.raises(DownloadFailedError): await download_handler.download_request(request) @pytest.mark.skipif( NON_EXISTING_RESOLVABLE, reason="Non-existing hosts are resolvable" ) @coroutine_test async def test_download_dns_error(self) -> None: # copy of TestCrawl.test_retry_dns_error() scheme = "https" if self.is_secure else "http" request = Request(f"{scheme}://dns.resolution.invalid./") async with self.get_dh() as download_handler: with pytest.raises(CannotResolveHostError): await download_handler.download_request(request) @coroutine_test async def test_protocol(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/host", is_secure=self.is_secure)) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.protocol == "HTTP/1.1" @pytest.mark.skipif( sys.platform == "darwin", reason="127.0.0.2 is not available on macOS by default", ) @pytest.mark.parametrize("setting_value", [("127.0.0.2", 0), "127.0.0.2"]) @coroutine_test async def test_download_bind_address_setting( self, mockserver: MockServer, setting_value: Any ) -> None: request = Request(mockserver.url("/client-ip", is_secure=self.is_secure)) async with self.get_dh( {"DOWNLOAD_BIND_ADDRESS": setting_value} ) as download_handler: response = await download_handler.download_request(request) assert response.body == b"127.0.0.2" @pytest.mark.skipif( sys.platform == "darwin", reason="127.0.0.2 is not available on macOS by default", ) @pytest.mark.parametrize("meta_value", [("127.0.0.2", 0), "127.0.0.2"]) @coroutine_test async def test_download_bind_address_meta( self, mockserver: MockServer, caplog: pytest.LogCaptureFixture, meta_value: Any ) -> None: request = Request( mockserver.url("/client-ip", is_secure=self.is_secure), meta={"bindaddress": meta_value}, ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) if self.handler_supports_bindaddress_meta: assert response.body == b"127.0.0.2" else: assert ( "The 'bindaddress' request meta key is not supported by" in caplog.text ) @coroutine_test async def test_verbatim_url(self, mockserver: MockServer) -> None: # Square brackets are encoded by safe_url_string (w3lib). path = "/uri/items?data[0]=a" url = mockserver.url(path, is_secure=self.is_secure) # Without verbatim_url, the brackets are percent-encoded before the # request reaches the server. request = Request(url) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == b"/uri/items?data%5B0%5D=a" # With verbatim_url=True the URL is sent to the server as-is. request = Request(url, meta={"verbatim_url": True}) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == path.encode() class TestHttpsBase(TestHttpBase): is_secure = True tls_log_message = ( ( 'SSL connection certificate: issuer "CN=localhost,O=Scrapy,C=IE", ' 'subject "CN=localhost,O=Scrapy,C=IE"' ) if PYOPENSSL_X509_DEPRECATED else ( 'SSL connection certificate: issuer "/C=IE/O=Scrapy/CN=localhost", ' 'subject "/C=IE/O=Scrapy/CN=localhost"' ) ) def test_download_conn_lost(self) -> None: # type: ignore[override] # For some reason (maybe related to TLS shutdown flow, and maybe the # mockserver resource can be fixed so that this works) HTTPS clients # (not just Scrapy) hang on /drop?abort=0. pytest.skip("Unable to test on HTTPS") @coroutine_test async def test_tls_logging( self, mockserver: MockServer, caplog: pytest.LogCaptureFixture ) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh( {"DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING": True} ) as download_handler: with caplog.at_level("DEBUG"): response = await download_handler.download_request(request) assert response.body == b"Works" assert self.tls_log_message in caplog.text @coroutine_test async def test_verify_certs_deprecated(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) with ( # noqa: PT031 pytest.warns( ScrapyDeprecationWarning, match="'DOWNLOADER_CLIENTCONTEXTFACTORY' setting is deprecated", ), pytest.warns( ScrapyDeprecationWarning, match="BrowserLikeContextFactory is deprecated", ), ): async with self.get_dh( { "DOWNLOADER_CLIENTCONTEXTFACTORY": "scrapy.core.downloader.contextfactory.BrowserLikeContextFactory" } ) as download_handler: with pytest.raises( (DownloadConnectionRefusedError, DownloadFailedError) ): await download_handler.download_request(request) @coroutine_test async def test_verify_certs(self, mockserver: MockServer) -> None: request = Request(mockserver.url("/text", is_secure=self.is_secure)) async with self.get_dh( {"DOWNLOAD_VERIFY_CERTIFICATES": True} ) as download_handler: with pytest.raises((DownloadConnectionRefusedError, DownloadFailedError)): await download_handler.download_request(request) class TestSimpleHttpsBase(ABC): """Base class for special cases tested with just one simple request""" keyfile = "keys/localhost.key" certfile = "keys/localhost.crt" host = "localhost" cipher_string: str | None = None @pytest.fixture(scope="class") @classmethod def simple_mockserver(cls) -> Generator[SimpleMockServer]: with SimpleMockServer( cls.keyfile, cls.certfile, cipher_string=cls.cipher_string ) as simple_mockserver: yield simple_mockserver @pytest.fixture(scope="class") @classmethod def url(cls, simple_mockserver: SimpleMockServer) -> str: return f"https://{cls.host}:{simple_mockserver.port(is_secure=True)}/file" @property @abstractmethod def download_handler_cls(self) -> type[DownloadHandlerProtocol]: raise NotImplementedError @asynccontextmanager async def get_dh(self) -> AsyncGenerator[DownloadHandlerProtocol]: if self.cipher_string is not None: settings_dict = {"DOWNLOADER_CLIENT_TLS_CIPHERS": self.cipher_string} else: settings_dict = None crawler = get_crawler(DefaultSpider, settings_dict=settings_dict) crawler.spider = crawler._create_spider() dh = build_from_crawler(self.download_handler_cls, crawler) try: yield dh finally: await dh.close() @coroutine_test async def test_download(self, url: str) -> None: request = Request(url) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.body == b"0123456789" class TestHttpsWrongHostnameBase(TestSimpleHttpsBase): # above tests use a server certificate for "localhost", # client connection to "localhost" too. # here we test that even if the server certificate is for another domain, # "www.example.com" in this case, # the tests still pass keyfile = "keys/example-com.key.pem" certfile = "keys/example-com.cert.pem" class TestHttpsInvalidDNSIdBase(TestSimpleHttpsBase): """Connect to HTTPS hosts with IP while certificate uses domain names IDs.""" host = "127.0.0.1" class TestHttpsInvalidDNSPatternBase(TestSimpleHttpsBase): """Connect to HTTPS hosts where the certificate are issued to an ip instead of a domain.""" keyfile = "keys/localhost.ip.key" certfile = "keys/localhost.ip.crt" class TestHttpsCustomCiphersBase(TestSimpleHttpsBase): cipher_string = "CAMELLIA256-SHA" class TestHttpsTLSVersionBase(ABC): keyfile = "keys/localhost.key" certfile = "keys/localhost.crt" @property @abstractmethod def download_handler_cls(self) -> type[DownloadHandlerProtocol]: raise NotImplementedError @asynccontextmanager async def get_dh( self, client_tls_min: str | None, client_tls_max: str | None ) -> AsyncGenerator[DownloadHandlerProtocol]: settings = {} if client_tls_min is not None: settings["DOWNLOAD_TLS_MIN_VERSION"] = client_tls_min if client_tls_max is not None: settings["DOWNLOAD_TLS_MAX_VERSION"] = client_tls_max crawler = get_crawler(DefaultSpider, settings_dict=settings) crawler.spider = crawler._create_spider() dh = build_from_crawler(self.download_handler_cls, crawler) try: yield dh finally: await dh.close() @pytest.mark.parametrize( ( "server_tls_min", "server_tls_max", "client_tls_min", "client_tls_max", "expect_success", ), [ pytest.param(None, None, None, None, True, id="no-limits"), pytest.param(None, None, None, "TLSv1.2", True, id="client-max-tls1.2"), pytest.param(None, None, "TLSv1.3", None, True, id="client-min-tls1.3"), pytest.param( "TLSv1.3", None, None, "TLSv1.2", False, id="client-max-below-server-min", ), pytest.param( None, "TLSv1.2", "TLSv1.3", None, False, id="client-min-above-server-max", ), pytest.param(None, "TLSv1.2", None, "TLSv1.2", True, id="both-tls1.2"), pytest.param("TLSv1.3", None, "TLSv1.3", None, True, id="both-tls1.3"), pytest.param( None, None, "TLSv1.0", None, True, id="client-min-tls1.0", marks=pytest.mark.filterwarnings( r"ignore:ssl\.TLSVersion\.TLSv1 is deprecated:DeprecationWarning" ), ), pytest.param( "TLSv1.0", None, "TLSv1.0", None, True, id="both-min-tls1.0", marks=pytest.mark.filterwarnings( r"ignore:ssl\.TLSVersion\.TLSv1 is deprecated:DeprecationWarning" ), ), pytest.param( "TLSv1.2", "TLSv1.3", "TLSv1.2", "TLSv1.3", True, id="both-tls1.2-1.3", marks=pytest.mark.xfail( TWISTED_TLS_LIMITS_OFFBY1, reason="Can't set max to 1.3 on this Twisted version", strict=True, ), ), ], ) @coroutine_test async def test_download( self, server_tls_min: str | None, server_tls_max: str | None, client_tls_min: str | None, client_tls_max: str | None, expect_success: bool, ) -> None: with SimpleMockServer( self.keyfile, self.certfile, tls_min_version=server_tls_min, tls_max_version=server_tls_max, ) as simple_mockserver: url = f"https://localhost:{simple_mockserver.port(is_secure=True)}/file" request = Request(url) async with self.get_dh(client_tls_min, client_tls_max) as dh: if expect_success: response = await dh.download_request(request) assert response.body == b"0123456789" else: with pytest.raises( (DownloadConnectionRefusedError, DownloadFailedError) ): await dh.download_request(request) @coroutine_test async def test_invalid_min_version_setting(self) -> None: with pytest.raises( ValueError, match="Unknown DOWNLOAD_TLS_MIN_VERSION value: invalid" ): async with self.get_dh(client_tls_min="invalid", client_tls_max=None): pass @coroutine_test async def test_invalid_max_version_setting(self) -> None: with pytest.raises( ValueError, match="Unknown DOWNLOAD_TLS_MAX_VERSION value: invalid" ): async with self.get_dh(client_tls_min=None, client_tls_max="invalid"): pass class TestHttpWithCrawlerBase(ABC): @property @abstractmethod def settings_dict(self) -> dict[str, Any] | None: raise NotImplementedError is_secure = False @coroutine_test async def test_download_with_content_length(self, mockserver: MockServer) -> None: crawler = get_crawler(SingleRequestSpider, self.settings_dict) # http://localhost:8998/partial set Content-Length to 1024, use download_maxsize= 1000 to avoid # download it await maybe_deferred_to_future( crawler.crawl( seed=Request( url=mockserver.url("/partial", is_secure=self.is_secure), meta={"download_maxsize": 1000}, ) ) ) assert crawler.spider failure = crawler.spider.meta["failure"] # type: ignore[attr-defined] assert isinstance(failure.value, DownloadCancelledError) @coroutine_test async def test_download(self, mockserver: MockServer) -> None: crawler = get_crawler(SingleRequestSpider, self.settings_dict) await maybe_deferred_to_future( crawler.crawl( seed=Request(url=mockserver.url("", is_secure=self.is_secure)) ) ) assert crawler.spider failure = crawler.spider.meta.get("failure") # type: ignore[attr-defined] assert failure is None reason = crawler.spider.meta["close_reason"] # type: ignore[attr-defined] assert reason == "finished" @pytest.mark.filterwarnings( r"ignore:.*You should use cryptography's X\.509 APIs:DeprecationWarning" ) @coroutine_test async def test_response_ssl_certificate(self, mockserver: MockServer) -> None: if not self.is_secure: pytest.skip("Only applies to HTTPS") # copy of TestCrawl.test_response_ssl_certificate() crawler = get_crawler(SingleRequestSpider, self.settings_dict) url = mockserver.url("/echo?body=test", is_secure=self.is_secure) await crawler.crawl_async(seed=url, mockserver=mockserver) assert isinstance(crawler.spider, SingleRequestSpider) cert = crawler.spider.meta["responses"][0].certificate assert cert is not None if isinstance(cert, Certificate): # Twisted assert cert.getSubject().commonName == b"localhost" assert cert.getIssuer().commonName == b"localhost" elif isinstance(cert, bytes): # DER bytes cert_x509 = load_der_x509_certificate(cert) assert cert_x509.subject.rfc4514_string() == "CN=localhost,O=Scrapy,C=IE" assert cert_x509.issuer.rfc4514_string() == "CN=localhost,O=Scrapy,C=IE" @coroutine_test async def test_response_ip_address(self, mockserver: MockServer) -> None: # copy of TestCrawl.test_response_ip_address() crawler = get_crawler(SingleRequestSpider, self.settings_dict) url = mockserver.url("/echo?body=test", is_secure=self.is_secure) expected_netloc, _ = urlparse(url).netloc.split(":") await crawler.crawl_async(seed=url, mockserver=mockserver) assert isinstance(crawler.spider, SingleRequestSpider) ip_address = crawler.spider.meta["responses"][0].ip_address assert isinstance(ip_address, IPv4Address) assert str(ip_address) == gethostbyname(expected_netloc) @coroutine_test async def test_bytes_received_stop_download_callback( self, mockserver: MockServer ) -> None: # copy of TestCrawl.test_bytes_received_stop_download_callback() crawler = get_crawler(BytesReceivedCallbackSpider, self.settings_dict) await crawler.crawl_async(mockserver=mockserver, is_secure=self.is_secure) assert isinstance(crawler.spider, BytesReceivedCallbackSpider) assert crawler.spider.meta.get("failure") is None assert isinstance(crawler.spider.meta["response"], Response) assert crawler.spider.meta["response"].body == crawler.spider.meta.get( "bytes_received" ) assert ( len(crawler.spider.meta["response"].body) < crawler.spider.full_response_length ) @coroutine_test async def test_bytes_received_stop_download_errback( self, mockserver: MockServer ) -> None: # copy of TestCrawl.test_bytes_received_stop_download_errback() crawler = get_crawler(BytesReceivedErrbackSpider, self.settings_dict) await crawler.crawl_async(mockserver=mockserver, is_secure=self.is_secure) assert isinstance(crawler.spider, BytesReceivedErrbackSpider) assert crawler.spider.meta.get("response") is None assert isinstance(crawler.spider.meta["failure"], Failure) assert isinstance(crawler.spider.meta["failure"].value, StopDownload) assert isinstance(crawler.spider.meta["failure"].value.response, Response) assert crawler.spider.meta[ "failure" ].value.response.body == crawler.spider.meta.get("bytes_received") assert ( len(crawler.spider.meta["failure"].value.response.body) < crawler.spider.full_response_length ) @coroutine_test async def test_headers_received_stop_download_callback( self, mockserver: MockServer ) -> None: # copy of TestCrawl.test_headers_received_stop_download_callback() crawler = get_crawler(HeadersReceivedCallbackSpider, self.settings_dict) await crawler.crawl_async(mockserver=mockserver, is_secure=self.is_secure) assert isinstance(crawler.spider, HeadersReceivedCallbackSpider) assert crawler.spider.meta.get("failure") is None assert isinstance(crawler.spider.meta["response"], Response) assert crawler.spider.meta["response"].headers == crawler.spider.meta.get( "headers_received" ) @coroutine_test async def test_headers_received_stop_download_errback( self, mockserver: MockServer ) -> None: # copy of TestCrawl.test_headers_received_stop_download_errback() crawler = get_crawler(HeadersReceivedErrbackSpider, self.settings_dict) await crawler.crawl_async(mockserver=mockserver, is_secure=self.is_secure) assert isinstance(crawler.spider, HeadersReceivedErrbackSpider) assert crawler.spider.meta.get("response") is None assert isinstance(crawler.spider.meta["failure"], Failure) assert isinstance(crawler.spider.meta["failure"].value, StopDownload) assert isinstance(crawler.spider.meta["failure"].value.response, Response) assert crawler.spider.meta[ "failure" ].value.response.headers == crawler.spider.meta.get("headers_received") class TestHttpProxyBase(ABC): is_secure = False expected_http_proxy_request_body = b"http://example.com" # whether the handler supports HTTPS proxies with HTTPS destinations handler_supports_tls_in_tls: bool = True @property @abstractmethod def download_handler_cls(self) -> type[DownloadHandlerProtocol]: raise NotImplementedError @pytest.fixture(scope="session") def proxy_mockserver(self) -> Generator[ProxyEchoMockServer]: with ProxyEchoMockServer() as proxy: yield proxy @asynccontextmanager async def get_dh(self) -> AsyncGenerator[DownloadHandlerProtocol]: crawler = get_crawler(DefaultSpider) crawler.spider = crawler._create_spider() dh = build_from_crawler(self.download_handler_cls, crawler) try: yield dh finally: await dh.close() @coroutine_test async def test_download_with_proxy( self, proxy_mockserver: ProxyEchoMockServer ) -> None: http_proxy = proxy_mockserver.url("", is_secure=self.is_secure) request = Request("http://example.com", meta={"proxy": http_proxy}) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 assert response.url == request.url assert response.body == self.expected_http_proxy_request_body @coroutine_test async def test_download_without_proxy( self, proxy_mockserver: ProxyEchoMockServer ) -> None: request = Request( proxy_mockserver.url("/path/to/resource", is_secure=self.is_secure) ) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 assert response.url == request.url assert response.body == b"/path/to/resource" @coroutine_test async def test_download_with_proxy_https_timeout( self, proxy_mockserver: ProxyEchoMockServer ) -> None: if NON_EXISTING_RESOLVABLE: pytest.skip("Non-existing hosts are resolvable") if self.is_secure and not self.handler_supports_tls_in_tls: pytest.skip("HTTPS proxies for HTTPS destinations are not supported") http_proxy = proxy_mockserver.url("", is_secure=self.is_secure) domain = "https://no-such-domain.nosuch" request = Request(domain, meta={"proxy": http_proxy, "download_timeout": 0.2}) async with self.get_dh() as download_handler: with pytest.raises(DownloadTimeoutError) as exc_info: await download_handler.download_request(request) assert domain in str(exc_info.value) @coroutine_test async def test_download_with_proxy_without_http_scheme( self, proxy_mockserver: ProxyEchoMockServer ) -> None: http_proxy = f"{proxy_mockserver.host}:{proxy_mockserver.port()}" request = Request("http://example.com", meta={"proxy": http_proxy}) async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 assert response.url == request.url assert response.body == self.expected_http_proxy_request_body PROXY_KINDS = ["http", "https", "socks5"] class TestMitmProxyBase(ABC): # whether the handler supports HTTPS proxies with HTTPS destinations handler_supports_tls_in_tls: bool = True handler_supports_socks: bool = False @property @abstractmethod def settings_dict(self) -> dict[str, Any] | None: raise NotImplementedError def _maybe_skip(self, proxy_kind: str, https_dest: bool) -> None: if proxy_kind == "socks5" and not self.handler_supports_socks: pytest.skip("SOCKS proxies are not supported") if ( proxy_kind == "https" and https_dest and not self.handler_supports_tls_in_tls ): pytest.skip("HTTPS proxies for HTTPS destinations are not supported") @pytest.mark.parametrize("proxy_server", PROXY_KINDS, indirect=True) @pytest.mark.parametrize( "https_dest", [False, True], ids=["HTTP dest", "HTTPS dest"] ) @coroutine_test async def test_proxy( self, caplog: pytest.LogCaptureFixture, proxy_server: str, mockserver: MockServer, https_dest: bool, ) -> None: """HTTP/HTTPS/SOCKS5 proxy, HTTP or HTTPS destination.""" self._maybe_skip(proxy_server, https_dest) crawler = get_crawler(SingleRequestSpider, self.settings_dict) with caplog.at_level(logging.DEBUG): await crawler.crawl_async( seed=mockserver.url("/status?n=200", is_secure=https_dest) ) assert isinstance(crawler.spider, SingleRequestSpider) self._assert_got_response_code(200, caplog.text) self._assert_headers(crawler.spider.meta["responses"][0].headers, https_dest) @pytest.mark.parametrize("proxy_server", PROXY_KINDS, indirect=True) @pytest.mark.parametrize( "https_dest", [False, True], ids=["HTTP dest", "HTTPS dest"] ) @coroutine_test async def test_proxy_auth_error( self, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch, proxy_server: str, mockserver: MockServer, https_dest: bool, ) -> None: """HTTP/HTTPS/SOCKS5 proxy, HTTP or HTTPS destination, wrong proxy creds.""" self._maybe_skip(proxy_server, https_dest) envvar = "https_proxy" if https_dest else "http_proxy" monkeypatch.setenv(envvar, wrong_credentials(os.environ[envvar])) crawler = get_crawler(SimpleSpider, self.settings_dict) with caplog.at_level(logging.DEBUG): await crawler.crawl_async( mockserver.url("/status?n=200", is_secure=https_dest) ) if proxy_server == "socks5": assert "DownloadConnectionRefusedError" in caplog.text else: # The proxy returns a 407 error code but it does not reach the # client; it just sees an exception. self._assert_got_auth_exception(caplog.text) @pytest.mark.parametrize("proxy_server", PROXY_KINDS, indirect=True) @pytest.mark.parametrize( "https_dest", [False, True], ids=["HTTP dest", "HTTPS dest"] ) @coroutine_test async def test_proxy_dont_leak_auth_header( self, caplog: pytest.LogCaptureFixture, proxy_server: str, mockserver: MockServer, https_dest: bool, ) -> None: """HTTP/HTTPS/SOCKS5 proxy, HTTP or HTTPS destination. Check that the auth header is not sent to the destination.""" self._maybe_skip(proxy_server, https_dest) request = Request(mockserver.url("/echo", is_secure=https_dest)) crawler = get_crawler(SingleRequestSpider, self.settings_dict) with caplog.at_level(logging.DEBUG): await crawler.crawl_async(seed=request) assert isinstance(crawler.spider, SingleRequestSpider) self._assert_got_response_code(200, caplog.text) self._assert_headers(crawler.spider.meta["responses"][0].headers, https_dest) echo = json.loads(crawler.spider.meta["responses"][0].text) assert "Proxy-Authorization" not in echo["headers"] @pytest.mark.parametrize("proxy_server", PROXY_KINDS, indirect=True) @pytest.mark.parametrize( "https_dest", [False, True], ids=["HTTP dest", "HTTPS dest"] ) @coroutine_test async def test_proxy_redirect( self, caplog: pytest.LogCaptureFixture, proxy_server: str, mockserver: MockServer, https_dest: bool, ) -> None: """HTTP/HTTPS/SOCKS5 proxy, HTTP or HTTPS destination, following a redirect. Check that the redirected request still goes through the proxy and doesn't lose the proxy auth. """ self._maybe_skip(proxy_server, https_dest) crawler = get_crawler(SingleRequestSpider, self.settings_dict) with caplog.at_level(logging.DEBUG): await crawler.crawl_async( seed=mockserver.url("/redirect", is_secure=https_dest) ) assert isinstance(crawler.spider, SingleRequestSpider) assert crawler.spider.meta.get("failure") is None responses = crawler.spider.meta.get("responses", []) assert len(responses) == 1 assert responses[0].status == 200 assert responses[0].url == mockserver.url("/redirected", is_secure=https_dest) self._assert_got_response_code(200, caplog.text) self._assert_headers(responses[0].headers, https_dest) @staticmethod def _assert_headers(headers: Headers, https_dest: bool) -> None: assert b"X-Via-Mitmproxy" in headers if https_dest: assert b"X-Via-Mitmproxy-TLS" in headers @staticmethod def _assert_got_response_code(code: int, log: str) -> None: assert str(log).count(f"Crawled ({code})") == 1 @staticmethod def _assert_got_auth_exception(log: str) -> None: assert "Proxy Authentication Required" in log or "407" in log class TestRealWebsiteBase(ABC): @property @abstractmethod def download_handler_cls(self) -> type[DownloadHandlerProtocol]: raise NotImplementedError @property @abstractmethod def settings_dict(self) -> dict[str, Any] | None: raise NotImplementedError @property def platform_cert_store_works(self) -> bool: """Whether valid certificates can be verified. Twisted on Windows cannot do that out of the box, see e.g. https://github.com/twisted/twisted/issues/6371. """ return True @asynccontextmanager async def get_dh( self, settings_dict: dict[str, Any] | None = None ) -> AsyncGenerator[DownloadHandlerProtocol]: crawler = get_crawler(DefaultSpider, settings_dict) crawler.spider = crawler._create_spider() dh = build_from_crawler(self.download_handler_cls, crawler) try: yield dh finally: await dh.close() @coroutine_test async def test_download(self) -> None: request = Request("https://books.toscrape.com/") async with self.get_dh() as download_handler: response = await download_handler.download_request(request) assert response.status == 200 assert "All products | Books to Scrape - Sandbox" in response.text @coroutine_test async def test_download_with_spider(self) -> None: crawler = get_crawler(SingleRequestSpider, self.settings_dict) await maybe_deferred_to_future( crawler.crawl(seed=Request("https://books.toscrape.com/")) ) assert isinstance(crawler.spider, SingleRequestSpider) failure = crawler.spider.meta.get("failure") assert failure is None reason = crawler.spider.meta["close_reason"] assert reason == "finished" @coroutine_test async def test_verify_certs(self) -> None: if not self.platform_cert_store_works: pytest.skip("Cannot verify certificates") request = Request("https://books.toscrape.com/") async with self.get_dh( {"DOWNLOAD_VERIFY_CERTIFICATES": True} ) as download_handler: response = await download_handler.download_request(request) assert response.status == 200 assert "All products | Books to Scrape - Sandbox" in response.text @pytest.mark.parametrize("verify_certs", [True, False]) @coroutine_test async def test_tls_logging( self, caplog: pytest.LogCaptureFixture, verify_certs: bool ) -> None: if verify_certs and not self.platform_cert_store_works: pytest.skip("Cannot verify certificates") request = Request("https://books.toscrape.com/") async with self.get_dh( { "DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING": True, "DOWNLOAD_VERIFY_CERTIFICATES": verify_certs, } ) as download_handler: with caplog.at_level("DEBUG"): response = await download_handler.download_request(request) assert response.status == 200 assert "SSL connection to books.toscrape.com using protocol" in caplog.text