From a4b3665abe0ab184d2a715c1be0af679c4cd5335 Mon Sep 17 00:00:00 2001 From: Adrian Chaves Date: Wed, 12 Aug 2026 21:11:24 +0200 Subject: [PATCH] Complete coverage for H2DownloadHandler --- scrapy/core/downloader/handlers/http2.py | 8 +- scrapy/core/http2/agent.py | 22 +--- scrapy/core/http2/protocol.py | 33 +++-- scrapy/core/http2/stream.py | 33 +---- tests/mockserver/http.py | 12 ++ tests/mockserver/http_resources.py | 124 +++++++++++++++++- .../test_downloader_handler_twisted_http2.py | 41 +++++- tests/test_http2_client_protocol.py | 14 +- tests/utils/bases/download_handlers_http.py | 77 +++++++++++ 9 files changed, 293 insertions(+), 71 deletions(-) diff --git a/scrapy/core/downloader/handlers/http2.py b/scrapy/core/downloader/handlers/http2.py index 9b3d4fbd4..99f16c9b4 100644 --- a/scrapy/core/downloader/handlers/http2.py +++ b/scrapy/core/downloader/handlers/http2.py @@ -38,14 +38,12 @@ class H2DownloadHandler(BaseDownloadHandler): super().__init__(crawler) self._crawler = crawler - from twisted.internet import reactor - - self._pool = H2ConnectionPool(reactor, crawler) + self._pool = H2ConnectionPool(crawler) self._context_factory = _load_context_factory_from_settings(crawler) self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS") async def download_request(self, request: Request) -> Response: - if urlparse_cached(request).scheme == "http": # pragma: no cover + if urlparse_cached(request).scheme == "http": raise UnsupportedURLSchemeError( f"{type(self).__name__} doesn't support plain HTTP." ) @@ -83,7 +81,7 @@ class _ScrapyH2Agent: def _get_agent(self, request: Request, timeout: float | None) -> H2Agent: from twisted.internet import reactor - if request.meta.get("proxy"): # pragma: no cover + if request.meta.get("proxy"): raise NotImplementedError(f"{type(self).__name__} doesn't support proxies.") bind_address = request.meta.get("bindaddress") or self._bind_address bind_address = normalize_bind_address(bind_address) diff --git a/scrapy/core/http2/agent.py b/scrapy/core/http2/agent.py index 042557208..6c1aba538 100644 --- a/scrapy/core/http2/agent.py +++ b/scrapy/core/http2/agent.py @@ -6,12 +6,7 @@ from typing import TYPE_CHECKING from twisted.internet import defer from twisted.internet.defer import Deferred from twisted.python.failure import Failure -from twisted.web.client import ( - URI, - BrowserLikePolicyForHTTPS, - ResponseFailed, - _StandardEndpointFactory, -) +from twisted.web.client import URI, BrowserLikePolicyForHTTPS, _StandardEndpointFactory from twisted.web.error import SchemeNotSupported from scrapy.core.downloader.contextfactory import _AcceptableProtocolsContextFactory @@ -30,8 +25,7 @@ ConnectionKeyT = tuple[bytes, bytes, int] class H2ConnectionPool: - def __init__(self, reactor: ReactorBase, crawler: Crawler) -> None: - self._reactor = reactor + def __init__(self, crawler: Crawler) -> None: self._crawler = crawler # Store a dictionary which is used to get the respective @@ -72,7 +66,7 @@ class H2ConnectionPool: ) -> Deferred[H2ClientProtocol]: self._pending_requests[key] = deque() - conn_lost_deferred: Deferred[list[BaseException]] = Deferred() + conn_lost_deferred: Deferred[None] = Deferred() conn_lost_deferred.addCallback(self._remove_connection, key) factory = H2ClientFactory( @@ -102,17 +96,9 @@ class H2ConnectionPool: return conn - def _remove_connection( - self, errors: list[BaseException], key: ConnectionKeyT - ) -> None: + def _remove_connection(self, _: None, key: ConnectionKeyT) -> None: self._connections.pop(key) - # Call the errback of all the pending requests for this connection - pending_requests = self._pending_requests.pop(key, None) - while pending_requests: - d = pending_requests.popleft() - d.errback(ResponseFailed(errors)) - def close_connections(self) -> None: """Close all the HTTP/2 connections and remove them from pool.""" for conn in self._connections.values(): diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index 2d59aba31..709ac25f5 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -56,11 +56,11 @@ PROTOCOL_NAME = b"h2" class InvalidNegotiatedProtocol(H2Error): def __init__(self, negotiated_protocol: bytes) -> None: + super().__init__( + f"Expected {PROTOCOL_NAME!r}, received {negotiated_protocol!r}" + ) self.negotiated_protocol = negotiated_protocol - def __str__(self) -> str: - return f"Expected {PROTOCOL_NAME!r}, received {self.negotiated_protocol!r}" - class RemoteTerminatedConnection(H2Error): def __init__( @@ -68,20 +68,18 @@ class RemoteTerminatedConnection(H2Error): remote_ip_address: IPv4Address | IPv6Address | None, event: ConnectionTerminated, ) -> None: + super().__init__(f"Received GOAWAY frame from {remote_ip_address!r}") self.remote_ip_address = remote_ip_address self.terminate_event = event - def __str__(self) -> str: - return f"Received GOAWAY frame from {self.remote_ip_address!r}" - class MethodNotAllowed405(H2Error): def __init__(self, remote_ip_address: IPv4Address | IPv6Address | None) -> None: + super().__init__( + f"Received 'HTTP/2.0 405 Method Not Allowed' from {remote_ip_address!r}" + ) self.remote_ip_address = remote_ip_address - def __str__(self) -> str: - return f"Received 'HTTP/2.0 405 Method Not Allowed' from {self.remote_ip_address!r}" - @implementer(IHandshakeListener) class H2ClientProtocol(Protocol, TimeoutMixin): @@ -91,7 +89,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin): self, uri: URI, crawler: Crawler, - conn_lost_deferred: Deferred[list[BaseException]], + conn_lost_deferred: Deferred[None], *, tls_verbose_logging: bool = False, ) -> None: @@ -101,12 +99,12 @@ class H2ClientProtocol(Protocol, TimeoutMixin): uri is used to verify that incoming client requests have correct base URL. crawler -- The crawler the requests belong to - conn_lost_deferred -- Deferred that fires with the list of underlying exceptions to notify - that connection was lost + conn_lost_deferred -- Deferred that fires to notify that the + connection was lost tls_verbose_logging -- Whether to log TLS details """ self._crawler: Crawler = crawler - self._conn_lost_deferred: Deferred[list[BaseException]] = conn_lost_deferred + self._conn_lost_deferred: Deferred[None] = conn_lost_deferred self._tls_verbose_logging: bool = tls_verbose_logging config = H2Configuration(client_side=True, header_encoding="utf-8") @@ -356,12 +354,11 @@ class H2ClientProtocol(Protocol, TimeoutMixin): # Cancel the timeout if not done yet self.setTimeout(None) # type: ignore[no-untyped-call] + self._conn_lost_errors.append(reason) + # Notify the connection pool instance such that no new requests are # sent over current connection - if not reason.check(connectionDone): - self._conn_lost_errors.append(reason) - - self._conn_lost_deferred.callback(self._conn_lost_errors) + self._conn_lost_deferred.callback(None) for stream in self.streams.values(): if stream.metadata["request_sent"]: @@ -464,7 +461,7 @@ class H2ClientFactory(Factory): self, uri: URI, crawler: Crawler, - conn_lost_deferred: Deferred[list[BaseException]], + conn_lost_deferred: Deferred[None], *, tls_verbose_logging: bool = False, ) -> None: diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index 4fc300d90..1cb37bea8 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -41,23 +41,23 @@ class InactiveStreamClosed(ConnectionClosed): streams to close and connection is lost.""" def __init__(self, request: Request) -> None: + super().__init__( + f"Connection was closed without sending the request {request!r}" + ) self.request = request - def __str__(self) -> str: - return f"InactiveStreamClosed: Connection was closed without sending the request {self.request!r}" - class InvalidHostname(H2Error): def __init__( self, request: Request, expected_hostname: str, expected_netloc: str ) -> None: + super().__init__( + f"Expected {expected_hostname} or {expected_netloc} in {request}" + ) self.request = request self.expected_hostname = expected_hostname self.expected_netloc = expected_netloc - def __str__(self) -> str: - return f"InvalidHostname: Expected {self.expected_hostname} or {self.expected_netloc} in {self.request}" - class StreamCloseReason(Enum): # Received a StreamEnded event from the remote @@ -177,9 +177,6 @@ class Stream: else: self.close(StreamCloseReason.CANCELLED) - def __repr__(self) -> str: - return f"Stream(id={self.stream_id!r})" - @property def _log_warnsize(self) -> bool: """Checks if we have received data which exceeds the download warnsize @@ -285,14 +282,7 @@ class Stream: If the content length is 0 initially then we end the stream immediately and wait for response data. - - Warning: Only call this method when stream not closed from client side - and has initiated request already by sending HEADER frame. If not then - stream will raise ProtocolError (raise by h2 state machine). """ - if self.metadata["stream_closed_local"]: - raise StreamClosedError(self.stream_id) - # Firstly, check what the flow control window is for current stream. window_size = self._protocol.conn.local_flow_control_window( stream_id=self.stream_id @@ -414,9 +404,6 @@ class Stream: def reset_stream(self, reason: StreamCloseReason = StreamCloseReason.RESET) -> None: """Close this stream by sending a RST_FRAME to the remote peer""" - if self.metadata["stream_closed_local"]: - raise StreamClosedError(self.stream_id) - # The data received so far is the body of the response built for a # stopped download, otherwise the buffer is cleared early to avoid # keeping data in memory for a long time @@ -438,14 +425,6 @@ class Stream: from_protocol: bool = False, ) -> None: """Based on the reason sent we will handle each case.""" - if self.metadata["stream_closed_server"]: - raise StreamClosedError(self.stream_id) - - if not isinstance(reason, StreamCloseReason): - raise TypeError( - f"Expected StreamCloseReason, received {reason.__class__.__qualname__}" - ) - # Have default value of errors as an empty list as # some cases can add a list of exceptions errors = errors or () diff --git a/tests/mockserver/http.py b/tests/mockserver/http.py index 6074f1475..b400e05f3 100644 --- a/tests/mockserver/http.py +++ b/tests/mockserver/http.py @@ -26,6 +26,12 @@ from .http_resources import ( EmptyContentTypeHeaderResource, Follow, ForeverTakingResource, + H2DataAndReset, + H2GoAway, + H2NoSupport, + H2Push, + H2Raw, + H2ResetStream, HostHeaderResource, LargeChunkedFileResource, NoMetaRefreshRedirect, @@ -95,6 +101,12 @@ class Root(BaseResource): put_child(self, b"response-headers", ResponseHeadersResource()) put_child(self, b"set-cookie", SetCookie()) put_child(self, b"uri", UriResource()) + put_child(self, b"h2-reset-stream", H2ResetStream()) + put_child(self, b"h2-data-and-reset", H2DataAndReset()) + put_child(self, b"h2-goaway", H2GoAway()) + put_child(self, b"h2-raw", H2Raw()) + put_child(self, b"h2-no-support", H2NoSupport()) + put_child(self, b"h2-push", H2Push()) def getChild(self, path: bytes, request: Request) -> Root: return self diff --git a/tests/mockserver/http_resources.py b/tests/mockserver/http_resources.py index 969e29a8c..f5ecf5ad7 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, Any, ParamSpec, TypeVar +from typing import TYPE_CHECKING, Any, ClassVar, ParamSpec, TypeVar, cast from urllib.parse import urlencode from twisted.internet.task import deferLater @@ -18,6 +18,7 @@ if TYPE_CHECKING: from twisted.internet.defer import Deferred from twisted.python.failure import Failure + from twisted.web._http2 import H2Connection, H2Stream from twisted.web.http import Request as HTTPRequest from twisted.web.server import Request @@ -450,3 +451,124 @@ class SetCookie(BaseResource): cookie = (cookie_name.decode() + "=" + cookie_value.decode()).encode() request.setHeader(b"Set-Cookie", cookie) return b"" + + +def _h2_connection(request: Request) -> H2Connection: + """Return the HTTP/2 connection that *request* was received on. + + Only works for requests received over HTTP/2. + """ + stream = cast("H2Stream", request.channel) + connection: H2Connection = stream._conn + return connection + + +def _h2_write(request: Request, data: bytes) -> None: + """Write raw data into the HTTP/2 connection that *request* was received + on.""" + transport = _h2_connection(request).transport + assert transport is not None + transport.write(data) + + +def _h2_frame( + frame_type: int, payload: bytes = b"", stream_id: int = 0, flags: int = 0 +) -> bytes: + """Return an HTTP/2 frame (RFC 9113 ยง4.1).""" + return ( + len(payload).to_bytes(3, "big") + + bytes((frame_type, flags)) + + stream_id.to_bytes(4, "big") + + payload + ) + + +class H2ResetStream(LeafResource): + """Reset the HTTP/2 stream of the request instead of answering it""" + + def render_GET(self, request: Request) -> int: + cast("H2Stream", request.channel).abortConnection() + return NOT_DONE_YET + + +class H2GoAway(LeafResource): + """End the HTTP/2 connection of the request with a GOAWAY frame instead of + answering it""" + + def render_GET(self, request: Request) -> int: + connection = _h2_connection(request).conn + connection.close_connection() + _h2_write(request, connection.data_to_send()) + return NOT_DONE_YET + + +class H2DataAndReset(LeafResource): + """Answer the request with response headers and then, within a single write, + a data frame and a reset of its HTTP/2 stream""" + + def render_GET(self, request: Request) -> int: + stream_id = cast("H2Stream", request.channel).streamID + request.write(b"") # sends the response headers + self.deferRequest( + request, + 0.1, + _h2_write, + request, + _h2_frame(0x0, b"a" * 1024, stream_id=stream_id) + # RST_STREAM with the NO_ERROR code + + _h2_frame(0x3, bytes(4), stream_id=stream_id), + ) + return NOT_DONE_YET + + +class H2Raw(LeafResource): + """Write into the HTTP/2 connection of the request the raw data chosen with + the raw url parameter, and then answer the request""" + + raw_data: ClassVar[dict[bytes, bytes]] = { + # a DATA frame above the default maximum frame size of 16 KiB + b"large-frame": _h2_frame(0x0, b"\0" * (2**14 + 1), stream_id=1), + # a frame of a type that is not part of HTTP/2 + b"unknown-frame": _h2_frame(0xFF), + } + + def render_GET(self, request: Request) -> bytes: + raw = getarg(request, b"raw") + _h2_write(request, self.raw_data[raw]) + return b"Works" + + +class H2NoSupport(LeafResource): + """Answer as servers without HTTP/2 support answer the connection preface, + with a 405 status line and nothing else""" + + def render_GET(self, request: Request) -> int: + _h2_write(request, b"HTTP/2.0 405 Method Not Allowed\r\n\r\n") + return NOT_DONE_YET + + +class H2Push(LeafResource): + """Push an empty response into the HTTP/2 connection of the request, and + then answer the request""" + + def render_GET(self, request: Request) -> bytes: + stream_id = cast("H2Stream", request.channel).streamID + authority = request.getHeader(b"host") or b"" + # HPACK (RFC 7541) indexed fields for ":method: GET", ":scheme: https" + # and ":path: /", followed by a literal ":authority" field + promised_request = b"\x82\x87\x84\x01" + bytes((len(authority),)) + authority + # HPACK indexed field for ":status: 200" + pushed_response = b"\x88" + _h2_write( + request, + # PUSH_PROMISE with the END_HEADERS flag set + _h2_frame( + 0x5, + (2).to_bytes(4, "big") + promised_request, + stream_id=stream_id, + flags=0x4, + ) + # HEADERS with the END_STREAM and END_HEADERS flags set + + _h2_frame(0x1, pushed_response, stream_id=2, flags=0x5), + ) + return b"Works" diff --git a/tests/test_downloader_handler_twisted_http2.py b/tests/test_downloader_handler_twisted_http2.py index 2c3954b5e..575490ee6 100644 --- a/tests/test_downloader_handler_twisted_http2.py +++ b/tests/test_downloader_handler_twisted_http2.py @@ -11,7 +11,11 @@ from twisted.web.http import H2_ENABLED from scrapy import Spider from scrapy.crawler import Crawler -from scrapy.exceptions import DownloadFailedError, NotConfigured +from scrapy.exceptions import ( + DownloadFailedError, + NotConfigured, + UnsupportedURLSchemeError, +) from scrapy.http import Request from scrapy.utils.misc import build_from_crawler from tests.utils.bases.download_handlers_http import ( @@ -157,6 +161,41 @@ class TestHttp2(H2DownloadHandlerMixin, TestHttpsBase): with pytest.raises(DownloadFailedError): await download_handler.download_request(request) + @coroutine_test + async def test_download_405_data(self, mockserver: MockServer) -> None: + """Servers without HTTP/2 support answer the connection preface with a + 405 status line, which this handler reports as a download failure + instead of waiting for frames that never arrive.""" + request = Request(mockserver.url("/h2-no-support", is_secure=self.is_secure)) + async with self.get_dh() as download_handler: + with pytest.raises(DownloadFailedError, match="405 Method Not Allowed"): + await download_handler.download_request(request) + + @coroutine_test + async def test_download_plain_http(self, mockserver: MockServer) -> None: + request = Request(mockserver.url("/text")) + async with self.get_dh() as download_handler: + with pytest.raises(UnsupportedURLSchemeError): + await download_handler.download_request(request) + + @coroutine_test + async def test_download_proxy(self, mockserver: MockServer) -> None: + request = Request( + mockserver.url("/text", is_secure=self.is_secure), + meta={"proxy": "https://example.com:8080"}, + ) + async with self.get_dh() as download_handler: + with pytest.raises(NotImplementedError): + await download_handler.download_request(request) + + @coroutine_test + async def test_download_pushed_stream(self, mockserver: MockServer) -> None: + """Pushed responses are ignored.""" + request = Request(mockserver.url("/h2-push", 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" + class TestSimpleHttp2(H2DownloadHandlerMixin, TestSimpleHttpsBase): pass diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py index 431b7a458..2d49729b2 100644 --- a/tests/test_http2_client_protocol.py +++ b/tests/test_http2_client_protocol.py @@ -15,6 +15,7 @@ from pytest_twisted import async_yield_fixture from twisted.internet.defer import Deferred, DeferredList, inlineCallbacks from twisted.internet.endpoints import SSL4ClientEndpoint, SSL4ServerEndpoint from twisted.internet.ssl import Certificate, PrivateCertificate, optionsForClientTLS +from twisted.internet.task import deferLater from twisted.web.client import URI, ResponseFailed from twisted.web.http import H2_ENABLED from twisted.web.http import Request as TxRequest @@ -258,7 +259,7 @@ class TestHttps2ClientProtocol: yield client - if client.connected: + if not client.transport.disconnecting: client.transport.loseConnection() client.transport.abortConnection() @@ -748,6 +749,17 @@ class TestHttps2ClientProtocol: else: pytest.fail("No TimeoutError raised.") + @deferred_f_from_coro_f + async def test_connection_timeout_no_streams( + self, client: H2ClientProtocol + ) -> None: + """A connection with no streams is closed when it times out.""" + from twisted.internet import reactor + + client.setTimeout(0.1) # type: ignore[no-untyped-call] + await maybe_deferred_to_future(deferLater(reactor, 0.5)) + assert client.transport.disconnecting + @deferred_f_from_coro_f async def test_request_headers_received( self, server_port: int, client: H2ClientProtocol diff --git a/tests/utils/bases/download_handlers_http.py b/tests/utils/bases/download_handlers_http.py index 9b4a38724..7857b46a7 100644 --- a/tests/utils/bases/download_handlers_http.py +++ b/tests/utils/bases/download_handlers_http.py @@ -16,6 +16,7 @@ from urllib.parse import urlparse import pytest from cryptography.x509 import load_der_x509_certificate +from twisted.internet.defer import DeferredList from twisted.internet.ssl import Certificate from twisted.python.failure import Failure @@ -131,6 +132,25 @@ class TestHttpBase(ABC): response = await download_handler.download_request(request) assert response.body == b"" + @coroutine_test + async def test_download_concurrent(self, mockserver: MockServer) -> None: + """Requests started before a connection to the server is available are + sent once it is.""" + url = mockserver.url("/text", is_secure=self.is_secure) + async with self.get_dh() as download_handler: + results = await maybe_deferred_to_future( + DeferredList( + [ + deferred_from_coro( + download_handler.download_request(Request(url)) + ) + for _ in range(2) + ], + fireOnOneErrback=True, + ) + ) + assert [response.body for _, response in results] == [b"Works", b"Works"] + @pytest.mark.parametrize( "http_status", [ @@ -758,6 +778,63 @@ class TestHttpBase(ABC): response = await download_handler.download_request(request) assert response.flags == ["dataloss"] + @coroutine_test + async def test_download_stream_reset(self, mockserver: MockServer) -> None: + if not self.http2: + pytest.skip("Streams are specific to HTTP/2") + request = Request(mockserver.url("/h2-reset-stream", is_secure=self.is_secure)) + async with self.get_dh() as download_handler: + with pytest.raises(DownloadFailedError): + await download_handler.download_request(request) + + @coroutine_test + async def test_download_stream_reset_after_data( + self, mockserver: MockServer + ) -> None: + """A stream reset that arrives along with the data that cancels the + download is ignored.""" + if not self.http2: + pytest.skip("Streams are specific to HTTP/2") + request = Request( + mockserver.url("/h2-data-and-reset", is_secure=self.is_secure), + meta={"download_maxsize": 1}, + ) + async with self.get_dh() as download_handler: + with pytest.raises(DownloadCancelledError): + await download_handler.download_request(request) + + @coroutine_test + async def test_download_goaway(self, mockserver: MockServer) -> None: + if not self.http2: + pytest.skip("GOAWAY frames are specific to HTTP/2") + request = Request(mockserver.url("/h2-goaway", is_secure=self.is_secure)) + async with self.get_dh() as download_handler: + with pytest.raises(DownloadFailedError): + await download_handler.download_request(request) + + @coroutine_test + async def test_download_large_h2_frame(self, mockserver: MockServer) -> None: + if not self.http2: + pytest.skip("Frames are specific to HTTP/2") + request = Request( + mockserver.url("/h2-raw?raw=large-frame", is_secure=self.is_secure) + ) + async with self.get_dh() as download_handler: + with pytest.raises(DownloadFailedError): + await download_handler.download_request(request) + + @coroutine_test + async def test_download_unknown_h2_frame(self, mockserver: MockServer) -> None: + """Frames of unknown types are ignored.""" + if not self.http2: + pytest.skip("Frames are specific to HTTP/2") + request = Request( + mockserver.url("/h2-raw?raw=unknown-frame", 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_conn_failed(self) -> None: # copy of TestCrawl.test_retry_conn_failed()