diff --git a/aiohttp/client.py b/aiohttp/client.py index 809fa63ed24..eb2ad26d2ab 100644 --- a/aiohttp/client.py +++ b/aiohttp/client.py @@ -7,8 +7,10 @@ import json import os import sys +import time import traceback import warnings +from collections import deque from collections.abc import ( Awaitable, Callable, @@ -232,22 +234,51 @@ async def _connect_and_send_request(req: ClientRequest) -> ClientResponse: connector = req._session._connector assert connector is not None try: - conn = await connector.connect(req, traces=req._traces, timeout=req._timeout) + async with connector.sem: + # at most just one + conn = await connector.connect( + req, traces=req._traces, timeout=req._timeout + ) except asyncio.TimeoutError as exc: raise ConnectionTimeoutError(f"Connection timeout to host {req.url}") from exc assert conn.protocol is not None - conn.protocol.set_response_params(**req._response_params) + assert conn.protocol.transport is not None + + ssl_object = conn.protocol.transport.get_extra_info("ssl_object") + alpn_protocol = ssl_object.selected_alpn_protocol() if ssl_object else "http/1.1" + + resp = None + started = False + + if alpn_protocol == "h2" and not connector._conns[conn._key]: + connector._conns[conn._key] = deque([(conn.protocol, time.monotonic())]) try: - resp = await req._send(conn) - try: + # backwards compatibility + if alpn_protocol == "h2": + body = await req.body.as_bytes() + resp = await conn.protocol.send( # type: ignore[attr-defined] + req.method, + req.url, + list(req.headers.items()), + body, + ) + + # we are done with the connection + conn._protocol = None + else: + conn.protocol.set_response_params(**req._response_params) + resp = await req._send(conn) await resp.start(conn) - except BaseException: + # h3 not implemented + + started = True + finally: + if resp is not None and not started: resp.close() - raise - except BaseException: - conn.close() - raise + conn.close() + if resp is None: + conn.close() return resp diff --git a/aiohttp/connector.py b/aiohttp/connector.py index c62d216648d..294c0d2b7c6 100644 --- a/aiohttp/connector.py +++ b/aiohttp/connector.py @@ -1,5 +1,6 @@ import asyncio import functools +import os import random import socket import sys @@ -51,6 +52,7 @@ set_exception, set_result, ) +from .http_protocol import HttpDispatcherProtocol from .log import client_logger from .resolver import DefaultResolver @@ -292,7 +294,7 @@ def __init__( ] = defaultdict(OrderedDict) self._loop = loop - self._factory = functools.partial(ResponseHandler, loop=loop) + self._factory = functools.partial(HttpDispatcherProtocol, loop=loop) # start keep-alive connection cleanup task self._cleanup_handle: asyncio.TimerHandle | None = None @@ -319,6 +321,12 @@ def __init__( self._placeholder_future.set_result(None) self._cleanup_closed() + # Semaphore for HTTP/2 connections + # avoids duplicate connections to the + # same host + # (HTTP/2 doesn't need connection pooling to send multiple requests) + self.sem = asyncio.Semaphore(1) + def __del__(self, _warnings: Any = warnings) -> None: if self._closed: return @@ -856,7 +864,11 @@ def _make_ssl_context(verified: bool) -> SSLContext: sslcontext.verify_mode = ssl.CERT_NONE sslcontext.options |= ssl.OP_NO_COMPRESSION sslcontext.set_default_verify_paths() - sslcontext.set_alpn_protocols(("http/1.1",)) + + protocols = ["http/1.1"] + if os.getenv("AIOHTTP_ENABLE_EXPERIMENTAL_PROTOCOLS", False): + protocols += ["h2"] + sslcontext.set_alpn_protocols(tuple(protocols)) return sslcontext @@ -1403,7 +1415,8 @@ async def _start_tls_connection( tls_transport ) # Kick the state machine of the new TLS protocol - return tls_transport, tls_proto + # HACK use the correct type + return tls_transport, tls_proto # type: ignore[return-value] def _convert_hosts_to_addr_infos( self, hosts: list[ResolveResult] @@ -1495,7 +1508,6 @@ async def _create_direct_connection( bad_peer = sock.getpeername() aiohappyeyeballs.remove_addr_infos(addr_infos, bad_peer) continue - return transp, proto assert last_exc is not None raise last_exc @@ -1627,7 +1639,7 @@ async def _create_connection( raise raise UnixClientConnectorError(self.path, req.connection_key, exc) from exc - return proto + return proto # type: ignore[return-value] class NamedPipeConnector(BaseConnector): diff --git a/aiohttp/helpers.py b/aiohttp/helpers.py index 1a45a060ff1..6748183f53e 100644 --- a/aiohttp/helpers.py +++ b/aiohttp/helpers.py @@ -47,7 +47,6 @@ from . import hdrs from .log import client_logger -from .typedefs import PathLike # noqa if sys.version_info >= (3, 11): import asyncio as async_timeout diff --git a/aiohttp/http2/connection.py b/aiohttp/http2/connection.py new file mode 100644 index 00000000000..f6b8b9c3b9f --- /dev/null +++ b/aiohttp/http2/connection.py @@ -0,0 +1,638 @@ +""" +Complete HTTP/2 client implementation (RFC 7540) for asyncio. + +This module provides: +- Frame-level binary wire protocol with debug logging. +- HPACK compression/decompression (using the `hpack` library). +- Full stream state machine (idle -> open -> half‑closed -> closed). +- Multiplexed connection handling (concurrent streams, flow control). +- Server settings tracking (MAX_CONCURRENT_STREAMS, INITIAL_WINDOW_SIZE, etc.). +- Integration point for aiohttp's TCPConnector via an asyncio.Protocol subclass. + +Dependencies: +- asyncio +- struct +- logging +- enum +- hpack (install with `pip install hpack`) +- collections.defaultdict + +Usage: + Connector replaces `ResponseHandler` with `Http2Protocol`. +""" + +import asyncio +import logging +import struct +from typing import Any, Dict, List, Optional, Set, Tuple + +from hpack import Decoder, Encoder + +from .response import Http2Response +from .settings import ( + DEFAULT_SETTINGS, + FlagData, + FlagHeaders, + FlagPing, + FlagSettings, + FrameType, + Setting, +) +from .stream import Stream, StreamState + +# ---------------------------------------------------------------------- +# Logging – plaintext wire‑format emission for debugging +# ---------------------------------------------------------------------- +logger = logging.getLogger("aiohttp.http2.connection") +# logger.setLevel(logging.DEBUG) + +FRAME_HEADER_LENGTH = 9 # 9 octets +STREAM_ID_MASK = 0x7FFFFFFF # to avoid setting the reserved bit to 1 + + +# ---------------------------------------------------------------------- +# Connection‑level management +# ---------------------------------------------------------------------- +class Http2Connection: + """Manages a single HTTP/2 connection. + + Handles: + - Connection preface and SETTINGS handshake. + - Frame parsing and dispatch. + - HPACK encoding/decoding. + - Stream multiplexing and flow control. + - Server settings tracking. + """ + + def __init__( + self, transport: asyncio.Transport, loop: asyncio.AbstractEventLoop + ) -> None: + self._transport = transport + self._loop = loop + + # HPACK + self.hpack_encoder: Encoder = Encoder() + self.hpack_decoder: Decoder = Decoder() + + # Settings + self.remote_settings: Dict[Setting, int] = DEFAULT_SETTINGS.copy() + self.local_settings: Dict[Setting, int] = DEFAULT_SETTINGS.copy() + + # Flow control + self.session_outbound_window: int = ( + 65535 # initial flow control (RFC 7540, 6.9.1) + ) + self.session_inbound_window: int = 65535 + self._flow_control_updated: asyncio.Event = asyncio.Event() + self._flow_control_updated.set() # initially writable + + # Streams + self.streams: Dict[int, Stream] = {} + self.next_stream_id: int = 1 # client streams are odd + self.max_concurrent_streams: int = DEFAULT_SETTINGS[ + Setting.MAX_CONCURRENT_STREAMS + ] + self._pending_streams: List[asyncio.Future[Stream]] = [] + self._last_peer_stream_id: int = ( + 0 # highest server‑initiated stream (even, unused for client) + ) + + # Frame buffers + self._frame_buffer: bytearray = bytearray() + + # GOAWAY state + self._goaway_received: bool = False + self._goaway_sent: bool = False + self._last_stream_id: int = 0 + self._error_code: int = 0 + + # Closed streams cleanup + self._closed_streams: Set[int] = set() + + # -------------------- Transport callbacks -------------------- + def data_received(self, data: bytes) -> None: + """Assemble frames from the byte stream and dispatch them.""" + self._frame_buffer.extend(data) + # Consume complete frames while enough bytes for the header exist + while len(self._frame_buffer) >= FRAME_HEADER_LENGTH: + # Parse 24-bit length, 8-bit type, 8-bit flags, 32-bit stream ID + length = ( + self._frame_buffer[0] << 16 + | self._frame_buffer[1] << 8 + | self._frame_buffer[2] + ) + frame_type_val = self._frame_buffer[3] + flags = self._frame_buffer[4] + stream_id = struct.unpack("!I", self._frame_buffer[5:9])[0] & STREAM_ID_MASK + + if len(self._frame_buffer) < FRAME_HEADER_LENGTH + length: + break # incomplete frame; wait for more data + + payload = bytes( + self._frame_buffer[FRAME_HEADER_LENGTH : FRAME_HEADER_LENGTH + length] + ) + del self._frame_buffer[: FRAME_HEADER_LENGTH + length] + + # invalid frames cause a value error + if 0 <= frame_type_val <= 9: + logger.debug( + "<- %s stream=%d flags=0x%02x len=%d", + FrameType(frame_type_val).name, + stream_id, + flags, + length, + ) + + self._dispatch_frame(frame_type_val, flags, stream_id, payload) + + def eof_received(self) -> bool: + logger.debug("EOF received from server") + self.close() + return False + + def connection_lost(self, exc: Optional[Exception]) -> None: + logger.debug(f"Connection lost: {exc}") + # Cancel all pending streams (including those in the queue) + for stream in list(self.streams.values()): + if not stream.response_future.done(): + stream.response_future.set_exception(ConnectionError("Connection lost")) + for fut in self._pending_streams: + fut.set_exception(ConnectionError("Connection lost")) + self.streams.clear() + + # -------------------- Frame dispatch -------------------- + def _dispatch_frame( + self, frame_type: int, flags: int, stream_id: int, payload: bytes + ) -> None: + if frame_type == FrameType.DATA: + self._handle_data_frame(flags, stream_id, payload) + elif frame_type == FrameType.HEADERS: + self._handle_headers_frame(flags, stream_id, payload) + elif frame_type in { + FrameType.PRIORITY, + FrameType.PUSH_PROMISE, + FrameType.CONTINUATION, + }: + logger.warning("%d frame ignored (not implemented)", frame_type) + elif frame_type == FrameType.RST_STREAM: + self._handle_rst_stream_frame(flags, stream_id, payload) + elif frame_type == FrameType.SETTINGS: + self._handle_settings_frame(flags, stream_id, payload) + elif frame_type == FrameType.PING: + self._handle_ping_frame(flags, stream_id, payload) + elif frame_type == FrameType.GOAWAY: + self._handle_goaway_frame(flags, stream_id, payload) + elif frame_type == FrameType.WINDOW_UPDATE: + self._handle_window_update_frame(flags, stream_id, payload) + else: + logger.warning("Ignoring unknown frame type %d", frame_type) + + # ---------- Individual frame handlers ---------- + def _handle_data_frame(self, flags: int, stream_id: int, payload: bytes) -> None: + stream = self.streams.get(stream_id) + if stream is None: + if stream_id > self._last_peer_stream_id: + self._send_rst_stream(stream_id, 1) # PROTOCOL_ERROR + return + + pad_length = 0 + pos = 0 + if flags & FlagData.PADDED: + pad_length = payload[0] + pos = 1 + + # padding might be too long + data = payload[pos : len(payload) - pad_length] + end_stream = bool(flags & FlagData.END_STREAM) + + # Update session flow control + self.session_inbound_window -= len(data) + # hardcoded values + if self.session_inbound_window < 32768: + self._send_window_update(0, 65535 - self.session_inbound_window) + self.session_inbound_window = 65535 + + stream.receive_data(data, end_stream) + + def _handle_headers_frame(self, flags: int, stream_id: int, payload: bytes) -> None: + if flags & FlagHeaders.PRIORITY: + # Exclusive flag + stream dependency + weight + # exclude priority data + payload = payload[5:] + + # Decode headers with HPACK + try: + headers = self.hpack_decoder.decode(payload) + except Exception as exc: # too general? + logger.error(f"HPACK decode error: {exc}") + self._send_rst_stream(stream_id, 1) # PROTOCOL_ERROR + return + + end_stream = bool(flags & FlagHeaders.END_STREAM) + + stream = self.streams.get(stream_id) + + if stream is None: + logger.error("Unknown stream_id: %d", stream_id) + else: + stream.receive_headers(headers, end_stream) + + def _handle_rst_stream_frame( + self, flags: int, stream_id: int, payload: bytes + ) -> None: + del flags # rst doesn't use flags + + error_code = struct.unpack("!I", payload)[0] + stream = self.streams.get(stream_id) + if stream: + stream.transition(StreamState.CLOSED) + if not stream.response_future.done(): + stream.response_future.set_exception( + RuntimeError(f"Stream reset by server (code={error_code})") + ) + self._close_stream(stream) + + def _handle_settings_frame( + self, flags: int, stream_id: int, payload: bytes + ) -> None: + """Process SETTINGS frame (6.5)""" + if flags & FlagSettings.ACK: + logger.debug("Received SETTINGS ACK") + return # Our settings were acknowledged + if stream_id != 0: + logger.error( + "SETTING frame received after the first stream in violation of the protocol standard (RFC-9113, 3.4)" + ) + self._protocol_error() + return + + if len(payload) % 6 != 0: + logger.error("SETTINGS payload length not a multiple of 6") + self._protocol_error() + return + + # Parse key‑value pairs + for i in range(0, len(payload), 6): + identifier, value = struct.unpack("!H I", payload[i : i + 6]) + if identifier < 0 or identifier > 9: + logger.warning("Unknown setting identifier %d", identifier) + continue + setting = Setting(identifier) + old_value = self.remote_settings.get(setting, value) + self.remote_settings[setting] = value + logger.info(f"Server SETTINGS: {setting.name} = {value}") + + # React to certain settings + if setting == Setting.INITIAL_WINDOW_SIZE and value != old_value: + # might become negative + # send WINDOW_UPDATE + delta = value - old_value + for s in self.streams.values(): + s.outbound_window += delta + elif setting == Setting.MAX_CONCURRENT_STREAMS: + self.max_concurrent_streams = value + self._maybe_unblock_streams() + elif setting == Setting.HEADER_TABLE_SIZE: + self.hpack_encoder.header_table_size = value + + # Acknowledge settings + self._send_settings_ack() + + def _handle_ping_frame(self, flags: int, stream_id: int, payload: bytes) -> None: + if stream_id != 0: + self._protocol_error() + return + if flags & FlagPing.ACK: + logger.debug("Received PING ACK") + else: + # Respond with ACK + logger.debug("Received PING, sending ACK") + self._send_ping(ack=True, opaque_data=payload) + + def _handle_goaway_frame(self, flags: int, stream_id: int, payload: bytes) -> None: + del flags, stream_id # interface + + self._goaway_received = True + last_stream_id, error_code = struct.unpack("!I I", payload[:8]) + extra = payload[8:] + self._last_stream_id = last_stream_id + self._error_code = error_code + logger.info( + "GOAWAY received: last_stream=%d, error=%d, extra=%s", + last_stream_id, + error_code, + extra.decode(), + ) + # Cancel streams with higher IDs + for sid, stream in list(self.streams.items()): + if sid > last_stream_id: + if not stream.response_future.done(): + stream.response_future.set_exception( + ConnectionError("GOAWAY received") + ) + self._close_stream(stream) + # clear pending streams? + + def _handle_window_update_frame( + self, flags: int, stream_id: int, payload: bytes + ) -> None: + increment = struct.unpack("!I", payload)[0] + if stream_id == 0: + # Session window update + self.session_outbound_window += increment + else: + stream = self.streams.get(stream_id) + if stream: + stream.outbound_window += increment + # Wake up any writer waiting for flow control + self._flow_control_updated.set() + + # -------------------- Frame sending helpers -------------------- + def _send_frame( + self, + frame_type: FrameType, + flags: int, + stream_id: int, + payload: bytes = b"", + ) -> None: + length = len(payload) & 0x00FFFFFF # 24 bits -> 3 bytes + header = struct.pack("!I", length)[ + 1: + ] + struct.pack( # drop the first (most‑significant) byte → 3 bytes + "!B B I", frame_type, flags, stream_id + ) + + logger.debug( + f"-> FRAME type={frame_type.name:>15} flags=0x{flags:02x} " + f"stream_id={stream_id:<5} length={len(payload)}" + ) + self._transport.write(header + payload) + + def _send_settings_ack(self) -> None: + self._send_frame(FrameType.SETTINGS, FlagSettings.ACK, 0) + + def _send_ping(self, ack: bool = False, opaque_data: bytes = b"\x00" * 8) -> None: + flags = FlagPing.ACK if ack else 0 + self._send_frame(FrameType.PING, flags, 0, opaque_data) + + def _send_goaway(self, last_stream_id: int, error_code: int) -> None: + payload = struct.pack("!I I", last_stream_id, error_code) + self._send_frame(FrameType.GOAWAY, 0, 0, payload) + self._goaway_sent = True + + def _send_rst_stream(self, stream_id: int, error_code: int) -> None: + payload = struct.pack("!I", error_code) + self._send_frame(FrameType.RST_STREAM, 0, stream_id, payload) + + def _send_window_update(self, stream_id: int, increment: int) -> None: + payload = struct.pack("!I", increment) + self._send_frame(FrameType.WINDOW_UPDATE, 0, stream_id, payload) + + # -------------------- Stream lifecycle -------------------- + def _close_stream(self, stream: Stream) -> None: + self.streams.pop(stream.stream_id, None) + self._closed_streams.add(stream.stream_id) + # Release stream concurrency slot + self._maybe_unblock_streams() + + def _maybe_unblock_streams(self) -> None: + """Create streams from pending requests if concurrency allows.""" + while self._pending_streams and len(self.streams) < self.max_concurrent_streams: + fut = self._pending_streams.pop(0) + if not fut.done(): + stream = self._create_stream_internal() + fut.set_result(stream) + + def _create_stream_internal(self) -> Stream: + sid = self.next_stream_id + self.next_stream_id += 2 # next client stream + stream = Stream(sid, self, self._loop) + self.streams[sid] = stream + return stream + + async def create_stream(self) -> Stream: + """Return a new client stream, waiting if concurrency limit is reached.""" + if self._goaway_sent or self._goaway_received: + raise ConnectionError("Connection is shutting down") + if len(self.streams) < self.max_concurrent_streams: + return self._create_stream_internal() + # Queue the request + fut: asyncio.Future[Stream] = self._loop.create_future() + self._pending_streams.append(fut) + return await fut + + # -------------------- Request sending -------------------- + async def send_data( + self, stream: Stream, data: bytes, end_stream: bool = True + ) -> None: + """Asynchronously send DATA frames, respecting flow control windows.""" + max_frame_size = self.remote_settings[Setting.MAX_FRAME_SIZE] + offset = 0 + total = len(data) + + while offset < total: + # Wait until both session and stream windows have capacity + while stream.outbound_window <= 0 or self.session_outbound_window <= 0: + self._flow_control_updated.clear() + await self._flow_control_updated.wait() + + chunk_size = min( + max_frame_size, + stream.outbound_window, + self.session_outbound_window, + total - offset, + ) + flags = 0 + if offset + chunk_size >= total and end_stream: + flags |= FlagData.END_STREAM + + if stream.state == StreamState.OPEN: + stream.transition(StreamState.HALF_CLOSED_LOCAL) + elif stream.state == StreamState.HALF_CLOSED_REMOTE: + stream.transition(StreamState.CLOSED) + self._close_stream(stream) + # else: error + + self._send_frame( + FrameType.DATA, + flags, + stream.stream_id, + data[offset : offset + chunk_size], + ) + stream.outbound_window -= chunk_size + self.session_outbound_window -= chunk_size + offset += chunk_size + + # there is space available + if self.session_outbound_window and stream.outbound_window: + self._flow_control_updated.set() + + async def send_request( + self, + stream: Stream, + method: str, + url: Any, # Usually a yarl.URL, but abstracted for mocking + headers: List[Tuple[str, str]], + body: Optional[bytes] = None, + ) -> None: + """Send HEADERS and optional DATA frames for a stream.""" + # :path needs the query as well + path_and_query = url.path + if url.query: + path_and_query += "?" + url.raw_query_string + + # Build pseudo‑headers + req_headers = [ + (":method", method), + (":path", path_and_query), + (":scheme", url.scheme), + (":authority", url.host), + ] + + for name, value in headers: + lname = name.lower() + # HTTP/2 forbids connection-specific headers and the Host header + if lname in ( + "host", + "connection", + "keep-alive", + "proxy-connection", + "transfer-encoding", + "upgrade", + ): + continue + req_headers.append((lname, value)) + + hdrs = self.hpack_encoder.encode(req_headers) + end_stream = body is None or len(body) == 0 + flags = FlagHeaders.END_HEADERS + + # stream transitions + stream.transition(StreamState.OPEN) + if end_stream: + flags |= FlagHeaders.END_STREAM + stream.transition(StreamState.HALF_CLOSED_LOCAL) + + self._send_frame(FrameType.HEADERS, flags, stream.stream_id, hdrs) + + if body: + await self.send_data(stream, body, end_stream=True) + + # -------------------- Connection lifecycle -------------------- + def initiate_connection(self) -> None: + """Send the connection preface and initial SETTINGS.""" + # Connection preface (RFC 7540, 3.5) + self._transport.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") + + # Send initial SETTINGS (our preferences) + settings_payload = struct.pack( + "!H I", Setting.ENABLE_PUSH, 0 # disable server push + ) + self._send_frame(FrameType.SETTINGS, 0, 0, settings_payload) + + # Update local HPACK table size if needed + self.hpack_encoder.header_table_size = self.local_settings[ + Setting.HEADER_TABLE_SIZE + ] + + logger.debug("Connection preface and initial SETTINGS sent") + + def _protocol_error(self) -> None: + self._send_goaway(0, 1) # PROTOCOL_ERROR + self._transport.close() + + # -------------------- Shutdown -------------------- + def close(self) -> None: + """Perform graceful shutdown.""" + self._transport.close() + + @property + def should_close(self) -> bool: + return self._goaway_sent or self._goaway_received + + def is_connected(self) -> bool: + return not self._transport.is_closing() + + +# ---------------------------------------------------------------------- +# Protocol wrapper for asyncio.Transport (connector integration) +# ---------------------------------------------------------------------- +class Http2Protocol(asyncio.Protocol): + """asyncio.Protocol subclass that bridges transport and Http2Connection. + + This replaces aiohttp's ResponseHandler in the connector. + """ + + def __init__(self, loop: asyncio.AbstractEventLoop) -> None: + self._loop = loop + self._connection: Optional[Http2Connection] = None + self._closed_future: asyncio.Future[None] = loop.create_future() + self.transport: Optional[asyncio.Transport] = None + + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self.transport = transport # type: ignore[assignment] + self._connection = Http2Connection(self.transport, self._loop) # type: ignore[arg-type] + self._connection.initiate_connection() + + def data_received(self, data: bytes) -> None: + if self._connection: + self._connection.data_received(data) + + def eof_received(self) -> bool: + if self._connection: + self._connection.eof_received() + return False + + def connection_lost(self, exc: Optional[Exception]) -> None: + if self._connection: + self._connection.connection_lost(exc) + if not self._closed_future.done(): + self._closed_future.set_result(None) + + # Compatibility with connector's expectations + @property + def should_close(self) -> bool: + return self._connection.should_close if self._connection else False + + def is_connected(self) -> bool: + return self._connection.is_connected() if self._connection else False + + @property + def closed(self) -> asyncio.Future[None]: + return self._closed_future + + async def send( + self, + method: str, + url: Any, + headers: List[Tuple[str, str]], + body: Optional[bytes] = None, + ) -> Http2Response: + """Send an HTTP/2 request and return a compatible response.""" + if self._connection is None: + raise RuntimeError("Connection not established") + + # Obtain a stream from the pool + stream = await self._connection.create_stream() + + await self._connection.send_request(stream, method, url, headers, body) + + # Wait for the response future to be set by the stream when complete + _, resp_headers, resp_body = await stream.response_future + + response = Http2Response( + headers=resp_headers, + body=resp_body, + method=method, + url=url, + ) + + return response + + def close(self) -> None: + if self._connection: + self._connection.close() + self.transport = None + + def abort(self) -> None: + self.close() diff --git a/aiohttp/http2/errors.py b/aiohttp/http2/errors.py new file mode 100644 index 00000000000..339578b44f4 --- /dev/null +++ b/aiohttp/http2/errors.py @@ -0,0 +1,2 @@ +class ProtocolError(Exception): + pass diff --git a/aiohttp/http2/response.py b/aiohttp/http2/response.py new file mode 100644 index 00000000000..ba78ddf4f7a --- /dev/null +++ b/aiohttp/http2/response.py @@ -0,0 +1,126 @@ +import json +from http.cookies import SimpleCookie +from typing import Any, Iterable, List, Optional, Tuple + +from hpack import HeaderTuple +from multidict import CIMultiDict + +from aiohttp.client_exceptions import ClientResponseError +from aiohttp.client_reqrep import RequestInfo + +from ..compression_utils import ZLibDecompressor +from ..http_writer import HttpVersion2 + + +class Http2Response: + """A fully aiohttp.ClientResponse-compatible response for HTTP/2.""" + + def __init__( + self, + headers: Iterable[HeaderTuple] | Iterable[Tuple[str, str]], + body: bytes, + *, + method: Optional[str] = None, + url: Optional[Any] = None, + ) -> None: + # Headers as case-insensitive multi-dict + self.headers: CIMultiDict[str] = CIMultiDict(headers) # type: ignore[arg-type] + encoding = self.headers.get("content-encoding", None) + if encoding in {"gzip", "deflate"}: + comp = ZLibDecompressor(encoding=encoding) + body = comp.decompress_sync(body) + + self.reason: str = "" # HTTP/2 doesn't carry a reason phrase + self._body: bytes = body + self.url: Optional[Any] = url + self.method: Optional[str] = method + # redirects + self._history: List["Http2Response"] = [] + + # no status error implies a server side error + self.status: int = int(self.headers.get(":status", 500)) + + # Cookie jar integration + self._cookies: Optional[SimpleCookie] = None + + # HTTP version pseudo-attribute + self.version = HttpVersion2 + + self._raw_cookie_headers: Optional[List[str]] = self.headers.getall( + "set-cookie", [] + ) + self.connection: Optional[Any] = None + + # ---------------------------------------------------------------- + # Body access (synchronous: entire body is already in memory) + # ---------------------------------------------------------------- + async def read(self) -> bytes: + """Return the response body.""" + return self._body + + @property + def body(self) -> bytes: + return self._body + + async def text(self, encoding: str = "utf-8") -> str: + """Decode the body to a string.""" + return self._body.decode(encoding) + + async def json(self, **kwargs: Any) -> Any: + """Parse JSON body.""" + return json.loads(self._body, **kwargs) + + # ---------------------------------------------------------------- + # Cookies + # ---------------------------------------------------------------- + @property + def cookies(self) -> SimpleCookie: + """Parse 'Set-Cookie' headers and return a SimpleCookie.""" + if self._cookies is None: + self._cookies = SimpleCookie() + for raw in self.headers.getall("set-cookie", []): + self._cookies.load(raw) + return self._cookies + + # ---------------------------------------------------------------- + # Status helpers + # ---------------------------------------------------------------- + @property + def ok(self) -> bool: + """True if status is < 400.""" + return 200 <= self.status < 400 + + def raise_for_status(self) -> None: + """Raise an HTTPError for 4xx/5xx responses.""" + if not self.ok: + raise ClientResponseError( + request_info=RequestInfo( + url=self.url, method=self.method, headers=self.headers # type: ignore[arg-type] + ), + history=self._history, # type: ignore[arg-type] + status=self.status, + message=f"{self.status}, message='{self.reason}'", + headers=self.headers, + ) + + # ---------------------------------------------------------------- + # Connection release (stream-level cleanup) + # ---------------------------------------------------------------- + def release(self) -> None: + """Release the HTTP/2 stream back to the connection.""" + pass # nothing to do; the stream has ended + + def close(self) -> None: + if self.connection: + self.connection.close() + + # ---------------------------------------------------------------- + # Context manager support + # ---------------------------------------------------------------- + async def __aenter__(self) -> "Http2Response": + return self + + async def __aexit__( + self, exc_type: Optional[type], exc: Optional[BaseException], tb: Any + ) -> None: + pass diff --git a/aiohttp/http2/settings.py b/aiohttp/http2/settings.py new file mode 100644 index 00000000000..c9c5b303a3c --- /dev/null +++ b/aiohttp/http2/settings.py @@ -0,0 +1,61 @@ +from enum import IntEnum, IntFlag +from typing import Dict + + +# ---------------------------------------------------------------------- +# HTTP/2 Frame Definitions (RFC 7540, 4) +# ---------------------------------------------------------------------- +class FrameType(IntEnum): + DATA = 0x0 + HEADERS = 0x1 + PRIORITY = 0x2 + RST_STREAM = 0x3 + SETTINGS = 0x4 + PUSH_PROMISE = 0x5 + PING = 0x6 + GOAWAY = 0x7 + WINDOW_UPDATE = 0x8 + CONTINUATION = 0x9 + + +class FlagData(IntFlag): + END_STREAM = 0x1 + PADDED = 0x8 + + +class FlagHeaders(IntFlag): + END_STREAM = 0x1 + END_HEADERS = 0x4 + PADDED = 0x8 + PRIORITY = 0x20 + + +class FlagSettings(IntFlag): + ACK = 0x1 + + +class FlagPing(IntFlag): + ACK = 0x1 + + +# Known settings parameters +class Setting(IntEnum): + HEADER_TABLE_SIZE = 0x1 + ENABLE_PUSH = 0x2 + MAX_CONCURRENT_STREAMS = 0x3 + INITIAL_WINDOW_SIZE = 0x4 + MAX_FRAME_SIZE = 0x5 + MAX_HEADER_LIST_SIZE = 0x6 + SETTINGS_ENABLE_CONNECT_PROTOCOL = 0x8 + NO_RFC7540_PRIORITIES = 0x9 + + +# Default values (RFC 7540, 6.5.2) +DEFAULT_SETTINGS: Dict[Setting, int] = { + Setting.HEADER_TABLE_SIZE: 4096, + Setting.ENABLE_PUSH: 1, + Setting.MAX_CONCURRENT_STREAMS: 2**32 - 1, + Setting.INITIAL_WINDOW_SIZE: 65535, + Setting.MAX_FRAME_SIZE: 16384, + Setting.MAX_HEADER_LIST_SIZE: 2**32 - 1, +} diff --git a/aiohttp/http2/stream.py b/aiohttp/http2/stream.py new file mode 100644 index 00000000000..c98722b29bd --- /dev/null +++ b/aiohttp/http2/stream.py @@ -0,0 +1,155 @@ +import asyncio +from enum import IntEnum +from typing import TYPE_CHECKING, Dict, Iterable, Optional, Set, Tuple + +from hpack import HeaderTuple + +from .errors import ProtocolError +from .settings import Setting + +if TYPE_CHECKING: + from .connection import Http2Connection + + +# ---------------------------------------------------------------------- +# Stream State Machine (RFC 7540 5.1) +# ---------------------------------------------------------------------- +class StreamState(IntEnum): + IDLE = 0 + RESERVED_LOCAL = 1 + RESERVED_REMOTE = 2 + OPEN = 3 + HALF_CLOSED_LOCAL = 4 + HALF_CLOSED_REMOTE = 5 + CLOSED = 6 + + +# Valid transitions (RFC 7540 Figure 2) +VALID_TRANSITIONS: Dict[StreamState, Set[StreamState]] = { + StreamState.IDLE: { + StreamState.OPEN, + StreamState.RESERVED_LOCAL, + StreamState.RESERVED_REMOTE, + }, + StreamState.RESERVED_LOCAL: {StreamState.HALF_CLOSED_REMOTE, StreamState.CLOSED}, + StreamState.RESERVED_REMOTE: {StreamState.HALF_CLOSED_LOCAL, StreamState.CLOSED}, + StreamState.OPEN: {StreamState.HALF_CLOSED_LOCAL, StreamState.HALF_CLOSED_REMOTE}, + StreamState.HALF_CLOSED_LOCAL: {StreamState.CLOSED}, + StreamState.HALF_CLOSED_REMOTE: {StreamState.CLOSED}, + StreamState.CLOSED: set(), +} + + +# ---------------------------------------------------------------------- +# Stream object – multiplexed stream handle +# ---------------------------------------------------------------------- +class Stream: + """A single HTTP/2 stream. + + Manages state, flow control, and provides a future for the response. + """ + + __slots__ = ( + "stream_id", + "state", + "conn", + "outbound_window", + "inbound_window", + "response_future", + "response_headers", + "response_data", + "closed_event", + "_inbound_window_initial", + ) + + def __init__( + self, stream_id: int, conn: "Http2Connection", loop: asyncio.AbstractEventLoop + ) -> None: + self.stream_id = stream_id + self.state = StreamState.IDLE + self.conn = conn + + # Flow‑control windows (stream‑level only) + self.outbound_window: int = conn.remote_settings[Setting.INITIAL_WINDOW_SIZE] + self.inbound_window: int = conn.local_settings[Setting.INITIAL_WINDOW_SIZE] + # Store the initial inbound window for refilling without hard-coded values + self._inbound_window_initial: int = self.inbound_window + + self.response_future: asyncio.Future[ + Tuple[int, Iterable[HeaderTuple] | Iterable[Tuple[str, str]], bytes] + ] = loop.create_future() + + self.response_headers: Optional[ + Iterable[HeaderTuple] | Iterable[Tuple[str, str]] + ] = None + self.response_data: bytearray = bytearray() + + # Event notified when the stream enters CLOSED state + self.closed_event: asyncio.Event = asyncio.Event() + + def transition(self, new_state: StreamState) -> None: + if ( + new_state not in VALID_TRANSITIONS[self.state] + and new_state != StreamState.CLOSED + ): + raise ProtocolError( + f"Invalid stream state transition {self.state.name} -> {new_state.name}" + ) + self.state = new_state + if new_state == StreamState.CLOSED: + self.closed_event.set() + + # ------------------------------------------------------------------ + # Data and header reception + # ------------------------------------------------------------------ + def receive_data(self, data: bytes, end_stream: bool) -> None: + """Process incoming DATA frame payload.""" + self.inbound_window -= len(data) + self.response_data.extend(data) + + # --- stream-level flow control refill --- + if self.inbound_window < self._inbound_window_initial // 2: + increment = self._inbound_window_initial - self.inbound_window + self.inbound_window = self._inbound_window_initial + # Use the connection’s helper to send the WINDOW_UPDATE frame + self.conn._send_window_update(self.stream_id, increment) + + if end_stream: + if self.state == StreamState.OPEN: + self.transition(StreamState.HALF_CLOSED_REMOTE) + elif self.state == StreamState.HALF_CLOSED_LOCAL: + self.transition(StreamState.CLOSED) + self.conn._close_stream(self) + else: + raise ProtocolError( + f"Unexpected stream state {self.state.name} for END_STREAM" + ) + + self.maybe_deliver_response() + + def maybe_deliver_response(self) -> None: + # Deliver the full response once headers have been received + if self.response_headers is not None and not self.response_future.done(): + self.response_future.set_result( + (self.stream_id, self.response_headers, bytes(self.response_data)) + ) + + def receive_headers( + self, + headers: Iterable[HeaderTuple] | Iterable[Tuple[str, str]], + end_stream: bool, + ) -> None: + """Process incoming HEADERS frame payload.""" + self.response_headers = headers + + if end_stream: + if self.state == StreamState.OPEN: + self.transition(StreamState.HALF_CLOSED_REMOTE) + elif self.state == StreamState.HALF_CLOSED_LOCAL: + self.transition(StreamState.CLOSED) + self.conn._close_stream(self) + else: + raise ProtocolError( + f"Unexpected stream state {self.state.name} for END_STREAM on headers" + ) + self.maybe_deliver_response() diff --git a/aiohttp/http_protocol.py b/aiohttp/http_protocol.py new file mode 100644 index 00000000000..596f23bd48f --- /dev/null +++ b/aiohttp/http_protocol.py @@ -0,0 +1,57 @@ +import asyncio +from typing import Any, Optional + +from .client_proto import ResponseHandler +from .http2.connection import Http2Protocol + + +class HttpDispatcherProtocol(asyncio.Protocol): + """Protocol that switches between HTTP/1.1 and HTTP/2 based on ALPN.""" + + __slots__ = ("_loop", "_transport", "_handler") + + def __init__(self, loop: asyncio.AbstractEventLoop) -> None: + self._loop = loop + self._transport: Optional[asyncio.Transport] = None + self._handler: Optional[asyncio.Protocol] = None + + # ---- Transport callbacks forwarded to the real handler ---- + def connection_made(self, transport: asyncio.BaseTransport) -> None: + if self._handler: + return self._handler.connection_made(transport) + + self._transport = transport # type: ignore[assignment] + + # Determine ALPN after TLS is established + ssl_object: Any = transport.get_extra_info("ssl_object") + alpn_protocol: str = ( + ssl_object.selected_alpn_protocol() if ssl_object else "http/1.1" + ) + + if alpn_protocol == "h2": + self._handler = Http2Protocol(self._loop) + else: + self._handler = ResponseHandler(self._loop) + + # Hand the real transport to the handler. The handler will now own + # all incoming data and callbacks. + self._handler.connection_made(transport) + + def __getattribute__(self, name: str) -> Any: + if not name.startswith("__") and name not in { + "connection_made", + "__getattribute__", + "_handler", + "_transport", + "_loop", + }: + return getattr(self._handler, name) + return super().__getattribute__(name) + + def __setattr__(self, name: str, value: Any) -> None: + if name not in {"_handler", "_transport", "_loop"}: + return self._handler.__setattr__(name, value) + return super().__setattr__(name, value) + + def __delattr__(self, name: str) -> None: + return self._handler.__delattr__(name) diff --git a/aiohttp/http_writer.py b/aiohttp/http_writer.py index a1168cfdebb..e648b8b2462 100644 --- a/aiohttp/http_writer.py +++ b/aiohttp/http_writer.py @@ -3,13 +3,11 @@ import asyncio import re import sys -from typing import ( # noqa +from typing import ( TYPE_CHECKING, - Any, Awaitable, Callable, Iterable, - List, NamedTuple, Optional, Union, @@ -44,6 +42,7 @@ class HttpVersion(NamedTuple): HttpVersion10 = HttpVersion(1, 0) HttpVersion11 = HttpVersion(1, 1) +HttpVersion2 = HttpVersion(2, 0) _T_OnChunkSent = Optional[ diff --git a/docs/conf.py b/docs/conf.py index a3254645cfb..9cf25a6082e 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -66,7 +66,7 @@ try: - import sphinxcontrib.spelling # noqa + import sphinxcontrib.spelling extensions.append("sphinxcontrib.spelling") except ImportError: diff --git a/pyproject.toml b/pyproject.toml index 0c27cc88bb5..a82ee886371 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,7 @@ dependencies = [ "propcache >= 0.2.0", "typing_extensions >= 4.4 ; python_version < '3.13'", "yarl >= 1.17.0, < 2.0", + "hpack >= 4.2.0" ] dynamic = [ "version", diff --git a/requirements/runtime-deps.in b/requirements/runtime-deps.in index d70fc5a9dbc..52b6be17191 100644 --- a/requirements/runtime-deps.in +++ b/requirements/runtime-deps.in @@ -8,6 +8,7 @@ backports.zstd; platform_python_implementation == 'CPython' and python_version < Brotli >= 1.2; platform_python_implementation == 'CPython' and sys_platform != 'android' and sys_platform != 'ios' brotlicffi >= 1.2; platform_python_implementation != 'CPython' frozenlist >= 1.1.1 +hpack >= 4.2.0 multidict >=4.5, < 7.0 propcache >= 0.2.0 typing_extensions >= 4.4 ; python_version < '3.13' diff --git a/tests/http2/test_http2.py b/tests/http2/test_http2.py new file mode 100644 index 00000000000..32ff86c2e59 --- /dev/null +++ b/tests/http2/test_http2.py @@ -0,0 +1,1354 @@ +""" +Test suite for aiohttp.http2 + +Categories: +- integration: against a real httpbin server (skipped, requires network) +- unit / protocol: black‑box frame‑level RFC compliance +- unit / misc: race conditions, deadlocks, edge cases +- unit / interface: high‑level features (JSON, redirects, compression, etc.) +""" + +import asyncio +import gzip +import json +import struct +from typing import Any, Dict, Generator, List, Optional, Tuple +from unittest.mock import MagicMock + +import pytest +from hpack import Encoder + +import aiohttp +from aiohttp.connector import TCPConnector +from aiohttp.http2.connection import Http2Connection, Http2Protocol +from aiohttp.http2.errors import ProtocolError +from aiohttp.http2.response import Http2Response +from aiohttp.http2.settings import ( + FlagData, + FlagHeaders, + FlagPing, + FlagSettings, + FrameType, + Setting, +) +from aiohttp.http2.stream import StreamState + + +# ---------------------------------------------------------------------- +# Helper: minimal URL mock +# ---------------------------------------------------------------------- +def url_mock(path: str = "/") -> Any: + """Create a simple URL-like object expected by the implementation.""" + return type( + "URL", + (), + {"scheme": "https", "host": "example.com", "path": path, "query": None}, + ) + + +# ---------------------------------------------------------------------- +# Fixtures +# ---------------------------------------------------------------------- +@pytest.fixture +def mock_transport() -> MagicMock: + """Return a mock asyncio.Transport that records writes.""" + t = MagicMock(spec=asyncio.Transport) + t.is_closing.return_value = False + t.write = MagicMock() + t.close = MagicMock() + return t + + +@pytest.fixture(scope="session") +def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]: + """Create a new event loop for each test.""" + loop = asyncio.new_event_loop() + try: + yield loop + finally: + loop.close() + + +@pytest.fixture +async def connection(mock_transport: MagicMock) -> Tuple[Http2Connection, MagicMock]: + """Set up Http2Connection with mock transport, send preface, and clear write log.""" + event_loop = asyncio.get_running_loop() + + conn = Http2Connection(mock_transport, event_loop) + conn.initiate_connection() + mock_transport.write.reset_mock() # discard preface + initial SETTINGS + return conn, mock_transport + + +@pytest.fixture +async def protocol(mock_transport: MagicMock) -> Tuple[Http2Protocol, MagicMock]: + """Create Http2Protocol and simulate connection_made.""" + event_loop = asyncio.get_running_loop() + proto = Http2Protocol(event_loop) + proto.connection_made(mock_transport) + mock_transport.write.reset_mock() + return proto, mock_transport + + +# ---------------------------------------------------------------------- +# Frame construction helpers +# ---------------------------------------------------------------------- +def frame_header(length: int, ftype: int, flags: int, stream_id: int) -> bytes: + # 24-bit length (3 bytes) + type + flags + stream_id + return struct.pack("!I", length)[1:] + struct.pack( + "!B B I", ftype, flags, stream_id + ) + + +def build_settings_frame( + settings_pairs: Optional[List[Tuple[Setting, int]]] = None, ack: bool = False +) -> bytes: + payload = b"" + if not ack and settings_pairs: + for setting_id, value in settings_pairs: + payload += struct.pack("!H I", setting_id, value) + flags = FlagSettings.ACK if ack else 0 + return frame_header(len(payload), FrameType.SETTINGS, flags, 0) + payload + + +def build_headers_frame( + stream_id: int, + headers: List[Tuple[str, str]], + end_headers: bool = True, + end_stream: bool = False, + priority: Optional[bytes] = None, +) -> bytes: + encoder = Encoder() + header_block = encoder.encode(headers) + flags = 0 + if end_headers: + flags |= FlagHeaders.END_HEADERS + if end_stream: + flags |= FlagHeaders.END_STREAM + if priority is not None: + flags |= FlagHeaders.PRIORITY + header_block = priority + header_block + return ( + frame_header(len(header_block), FrameType.HEADERS, flags, stream_id) + + header_block + ) + + +def build_data_frame( + stream_id: int, data: bytes, end_stream: bool = False, pad: bool = False +) -> bytes: + payload = data + flags = 0 + if end_stream: + flags |= FlagData.END_STREAM + if pad: + pad_len = 1 # minimal padding for test + payload = bytes([pad_len]) + data + b"\x00" * pad_len + flags |= FlagData.PADDED + return frame_header(len(payload), FrameType.DATA, flags, stream_id) + payload + + +def build_rst_stream(stream_id: int, error_code: int = 0) -> bytes: + payload = struct.pack("!I", error_code) + return frame_header(4, FrameType.RST_STREAM, 0, stream_id) + payload + + +def build_goaway(last_stream_id: int, error_code: int, extra: bytes = b"") -> bytes: + payload = struct.pack("!I I", last_stream_id, error_code) + extra + return frame_header(len(payload), FrameType.GOAWAY, 0, 0) + payload + + +def build_window_update(stream_id: int, increment: int) -> bytes: + payload = struct.pack("!I", increment) + return frame_header(4, FrameType.WINDOW_UPDATE, 0, stream_id) + payload + + +def build_ping(ack: bool = False, opaque: bytes = b"\x00" * 8) -> bytes: + flags = FlagPing.ACK if ack else 0 + return frame_header(8, FrameType.PING, flags, 0) + opaque + + +@pytest.mark.asyncio +async def test_incomplete_frame(connection: Tuple[Http2Connection, MagicMock]) -> None: + conn, _ = connection + frame = b"111111111" + conn.data_received(frame) + assert conn._frame_buffer == frame + + +# ====================================================================== +# UNIT TESTS +# ====================================================================== + + +# ---------------------------------------------------------------------- +# 1. Protocol compliance (black‑box, frame‑by‑frame) +# ---------------------------------------------------------------------- +class TestProtocolCompliance: + @pytest.mark.asyncio + async def test_receive_settings_updates_remote_and_acks( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + conn, transport = connection + # Send server SETTINGS (HEADER_TABLE_SIZE=8192, MAX_CONCURRENT_STREAMS=50) + frame = build_settings_frame( + [ + (Setting.HEADER_TABLE_SIZE, 8192), + (Setting.MAX_CONCURRENT_STREAMS, 50), + ] + ) + conn.data_received(frame) + assert conn.remote_settings[Setting.HEADER_TABLE_SIZE] == 8192 + assert conn.remote_settings[Setting.MAX_CONCURRENT_STREAMS] == 50 + # Must have sent an ACK + assert any( + call[0][0][3:4] == FrameType.SETTINGS.to_bytes(1, "big") + and call[0][0][4] & FlagSettings.ACK + for call in transport.write.call_args_list + ) + + @pytest.mark.asyncio + async def test_receive_headers_for_new_stream( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + conn, _ = connection + # Create a stream from client side first (simulate request sent) + stream = await conn.create_stream() + await conn.send_request(stream, "GET", url_mock(), []) + stream.state = StreamState.OPEN # skip real send, just set state + + # Send response HEADERS with END_STREAM + headers = [(":status", "200"), ("content-type", "text/plain")] + frame = build_headers_frame(stream.stream_id, headers, end_stream=True) + conn.data_received(frame) + + # Verify stream received response + assert stream.response_future.done() + sid, resp_headers, body = stream.response_future.result() + assert sid == stream.stream_id + assert dict(resp_headers)[":status"] == "200" + assert body == b"" + + @pytest.mark.asyncio + async def test_receive_data_flow_control( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + conn, transport = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN # assume request already sent + + # Send DATA with some bytes + data = b"hello" + frame = build_data_frame(stream.stream_id, data, end_stream=False) + initial_window = conn.session_inbound_window + conn.data_received(frame) + assert conn.session_inbound_window == initial_window - len(data) + # Should trigger WINDOW_UPDATE when below threshold (32768) + # Because initial window is 65535 and we just consumed 5, still above threshold + assert not any( + b"WINDOW_UPDATE" in call.args[0] for call in transport.write.call_args_list + ) + + # Send more data to drop below 32768 + big_data = b"x" * 40000 + frame2 = build_data_frame(stream.stream_id, big_data, end_stream=False) + conn.data_received(frame2) + # Now session window should have triggered an update + updates = [ + call.args[0] + for call in transport.write.call_args_list + if FrameType.WINDOW_UPDATE.to_bytes(1, "big") in call.args[0] + ] + assert len(updates) >= 1 + + @pytest.mark.asyncio + async def test_rst_stream_handling( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + + frame = build_rst_stream(stream.stream_id, error_code=0x8) # CANCEL + conn.data_received(frame) + + assert stream.state == StreamState.CLOSED + assert stream.response_future.done() + with pytest.raises(RuntimeError): + stream.response_future.result() + assert stream.stream_id not in conn.streams + + @pytest.mark.asyncio + async def test_goaway_cancels_higher_streams( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + conn, _ = connection + # create three streams (1,3,5) + s1 = await conn.create_stream() + s3 = await conn.create_stream() + s5 = await conn.create_stream() + s1.state = s3.state = s5.state = StreamState.OPEN + + # GOAWAY with last_stream_id = 3 + frame = build_goaway(last_stream_id=3, error_code=0) + conn.data_received(frame) + + # s1 (1) and s3 (3) should be unaffected, s5 (5) cancelled + assert s1.stream_id in conn.streams + assert s3.stream_id in conn.streams + assert s5.stream_id not in conn.streams + assert s5.response_future.exception() is not None + + @pytest.mark.asyncio + async def test_ping_ack( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + conn, transport = connection + frame = build_ping(ack=False, opaque=b"12345678") + conn.data_received(frame) + # Expect ACK sent back with same data + acks: List[bytes] = [] + for call in transport.write.call_args_list: + arg = call.args[0] + if FrameType.PING.to_bytes(1, "big") in arg and arg[4] & FlagPing.ACK: + acks.append(arg) + assert len(acks) == 1 + assert b"12345678" in acks[0] + + @pytest.mark.asyncio + async def test_max_concurrent_streams_blocking( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + conn, _ = connection + conn.max_concurrent_streams = 1 + s1 = await conn.create_stream() + # second stream should block + create_task = asyncio.ensure_future(conn.create_stream()) + await asyncio.sleep(0.01) + assert not create_task.done() + # close s1 to release slot + conn._close_stream(s1) + s2 = await create_task + assert s2.stream_id > s1.stream_id + assert len(conn.streams) == 1 + + @pytest.mark.asyncio + async def test_unknown_frame_ignored( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + conn, _ = connection + # send frame type 0x1a (unused) + frame = frame_header(0, 0x1A, 0, 0) + conn.data_received(frame) + # Should not raise, connection stays intact + assert not conn._goaway_sent + + @pytest.mark.asyncio + async def test_bad_hpack_triggers_protocol_error( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + conn, transport = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + # corrupted headers + payload = b"\xff\xff\xff" + frame = ( + frame_header( + len(payload), + FrameType.HEADERS, + FlagHeaders.END_HEADERS, + stream.stream_id, + ) + + payload + ) + conn.data_received(frame) + # Should have sent RST_STREAM and/or GOAWAY + rst = any( + call.args[0][3:4] == FrameType.RST_STREAM.to_bytes(1, "big") + for call in transport.write.call_args_list + ) + goaway = any( + call.args[0][3:4] == FrameType.GOAWAY.to_bytes(1, "big") + for call in transport.write.call_args_list + ) + assert rst or goaway + + @pytest.mark.asyncio + async def test_data_frame_with_padding( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """Cover DATA frame with PADDED flag.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + # padded data: pad length 1, data 'x', zero padding byte + frame = build_data_frame(stream.stream_id, b"x", pad=True) + conn.data_received(frame) + assert stream.response_data == b"x" + + @pytest.mark.asyncio + async def test_headers_frame_with_priority( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """Cover HEADERS frame with PRIORITY flag.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + # priority block: exclusive (1 byte) + dependency (4 bytes) + weight (1 byte) + priority_data = b"\x00\x00\x00\x00\x10" + headers = [(":status", "200")] + frame = build_headers_frame( + stream.stream_id, + headers, + end_headers=True, + end_stream=True, + priority=priority_data, + ) + conn.data_received(frame) + assert stream.response_future.done() + assert stream.response_headers is not None + + @pytest.mark.asyncio + async def test_rst_stream_for_unknown_stream( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """RST_STREAM on an unknown stream ID is silently ignored.""" + conn, _ = connection + frame = build_rst_stream(999, error_code=0) + conn.data_received(frame) # must not raise + + @pytest.mark.asyncio + async def test_rst_stream_when_future_done( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """RST_STREAM when response future already completed (else branch of line 256).""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + stream.response_future.set_result((stream.stream_id, [], b"")) + frame = build_rst_stream(stream.stream_id, error_code=0) + conn.data_received(frame) + assert stream.state == StreamState.CLOSED + + @pytest.mark.asyncio + async def test_receive_settings_ack( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """Receive SETTINGS ACK (should be a no‑op).""" + conn, _ = connection + frame = build_settings_frame(ack=True) + conn.data_received(frame) # no crash + + @pytest.mark.asyncio + async def test_settings_on_nonzero_stream( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """SETTINGS frame on stream_id != 0 triggers GOAWAY.""" + conn, transport = connection + frame = frame_header(0, FrameType.SETTINGS, 0, 5) # stream 5 + conn.data_received(frame) + assert any( + call[0][0][3:4] == FrameType.GOAWAY.to_bytes(1, "big") + for call in transport.write.call_args_list + ) + + @pytest.mark.asyncio + async def test_settings_invalid_payload_length( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """SETTINGS payload not a multiple of 6 triggers protocol error.""" + conn, transport = connection + payload = b"\x00\x01\x02" # 3 bytes + frame = frame_header(len(payload), FrameType.SETTINGS, 0, 0) + payload + conn.data_received(frame) + assert any( + call[0][0][3:4] == FrameType.GOAWAY.to_bytes(1, "big") + for call in transport.write.call_args_list + ) + + @pytest.mark.asyncio + async def test_settings_initial_window_size_update( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """INITIAL_WINDOW_SIZE setting updates stream windows.""" + conn, _ = connection + stream = await conn.create_stream() + old_window = stream.outbound_window + frame = build_settings_frame([(Setting.INITIAL_WINDOW_SIZE, 131072)]) + conn.data_received(frame) + assert stream.outbound_window == old_window + (131072 - 65535) + + @pytest.mark.asyncio + async def test_settings_header_table_size( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """HEADER_TABLE_SIZE setting is processed.""" + conn, _ = connection + frame = build_settings_frame([(Setting.HEADER_TABLE_SIZE, 4096)]) + conn.data_received(frame) + # internal effect on encoder, just confirm no crash + + @pytest.mark.asyncio + async def test_ping_on_nonzero_stream( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """PING on non‑zero stream triggers GOAWAY.""" + conn, transport = connection + frame = frame_header(8, FrameType.PING, 0, 1) + b"\x00" * 8 + conn.data_received(frame) + assert any( + call[0][0][3:4] == FrameType.GOAWAY.to_bytes(1, "big") + for call in transport.write.call_args_list + ) + + @pytest.mark.asyncio + async def test_receive_ping_ack( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """Receiving a PING ACK is logged and does not cause another response.""" + conn, _ = connection + frame = build_ping(ack=True, opaque=b"12345678") + conn.data_received(frame) # no crash, no additional PING sent + + @pytest.mark.asyncio + async def test_goaway_when_future_already_done( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """GOAWAY should not fail when a stream's future is already complete.""" + conn, _ = connection + s1 = await conn.create_stream() + s3 = await conn.create_stream() + s1.state = s3.state = StreamState.OPEN + s1.response_future.set_result((s1.stream_id, [], b"")) + frame = build_goaway(last_stream_id=1, error_code=0) + conn.data_received(frame) + assert s1.stream_id in conn.streams + assert s3.stream_id not in conn.streams + + @pytest.mark.asyncio + async def test_window_update_for_unknown_stream( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """WINDOW_UPDATE on unknown stream must be ignored (no crash).""" + conn, _ = connection + frame = build_window_update(123, 100) + conn.data_received(frame) + assert conn.session_outbound_window == 65535 # unchanged + + @pytest.mark.asyncio + async def test_continuation_frame_ignored( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """CONTINUATION frame is ignored with a warning.""" + conn, _ = connection + frame = frame_header(0, FrameType.CONTINUATION, 0, 1) + conn.data_received(frame) # no crash + + @pytest.mark.asyncio + async def test_unknown_frame_type_ignored( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """Unknown frame type (>9) is ignored.""" + conn, _ = connection + frame = frame_header(0, 0x1A, 0, 0) + conn.data_received(frame) # no crash + + @pytest.mark.asyncio + async def test_send_data_end_stream_last_chunk( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """send_data with end_stream=True sets END_STREAM on the last chunk.""" + conn, transport = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + conn.session_outbound_window = 100 + stream.outbound_window = 100 + await conn.send_data(stream, b"x" * 10, end_stream=True) + data_frames = [ + call[0][0] + for call in transport.write.call_args_list + if FrameType.DATA.to_bytes(1, "big") in call[0][0] + ] + assert len(data_frames) == 1 + assert data_frames[0][4] & FlagData.END_STREAM + + @pytest.mark.asyncio + async def test_send_request_with_body( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """send_request with body sends HEADERS (no END_STREAM) followed by DATA.""" + conn, transport = connection + stream = await conn.create_stream() + await conn.send_request(stream, "POST", url_mock("/"), [], body=b"hello") + sent = [ + FrameType(call[0][0][3]).name + for call in transport.write.call_args_list + if call[0][0][3] in (FrameType.HEADERS, FrameType.DATA) + ] + assert sent == ["HEADERS", "DATA"] + + @pytest.mark.asyncio + async def test_create_stream_after_goaway( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """create_stream raises ConnectionError when GOAWAY has been sent.""" + conn, _ = connection + conn._goaway_sent = True + with pytest.raises(ConnectionError): + await conn.create_stream() + + @pytest.mark.asyncio + async def test_maybe_unblock_streams_done_future( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """_maybe_unblock_streams skips futures that are already done.""" + conn, _ = connection + fut = conn._loop.create_future() + fut.set_result(None) + conn._pending_streams.append(fut) + conn._maybe_unblock_streams() + assert len(conn.streams) == 0 # no new stream created + + @pytest.mark.asyncio + async def test_connection_lost_done_futures( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """connection_lost must handle streams whose futures are already done.""" + conn, _ = connection + stream = await conn.create_stream() + stream.response_future.set_result((stream.stream_id, [], b"")) + conn.connection_lost(None) # no exception + + +# ---------------------------------------------------------------------- +# 2. Miscellaneous tests (race conditions, deadlocks, edge cases) +# ---------------------------------------------------------------------- +class TestMiscellaneous: + @pytest.mark.asyncio + async def test_concurrent_send_data_does_not_deadlock( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """Multiple tasks sending data on the same stream should not deadlock.""" + conn, transport = connection + stream = await conn.create_stream() + # Set window large enough + conn.session_outbound_window = 1_000_000 + stream.outbound_window = 1_000_000 + + async def send_chunk() -> None: + await conn.send_data(stream, b"x" * 100, end_stream=False) + + tasks = [asyncio.create_task(send_chunk()) for _ in range(5)] + await asyncio.gather(*tasks, return_exceptions=True) + # All writes should complete eventually + assert transport.write.call_count >= 5 + + @pytest.mark.asyncio + async def test_window_update_wakes_all_waiters( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """When window is zero, multiple blocked tasks resume on WINDOW_UPDATE.""" + conn, transport = connection + stream = await conn.create_stream() + conn.session_outbound_window = 0 + stream.outbound_window = 0 + + async def blocked_send() -> None: + await conn.send_data(stream, b"hello", end_stream=False) + + task1 = asyncio.create_task(blocked_send()) + task2 = asyncio.create_task(blocked_send()) + await asyncio.sleep(0.01) # both waiting + + # Simulate WINDOW_UPDATE that opens 10 bytes + frame = build_window_update(0, 10) # session + conn.data_received(frame) + frame2 = build_window_update(stream.stream_id, 10) + conn.data_received(frame2) + await asyncio.sleep(0.01) + assert task1.done() + assert task2.done() + + @pytest.mark.asyncio + async def test_stream_cancelled_before_response( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """Pending create_stream futures are cancelled on connection loss.""" + conn, _ = connection + conn.max_concurrent_streams = 1 + await conn.create_stream() + # Queue a second stream + fut = asyncio.ensure_future(conn.create_stream()) + await asyncio.sleep(0.01) + assert not fut.done() + # Simulate connection loss + conn.connection_lost(ConnectionError("test")) + await asyncio.sleep(0.01) + assert fut.done() + with pytest.raises(ConnectionError): + fut.result() + + @pytest.mark.asyncio + async def test_close_stream_on_rst_without_headers( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """RST_STREAM before headers deliver must resolve future with error.""" + conn, _ = connection + stream = await conn.create_stream() + frame = build_rst_stream(stream.stream_id, error_code=0x8) + conn.data_received(frame) + assert stream.response_future.done() + with pytest.raises(RuntimeError): + stream.response_future.result() + + @pytest.mark.asyncio + async def test_data_received_without_connection( + self, protocol: Tuple[Http2Protocol, MagicMock] + ) -> None: + """Http2Protocol.data_received is a no‑op before connection_made.""" + proto, _ = protocol + proto._connection = None + proto.data_received(b"anything") # must not raise + + +# ---------------------------------------------------------------------- +# 3. Interface tests (high‑level features via Http2Protocol.send) +# ---------------------------------------------------------------------- +class TestHighLevelInterface: + @pytest.mark.asyncio + async def test_json_response( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """send() returns a Http2Response with correct JSON body.""" + proto, transport = protocol + + async def do_request() -> Http2Response: + return await proto.send( + "GET", url_mock("/json"), headers=[("accept", "application/json")] + ) + + task = asyncio.create_task(do_request()) + # Let it run until it creates a stream and sends HEADERS + await asyncio.sleep(0.01) + headers = [(":status", "200"), ("content-type", "application/json")] + body = b'{"key": "value"}' + resp_frame = build_headers_frame( + 1, headers, end_stream=False + ) + build_data_frame(1, body, end_stream=True) + proto.data_received(resp_frame) + response = await task + assert response.status == 200 + assert response.body == body + assert json.loads(response.body) == {"key": "value"} + + @pytest.mark.asyncio + async def test_redirect_response( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """302 redirect headers are correctly returned.""" + proto, transport = protocol + task = asyncio.create_task(proto.send("GET", url_mock("/redirect"), headers=[])) + await asyncio.sleep(0.01) + headers = [(":status", "302"), ("location", "/new")] + resp_frame = build_headers_frame(1, headers, end_stream=True) + proto.data_received(resp_frame) + resp = await task + assert resp.status == 302 + assert dict(resp.headers).get("location") == "/new" + + @pytest.mark.asyncio + async def test_compressed_body_delivery( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """Response with content-encoding: gzip delivers raw compressed bytes.""" + proto, transport = protocol + task = asyncio.create_task(proto.send("GET", url_mock("/gzip"), headers=[])) + await asyncio.sleep(0.01) + raw_data = gzip.compress(b"uncompressed") + headers = [(":status", "200"), ("content-encoding", "gzip")] + frames = build_headers_frame(1, headers, end_stream=False) + build_data_frame( + 1, raw_data, end_stream=True + ) + proto.data_received(frames) + resp = await task + assert resp.body == b"uncompressed" + + @pytest.mark.asyncio + async def test_multiple_requests_concurrently( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """Multiple send() calls create distinct streams and receive responses.""" + proto, transport = protocol + tasks = [] + for i in range(2): + tasks.append(asyncio.create_task(proto.send("GET", url_mock(f"/r{i}"), []))) + await asyncio.sleep(0.01) + + # Stream 1 and 3 should have been created + # Respond to each + resp1 = build_headers_frame(1, [(":status", "200")], end_stream=True) + resp3 = build_headers_frame(3, [(":status", "201")], end_stream=True) + proto.data_received(resp1) + proto.data_received(resp3) + + r1, r2 = await asyncio.gather(*tasks) + assert r1.status == 200 + assert r2.status == 201 + + @pytest.mark.asyncio + async def test_response_all_methods(self) -> None: + """Cover Http2Response.read, .text, .json, .cookies, .raise_for_status, .release, .close, context manager.""" + headers = [ + (":status", "200"), + ("set-cookie", "a=b"), + ("content-type", "application/json"), + ] + body = b'{"ok":true}' + resp = Http2Response(headers, body, method="GET", url=url_mock("/")) + assert resp.status == 200 + assert await resp.read() == body + assert await resp.text() == '{"ok":true}' + assert await resp.json() == {"ok": True} + assert "a" in resp.cookies + resp.raise_for_status() # 200 is ok + resp.release() + resp.close() + async with resp: + pass + + @pytest.mark.asyncio + async def test_response_raise_for_status_error(self) -> None: + """Http2Response.raise_for_status raises for 4xx.""" + headers = [(":status", "404")] + resp = Http2Response(headers, b"", method="GET", url=url_mock("/")) + from aiohttp.client_exceptions import ClientResponseError + + with pytest.raises(ClientResponseError): + resp.raise_for_status() + + @pytest.mark.asyncio + async def test_send_with_body_through_protocol( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """Full send() with a body completes successfully.""" + proto, transport = protocol + url = url_mock("/upload") + task = asyncio.create_task(proto.send("POST", url, [], body=b"data")) + await asyncio.sleep(0.01) + resp_frame = build_headers_frame(1, [(":status", "200")], end_stream=True) + proto.data_received(resp_frame) + response = await task + assert response.status == 200 + + +class TestStreamStateMachine: + @pytest.mark.asyncio + async def test_invalid_transition_raises( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """Invalid state transition raises ProtocolError.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + with pytest.raises(ProtocolError): + stream.transition(StreamState.RESERVED_LOCAL) + + @pytest.mark.asyncio + async def test_receive_data_end_stream_half_closed_local( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """DATA END_STREAM when HALF_CLOSED_LOCAL -> CLOSED and stream removed.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.HALF_CLOSED_LOCAL + stream.response_headers = [(":status", "200")] + stream.receive_data(b"body", end_stream=True) + assert stream.state == StreamState.CLOSED + assert stream.stream_id not in conn.streams + + @pytest.mark.asyncio + async def test_receive_data_end_stream_invalid_state( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """DATA END_STREAM in CLOSED state raises ProtocolError.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.CLOSED + with pytest.raises(ProtocolError): + stream.receive_data(b"x", end_stream=True) + + @pytest.mark.asyncio + async def test_receive_headers_end_stream_half_closed_local( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """HEADERS END_STREAM when HALF_CLOSED_LOCAL -> CLOSED and stream removed.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.HALF_CLOSED_LOCAL + stream.receive_headers([(":status", "200")], end_stream=True) + assert stream.state == StreamState.CLOSED + assert stream.stream_id not in conn.streams + + @pytest.mark.asyncio + async def test_receive_headers_end_stream_invalid_state( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """HEADERS END_STREAM in CLOSED state raises ProtocolError.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.CLOSED + with pytest.raises(ProtocolError): + stream.receive_headers([(":status", "200")], end_stream=True) + + @pytest.mark.asyncio + async def test_data_stream_before_headers( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """DATA before headers does NOT set future until headers arrive.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + # double end stream is invalid + stream.receive_data(b"body", end_stream=False) + assert not stream.response_future.done() + stream.receive_headers([(":status", "200")], end_stream=True) + assert stream.response_future.done() + _, headers, body = stream.response_future.result() + assert body == b"body" + + @pytest.mark.asyncio + async def test_future_already_done_data_end_stream( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """receive_data with END_STREAM does not double‑set an already done future.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + stream.response_headers = [(":status", "200")] + stream.response_future.set_result( + (stream.stream_id, stream.response_headers, b"") + ) + stream.receive_data(b"more", end_stream=True) # no exception + + @pytest.mark.asyncio + async def test_future_already_done_headers_end_stream( + self, connection: Tuple[Http2Connection, MagicMock] + ) -> None: + """receive_headers with END_STREAM does not double‑set an already done future.""" + conn, _ = connection + stream = await conn.create_stream() + stream.state = StreamState.OPEN + stream.response_future.set_result((stream.stream_id, [], b"")) + stream.receive_headers([(":status", "200")], end_stream=True) # no exception + + +# ---------------------------------------------------------------------- +# Mock transport – records writes and lies about ALPN +# ---------------------------------------------------------------------- +class MockH2Transport(asyncio.Transport): + def __init__(self, extra_info: Optional[Dict[str, Any]] = None) -> None: + super().__init__() + self.written = bytearray() + self._closing = False + self._extra = extra_info or {} + self._protocol: Optional[Http2Protocol] = None + + def write(self, data: bytes | bytearray | memoryview) -> None: + self.written.extend(data) + + def close(self) -> None: + self._closing = True + + def is_closing(self) -> bool: + return self._closing + + def get_extra_info(self, name: str, default: Any = None) -> Any: + if name == "ssl_object": + return self._extra.get("ssl_object", MagicMock()) + return self._extra.get(name, default) + + +# ---------------------------------------------------------------------- +# Custom connector – always returns Http2Protocol for h2 connections +# ---------------------------------------------------------------------- +class H2TestConnector(TCPConnector): + def _get_protocol(self, loop: asyncio.AbstractEventLoop) -> type: + # Return the class; aiohttp will instantiate it + return Http2Protocol + + async def close(self, *, abort_ssl: bool = False) -> None: + self._closed = True + return None + + +# ---------------------------------------------------------------------- +# Fixture: session + mock transport + captured protocol +# ---------------------------------------------------------------------- +@pytest.fixture +async def h2_client() -> Any: # returns a generator of (session, transport, protocol) + """Create a ClientSession that uses our Http2Protocol over a mock transport.""" + # Mock SSL object that tells aiohttp we’ve negotiated h2 + mock_ssl = MagicMock() + mock_ssl.selected_alpn_protocol.return_value = "h2" + transport = MockH2Transport(extra_info={"ssl_object": mock_ssl}) + + protocol_instance: Optional[Http2Protocol] = None + + async def fake_create_connection( + protocol_factory: Any, *args: Any, **kwargs: Any + ) -> Tuple[MockH2Transport, Http2Protocol]: + nonlocal protocol_instance + protocol_instance = protocol_factory() # Http2Protocol() + protocol_instance.connection_made(transport) + transport._protocol = protocol_instance + return transport, protocol_instance + + connector = H2TestConnector() + connector._wrap_create_connection = fake_create_connection # type: ignore[assignment] + async with aiohttp.ClientSession(connector=connector) as session: + yield session, transport, protocol_instance + + +# if it's a real URL it hangs (missing mock?) +URL = "https://127.3.3.3" + + +class TestIncomingResponses: + @pytest.mark.asyncio + async def test_get_200_response(self, h2_client: Any) -> None: # type: ignore[misc] + session, transport, _ = h2_client + task = asyncio.create_task(session.get(URL)) + await asyncio.sleep(0.01) # request sent + + # Feed a minimal 200 response + hframe = build_headers_frame(1, [(":status", "200")], end_stream=True) + proto = transport._protocol + proto.data_received(hframe) + + resp = await task + assert resp.status == 200 + assert await resp.read() == b"" + + @pytest.mark.asyncio + async def test_response_with_body(self, h2_client: Any) -> None: # type: ignore[misc] + session, transport, _ = h2_client + task = asyncio.create_task(session.get(URL)) + await asyncio.sleep(0.01) + + # Send HEADERS (no END_STREAM) then DATA with body + hframe = build_headers_frame(1, [(":status", "200")], end_stream=False) + dframe = build_data_frame(1, b"Hello, h2!", end_stream=True) + proto = transport._protocol + proto.data_received(hframe) + proto.data_received(dframe) + + resp = await task + assert resp.status == 200 + assert await resp.text() == "Hello, h2!" + + @pytest.mark.asyncio + async def test_json_response(self, h2_client: Any) -> None: # type: ignore[misc] + session, transport, _ = h2_client + task = asyncio.create_task(session.get(URL)) + await asyncio.sleep(0.01) + + headers = [(":status", "200"), ("content-type", "application/json")] + body = b'{"key":"value"}' + proto = transport._protocol + proto.data_received(build_headers_frame(1, headers, end_stream=False)) + proto.data_received(build_data_frame(1, body, end_stream=True)) + + resp = await task + assert await resp.json() == {"key": "value"} + + @pytest.mark.asyncio + async def test_response_cookies(self, h2_client: Any) -> None: # type: ignore[misc] + session, transport, _ = h2_client + task = asyncio.create_task(session.get(URL)) + await asyncio.sleep(0.01) + + headers = [(":status", "200"), ("set-cookie", "session=abc123; Path=/")] + proto = transport._protocol + proto.data_received(build_headers_frame(1, headers, end_stream=True)) + + resp = await task + assert "session" in resp.cookies + assert resp.cookies["session"].value == "abc123" + + @pytest.mark.asyncio + async def test_concurrent_requests_mux(self, h2_client: Any) -> None: # type: ignore[misc] + session, transport, _ = h2_client + t1 = asyncio.create_task(session.get(URL)) + t2 = asyncio.create_task(session.get(URL)) + await asyncio.sleep(0.01) + + # Stream 1 gets response, stream 3 gets response + proto = transport._protocol + proto.data_received( + build_headers_frame(1, [(":status", "200")], end_stream=True) + ) + proto.data_received( + build_headers_frame(3, [(":status", "201")], end_stream=True) + ) + + r1, r2 = await asyncio.gather(t1, t2) + assert r1.status == 200 + assert r2.status == 201 + + @pytest.mark.asyncio + async def test_redirect_headers(self, h2_client: Any) -> None: # type: ignore[misc] + session, transport, _ = h2_client + task = asyncio.create_task(session.get(URL)) + await asyncio.sleep(0.01) + + headers = [(":status", "302"), ("location", "/new")] + proto = transport._protocol + proto.data_received(build_headers_frame(1, headers, end_stream=True)) + + await asyncio.sleep(0.01) + + headers = [(":status", "200")] + proto.data_received(build_headers_frame(3, headers, end_stream=True)) + + resp = await task + first = resp._history[0] + assert first.status == 302 + assert first.headers.get("location") == "/new" + + assert resp.status == 200 + + @pytest.mark.asyncio + async def test_error_response_raises(self, h2_client: Any) -> None: # type: ignore[misc] + session, transport, _ = h2_client + task = asyncio.create_task(session.get(URL)) + await asyncio.sleep(0.01) + + proto = transport._protocol + proto.data_received( + build_headers_frame(1, [(":status", "404")], end_stream=True) + ) + resp = await task + with pytest.raises(aiohttp.ClientResponseError): + resp.raise_for_status() + + +# ---------------------------------------------------------------------- +# Additional tests for full coverage of missing paths +# ---------------------------------------------------------------------- + + +class TestConnectionEdgeCases: + @pytest.mark.asyncio + async def test_eof_received_calls_close( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """EOF triggers connection close.""" + conn, transport = connection + result = conn.eof_received() + assert result is False + transport.close.assert_called_once() + + @pytest.mark.asyncio + async def test_protocol_eof_received( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """Http2Protocol.eof_received delegates to connection and returns False.""" + proto, transport = protocol + # connection is established + assert proto._connection is not None + result = proto.eof_received() + assert result is False + # transport.close should be called via conn.eof_received -> conn.close() + transport.close.assert_called_once() + + @pytest.mark.asyncio + async def test_data_frame_unknown_stream_above_last_peer( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """DATA frame for unknown stream_id > _last_peer_stream_id sends RST_STREAM.""" + conn, transport = connection + # last_peer_stream_id is initially 0 + unknown_id = 5 # > 0 + frame = build_data_frame(unknown_id, b"x", end_stream=False) + conn.data_received(frame) + # Should have sent RST_STREAM with PROTOCOL_ERROR (1) + rst_frames = [ + call.args[0] + for call in transport.write.call_args_list + if FrameType.RST_STREAM.to_bytes(1, "big") in call.args[0] + ] + assert len(rst_frames) == 1 + payload = rst_frames[0][9:] # after 9-byte header + error_code = struct.unpack("!I", payload)[0] + assert error_code == 1 # PROTOCOL_ERROR + + @pytest.mark.asyncio + async def test_data_frame_unknown_stream_not_above_last_peer( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """DATA frame for unknown stream_id <= _last_peer_stream_id is ignored.""" + conn, transport = connection + # artificially raise last_peer_stream_id + conn._last_peer_stream_id = 10 + frame = build_data_frame(5, b"x", end_stream=False) + conn.data_received(frame) + # No RST_STREAM sent + assert not any( + FrameType.RST_STREAM.to_bytes(1, "big") in call.args[0] + for call in transport.write.call_args_list + ) + + @pytest.mark.asyncio + async def test_send_data_end_stream_when_half_closed_remote( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """When stream is HALF_CLOSED_REMOTE, final DATA with END_STREAM closes it.""" + conn, transport = connection + stream = await conn.create_stream() + # Simulate remote half-close + stream.state = StreamState.HALF_CLOSED_REMOTE + conn.session_outbound_window = 1000 + stream.outbound_window = 1000 + await conn.send_data(stream, b"done", end_stream=True) + # Stream should be CLOSED and removed + assert stream.state == StreamState.CLOSED + assert stream.stream_id not in conn.streams + + @pytest.mark.asyncio + async def test_connection_lost_cancels_open_streams( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """connection_lost sets ConnectionError on all unfinished streams.""" + conn, _ = connection + s1 = await conn.create_stream() + s2 = await conn.create_stream() + s1.state = s2.state = StreamState.OPEN + # Both futures not done + conn.connection_lost(ConnectionError("test")) + assert s1.response_future.exception() is not None + assert s2.response_future.exception() is not None + # Also check pending streams cleared + assert len(conn._pending_streams) == 0 + + @pytest.mark.asyncio + async def test_connection_lost_clears_pending_futures( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """Pending create_stream futures are cancelled on connection loss.""" + conn, _ = connection + conn.max_concurrent_streams = 1 + await conn.create_stream() # fills one slot + fut = asyncio.ensure_future(conn.create_stream()) + await asyncio.sleep(0.01) + assert not fut.done() + conn.connection_lost(ConnectionError("test")) + await asyncio.sleep(0.01) + assert fut.done() + with pytest.raises(ConnectionError): + fut.result() + + @pytest.mark.asyncio + async def test_close_transport( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """close() calls transport.close().""" + conn, transport = connection + conn.close() + transport.close.assert_called_once() + + @pytest.mark.asyncio + async def test_should_close_reflects_goaway( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """should_close returns True if GOAWAY sent or received.""" + conn, _ = connection + assert not conn.should_close + conn._goaway_received = True + assert conn.should_close + + @pytest.mark.asyncio + async def test_is_connected( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """is_connected mirrors transport.is_closing().""" + conn, transport = connection + assert conn.is_connected() is True + transport.is_closing.return_value = True + assert conn.is_connected() is False + + @pytest.mark.asyncio + async def test_protocol_close_and_abort( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """Http2Protocol.close() and abort() propagate to transport.""" + proto, transport = protocol + proto.close() + transport.close.assert_called_once() + # abort is same as close + proto.abort() + assert transport.close.call_count == 2 + + @pytest.mark.asyncio + async def test_protocol_should_close_no_connection( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """should_close returns False when connection is None.""" + proto, _ = protocol + proto._connection = None + assert not proto.should_close + + @pytest.mark.asyncio + async def test_protocol_is_connected_no_connection( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """is_connected returns False when connection is None.""" + proto, _ = protocol + proto._connection = None + assert not proto.is_connected() + + @pytest.mark.asyncio + async def test_initiate_connection_sends_preface_and_settings( + self, mock_transport: MagicMock + ) -> None: + """initiate_connection writes HTTP/2 preface and initial SETTINGS.""" + loop = asyncio.get_running_loop() + conn = Http2Connection(mock_transport, loop) + conn.initiate_connection() + written = b"".join(call.args[0] for call in mock_transport.write.call_args_list) + assert b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" in written + assert FrameType.SETTINGS.to_bytes(1, "big") in written + + @pytest.mark.asyncio + async def test_create_stream_protocol_raises_without_connection( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """Http2Protocol.create_stream raises RuntimeError if no connection.""" + proto, _ = protocol + proto._connection = None + with pytest.raises(RuntimeError): + await proto.send("get", "", []) + + @pytest.mark.asyncio + async def test_send_protocol_raises_without_connection( + self, protocol: Tuple[Http2Protocol, MagicMock], mock_transport: MagicMock + ) -> None: + """Http2Protocol.send raises RuntimeError if no connection.""" + proto, _ = protocol + proto._connection = None + with pytest.raises(RuntimeError): + await proto.send("GET", url_mock(), []) + + @pytest.mark.asyncio + async def test_send_data_flow_control_releases_waiters( + self, connection: Tuple[Http2Connection, MagicMock], mock_transport: MagicMock + ) -> None: + """After a send, if window space remains, _flow_control_updated is set.""" + conn, transport = connection + stream = await conn.create_stream() + conn.session_outbound_window = 100 + stream.outbound_window = 100 + # set event cleared before wait + conn._flow_control_updated.clear() + await conn.send_data(stream, b"x" * 10, end_stream=False) + # After send, there is still window space, so event should be set + assert conn._flow_control_updated.is_set()