diff --git a/benchmark/README.md b/benchmark/README.md index d1678966..94ffe87b 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -28,7 +28,14 @@ git worktree add ../httpx2-main main && (cd ../httpx2-main && uv sync) scripts/benchmark --python main=../httpx2-main/.venv/bin/python --python work=.venv/bin/python --lib httpx2 ``` -`punkreq` is supported as an extra reference point (`--lib punkreq`) if it is installed. +The original `httpx` and `punkreq` are supported as reference points (`--lib httpx`, `--lib punkreq`). +Neither is part of the project's environment (installing `httpx` next to `httpx2` interferes with the +alias tests), so provision them in a separate interpreter and pass it with `--python`: + +``` +uv venv --python 3.14 /tmp/refs && uv pip install --python /tmp/refs/bin/python httpx punkreq zuvloop +scripts/benchmark --python refs=/tmp/refs/bin/python --lib httpx --lib punkreq +``` ## Reading the results diff --git a/benchmark/client.py b/benchmark/client.py index 5e18a13c..8399da7c 100644 --- a/benchmark/client.py +++ b/benchmark/client.py @@ -65,6 +65,19 @@ def check_size(received: int, expected: int) -> None: def build_httpx2(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn]: import httpx2 + return _build_httpx_like(httpx2, scenario, payload) + + +def build_httpx(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn]: + # The original httpx, for reference; it shares the httpx2 API. Not part of the + # project environment: see the README for how to provision it. + import httpx + + return _build_httpx_like(httpx, scenario, payload) + + +def _build_httpx_like(httpx2: Any, scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn]: + client = httpx2.AsyncClient( limits=httpx2.Limits( max_connections=scenario.max_connections, @@ -79,13 +92,15 @@ def build_httpx2(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn if scenario.post: headers.append(("content-length", str(len(payload)))) - class RequestBody(httpx2.AsyncByteStream): - async def __aiter__(self) -> AsyncIterator[bytes]: - if scenario.post: - yield payload + async def request_body(self: object) -> AsyncIterator[bytes]: + if scenario.post: + yield payload + + # Built dynamically so the helper can serve both httpx2 and httpx. + request_body_stream = type("RequestBody", (httpx2.AsyncByteStream,), {"__aiter__": request_body}) async def stream_one() -> None: - request = httpx2.Request(scenario.method, scenario.url, headers=headers, stream=RequestBody()) + request = httpx2.Request(scenario.method, scenario.url, headers=headers, stream=request_body_stream()) response = await client.send(request, stream=True, follow_redirects=False) received = 0 async for chunk in response.aiter_raw(scenario.chunk_size): @@ -198,6 +213,7 @@ async def read_one() -> None: BUILDERS: dict[str, Builder] = { "httpx2": build_httpx2, + "httpx": build_httpx, "httpcore2": build_httpcore2, "aiohttp": build_aiohttp, "punkreq": build_punkreq, diff --git a/scripts/unasync.py b/scripts/unasync.py index 908662b0..43e28ac3 100644 --- a/scripts/unasync.py +++ b/scripts/unasync.py @@ -11,6 +11,7 @@ ), ("import trio as concurrency", "from tests.httpcore2 import concurrency"), ("anyio.sleep", "concurrency.sleep"), + ("BACKENDS = \\[None, httpcore2.AnyIOBackend\\(\\)\\]", "BACKENDS = [None]"), ("AsyncIterator", "Iterator"), ("Async([A-Z][A-Za-z0-9_]*)", r"\2"), ("async def", "def"), diff --git a/src/httpcore2/httpcore2/__init__.py b/src/httpcore2/httpcore2/__init__.py index 708eea9a..0f279f9b 100644 --- a/src/httpcore2/httpcore2/__init__.py +++ b/src/httpcore2/httpcore2/__init__.py @@ -10,6 +10,7 @@ AsyncHTTPProxy, AsyncSOCKSProxy, ) +from ._backends.asyncio import AsyncioBackend from ._backends.base import ( SOCKET_OPTION, AsyncNetworkBackend, @@ -99,6 +100,7 @@ def __init__(self, *args, **kwargs): # type: ignore # network backends, implementations "SyncBackend", "AnyIOBackend", + "AsyncioBackend", "TrioBackend", # network backends, mock implementations "AsyncMockBackend", diff --git a/src/httpcore2/httpcore2/_backends/asyncio.py b/src/httpcore2/httpcore2/_backends/asyncio.py new file mode 100644 index 00000000..0d2b60a8 --- /dev/null +++ b/src/httpcore2/httpcore2/_backends/asyncio.py @@ -0,0 +1,299 @@ +from __future__ import annotations + +import asyncio +import collections +import inspect +import ssl +import typing + +from .._exceptions import ( + ConnectError, + ConnectTimeout, + ReadError, + ReadTimeout, + WriteError, + WriteTimeout, +) +from .base import SOCKET_OPTION, AsyncNetworkBackend, AsyncNetworkStream + +# Stop reading from the socket once this much data is buffered but unread, +# and start again once the buffer drains below the low-water mark. +RECEIVE_HIGH_WATER = 256 * 1024 +RECEIVE_LOW_WATER = 64 * 1024 + +# Stagger connection attempts across resolved addresses, as in RFC 8305. +HAPPY_EYEBALLS_DELAY = 0.25 +# The event loop insists on a TLS handshake timeout; this stands in for "none". +NO_HANDSHAKE_TIMEOUT = 365 * 24 * 60 * 60.0 + +_happy_eyeballs_support: dict[type[asyncio.AbstractEventLoop], bool] = {} + + +def _connection_kwargs(loop: asyncio.AbstractEventLoop) -> dict[str, typing.Any]: + # Not every event loop implements Happy Eyeballs; use it where available. + supported = _happy_eyeballs_support.get(type(loop)) + if supported is None: + try: + supported = "happy_eyeballs_delay" in inspect.signature(loop.create_connection).parameters + except (TypeError, ValueError): + # Some extension-implemented loops expose no signature to inspect. + supported = False + _happy_eyeballs_support[type(loop)] = supported + return {"happy_eyeballs_delay": HAPPY_EYEBALLS_DELAY} if supported else {} + + +class _Timeout(Exception): + """ + Raised on a waiter future when its deadline passes. + """ + + +def _timeout(waiter: asyncio.Future[None]) -> None: + if not waiter.done(): + waiter.set_exception(_Timeout()) + + +def _wake(waiter: asyncio.Future[None] | None) -> None: + if waiter is not None and not waiter.done(): + waiter.set_result(None) + + +class AsyncioStreamProtocol(asyncio.Protocol): + """ + Buffers received data for `AsyncioStream`, applying backpressure to the + transport once too much is buffered, and wakes up pending reads and writes. + """ + + def __init__(self) -> None: + self.transport: asyncio.Transport | None = None + self.chunks: collections.deque[bytes] = collections.deque() + self.buffered = 0 + self.reading_paused = False + self.writing_paused = False + self.eof = False + self.closed = False + self.exception: Exception | None = None + self.read_waiter: asyncio.Future[None] | None = None + self.write_waiter: asyncio.Future[None] | None = None + + def connection_made(self, transport: asyncio.BaseTransport) -> None: + # The transport implements the interface without necessarily subclassing it. + self.transport = typing.cast(asyncio.Transport, transport) + + def data_received(self, data: bytes) -> None: + self.chunks.append(data) + self.buffered += len(data) + if self.buffered >= RECEIVE_HIGH_WATER and not self.reading_paused: + assert self.transport is not None + self.transport.pause_reading() + self.reading_paused = True + _wake(self.read_waiter) + + def eof_received(self) -> bool: + self.eof = True + _wake(self.read_waiter) + # Let the transport close: an HTTP peer that has sent a FIN is done. + return False + + def connection_lost(self, exc: Exception | None) -> None: + self.closed = True + self.exception = exc + _wake(self.read_waiter) + _wake(self.write_waiter) + + def pause_writing(self) -> None: + self.writing_paused = True + + def resume_writing(self) -> None: + self.writing_paused = False + _wake(self.write_waiter) + + +class AsyncioStream(AsyncNetworkStream): + def __init__(self, transport: asyncio.Transport, protocol: AsyncioStreamProtocol) -> None: + self._transport = transport + self._protocol = protocol + + async def read(self, max_bytes: int, timeout: float | None = None) -> bytes: + protocol = self._protocol + if not protocol.chunks: + if protocol.eof or protocol.closed: + return self._read_at_end() + try: + await self._wait("read", timeout) + except _Timeout: + raise ReadTimeout("timed out") from None + if not protocol.chunks: + return self._read_at_end() + + chunk = protocol.chunks[0] + if len(chunk) <= max_bytes: + protocol.chunks.popleft() + else: + protocol.chunks[0] = chunk[max_bytes:] + chunk = chunk[:max_bytes] + protocol.buffered -= len(chunk) + if protocol.reading_paused and protocol.buffered <= RECEIVE_LOW_WATER: + protocol.reading_paused = False + self._transport.resume_reading() + return chunk + + def _read_at_end(self) -> bytes: + # No buffered data and no more coming: a clean EOF reads as empty, + # a connection dropped by an error is a read error. + if self._protocol.exception is not None: + raise ReadError(str(self._protocol.exception)) from self._protocol.exception + return b"" + + async def write(self, buffer: bytes, timeout: float | None = None) -> None: + if not buffer: + return + protocol = self._protocol + if protocol.closed or self._transport.is_closing(): + raise WriteError("Connection closed") + self._transport.write(buffer) + if protocol.writing_paused: + # The transport's send buffer is full; wait for it to drain. + try: + await self._wait("write", timeout) + except _Timeout: + raise WriteTimeout("timed out") from None + if protocol.closed: + raise WriteError(str(protocol.exception or "Connection closed")) + + async def _wait(self, kind: str, timeout: float | None) -> None: + loop = asyncio.get_running_loop() + waiter: asyncio.Future[None] = loop.create_future() + protocol = self._protocol + if kind == "read": + protocol.read_waiter = waiter + else: + protocol.write_waiter = waiter + handle = None if timeout is None else loop.call_later(timeout, _timeout, waiter) + try: + await waiter + finally: + if handle is not None: + handle.cancel() + if kind == "read": + protocol.read_waiter = None + else: + protocol.write_waiter = None + + async def aclose(self) -> None: + if self._protocol.closed: + return + self._transport.close() + # Closing only schedules the socket close on the event loop. Yield once + # so it runs now, then force it if unsent data is still holding it up. + await asyncio.sleep(0) + if not self._protocol.closed: + self._transport.abort() + + async def start_tls( + self, + ssl_context: ssl.SSLContext, + server_hostname: str | None = None, + timeout: float | None = None, + ) -> AsyncNetworkStream: + protocol = self._protocol + if protocol.chunks or protocol.eof or protocol.closed: + # Nothing may arrive before the handshake: anything already buffered + # is plaintext that must not be mistaken for data received over TLS. + await self.aclose() + raise ConnectError("Received unexpected data before the TLS handshake") + + loop = asyncio.get_running_loop() + # The loop reports its own handshake timeout as a connection error, so + # the deadline is applied here to raise a timeout, with the loop's own + # deadline kept out of the way. + handshake = loop.start_tls( + self._transport, + protocol, + ssl_context, + server_hostname=server_hostname, + ssl_handshake_timeout=NO_HANDSHAKE_TIMEOUT if timeout is None else timeout + 1.0, + ) + try: + transport = await asyncio.wait_for(handshake, timeout) + except (TimeoutError, asyncio.TimeoutError): + self._transport.close() + raise ConnectTimeout("timed out") from None + except (OSError, ssl.SSLError) as exc: + self._transport.close() + raise ConnectError(str(exc)) from exc + if transport is None: # pragma: no cover + raise ConnectError("TLS handshake failed") + protocol.transport = transport + return AsyncioStream(transport, protocol) + + def get_extra_info(self, info: str) -> typing.Any: + if info == "ssl_object": + return self._transport.get_extra_info("ssl_object") + if info == "client_addr": + return self._transport.get_extra_info("sockname") + if info == "server_addr": + return self._transport.get_extra_info("peername") + if info == "socket": + return self._transport.get_extra_info("socket") + if info == "is_readable": + # The event loop keeps reading while the connection is idle, so a + # FIN or stray data from the server is already known here without + # touching the socket. + protocol = self._protocol + return bool(protocol.chunks) or protocol.eof or protocol.closed + return None + + +class AsyncioBackend(AsyncNetworkBackend): + async def connect_tcp( + self, + host: str, + port: int, + timeout: float | None = None, + local_address: str | None = None, + socket_options: typing.Iterable[SOCKET_OPTION] | None = None, + ) -> AsyncNetworkStream: + loop = asyncio.get_running_loop() + local_addr = None if local_address is None else (local_address, 0) + # By default TCP sockets opened in `asyncio` include TCP_NODELAY. + connect = loop.create_connection( + AsyncioStreamProtocol, host, port, local_addr=local_addr, **_connection_kwargs(loop) + ) + return await self._connect(connect, timeout, socket_options) + + async def connect_unix_socket( + self, + path: str, + timeout: float | None = None, + socket_options: typing.Iterable[SOCKET_OPTION] | None = None, + ) -> AsyncNetworkStream: + loop = asyncio.get_running_loop() + connect = loop.create_unix_connection(AsyncioStreamProtocol, path) + return await self._connect(connect, timeout, socket_options) + + async def _connect( + self, + connect: typing.Coroutine[typing.Any, typing.Any, tuple[asyncio.BaseTransport, AsyncioStreamProtocol]], + timeout: float | None, + socket_options: typing.Iterable[SOCKET_OPTION] | None, + ) -> AsyncioStream: + try: + transport, protocol = await asyncio.wait_for(connect, timeout) + except (TimeoutError, asyncio.TimeoutError): + raise ConnectTimeout("timed out") from None + except OSError as exc: + raise ConnectError(str(exc)) from exc + stream = AsyncioStream(typing.cast(asyncio.Transport, transport), protocol) + if socket_options: + sock = transport.get_extra_info("socket") + try: + for option in socket_options: + sock.setsockopt(*option) + except OSError as exc: + await stream.aclose() + raise ConnectError(str(exc)) from exc + return stream + + async def sleep(self, seconds: float) -> None: + await asyncio.sleep(seconds) diff --git a/src/httpcore2/httpcore2/_backends/auto.py b/src/httpcore2/httpcore2/_backends/auto.py index c946c9e6..ea92817a 100644 --- a/src/httpcore2/httpcore2/_backends/auto.py +++ b/src/httpcore2/httpcore2/_backends/auto.py @@ -15,9 +15,9 @@ async def _init_backend(self) -> None: self._backend: AsyncNetworkBackend = TrioBackend() else: - from .anyio import AnyIOBackend + from .asyncio import AsyncioBackend - self._backend = AnyIOBackend() + self._backend = AsyncioBackend() async def connect_tcp( self, diff --git a/tests/httpcore2/_async/test_integration.py b/tests/httpcore2/_async/test_integration.py index 1325c734..8a7f12a9 100644 --- a/tests/httpcore2/_async/test_integration.py +++ b/tests/httpcore2/_async/test_integration.py @@ -5,30 +5,36 @@ import httpcore2 +# The automatic backend, and the explicit anyio backend it no longer selects under asyncio. +BACKENDS = [None, httpcore2.AnyIOBackend()] + @pytest.mark.anyio -async def test_request(httpbin: Server) -> None: - async with httpcore2.AsyncConnectionPool() as pool: +@pytest.mark.parametrize("network_backend", BACKENDS) +async def test_request(httpbin: Server, network_backend: httpcore2.AsyncNetworkBackend | None) -> None: + async with httpcore2.AsyncConnectionPool(network_backend=network_backend) as pool: response = await pool.request("GET", httpbin.url) assert response.status == 200 @pytest.mark.anyio -async def test_ssl_request(httpbin_secure: Server) -> None: +@pytest.mark.parametrize("network_backend", BACKENDS) +async def test_ssl_request(httpbin_secure: Server, network_backend: httpcore2.AsyncNetworkBackend | None) -> None: ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE - async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context) as pool: + async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool: response = await pool.request("GET", httpbin_secure.url) assert response.status == 200 @pytest.mark.anyio -async def test_extra_info(httpbin_secure: Server) -> None: +@pytest.mark.parametrize("network_backend", BACKENDS) +async def test_extra_info(httpbin_secure: Server, network_backend: httpcore2.AsyncNetworkBackend | None) -> None: ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE - async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context) as pool: + async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool: async with pool.stream("GET", httpbin_secure.url) as response: assert response.status == 200 stream = response.extensions["network_stream"] diff --git a/tests/httpcore2/_sync/test_integration.py b/tests/httpcore2/_sync/test_integration.py index 06f1aafb..2cf9d9b0 100644 --- a/tests/httpcore2/_sync/test_integration.py +++ b/tests/httpcore2/_sync/test_integration.py @@ -5,30 +5,36 @@ import httpcore2 +# The automatic backend, and the explicit anyio backend it no longer selects under asyncio. +BACKENDS = [None] -def test_request(httpbin: Server) -> None: - with httpcore2.ConnectionPool() as pool: + +@pytest.mark.parametrize("network_backend", BACKENDS) +def test_request(httpbin: Server, network_backend: httpcore2.NetworkBackend | None) -> None: + with httpcore2.ConnectionPool(network_backend=network_backend) as pool: response = pool.request("GET", httpbin.url) assert response.status == 200 -def test_ssl_request(httpbin_secure: Server) -> None: +@pytest.mark.parametrize("network_backend", BACKENDS) +def test_ssl_request(httpbin_secure: Server, network_backend: httpcore2.NetworkBackend | None) -> None: ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE - with httpcore2.ConnectionPool(ssl_context=ssl_context) as pool: + with httpcore2.ConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool: response = pool.request("GET", httpbin_secure.url) assert response.status == 200 -def test_extra_info(httpbin_secure: Server) -> None: +@pytest.mark.parametrize("network_backend", BACKENDS) +def test_extra_info(httpbin_secure: Server, network_backend: httpcore2.NetworkBackend | None) -> None: ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE - with httpcore2.ConnectionPool(ssl_context=ssl_context) as pool: + with httpcore2.ConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool: with pool.stream("GET", httpbin_secure.url) as response: assert response.status == 200 stream = response.extensions["network_stream"] diff --git a/tests/httpcore2/test_asyncio_backend.py b/tests/httpcore2/test_asyncio_backend.py new file mode 100644 index 00000000..d74aeec0 --- /dev/null +++ b/tests/httpcore2/test_asyncio_backend.py @@ -0,0 +1,360 @@ +from __future__ import annotations + +import asyncio +import contextlib +import socket +import ssl +import sys +import typing +from collections.abc import AsyncIterator, Awaitable, Callable + +import pytest +import trustme + +import httpcore2 +from httpcore2._backends.asyncio import ( + RECEIVE_HIGH_WATER, + RECEIVE_LOW_WATER, + AsyncioStream, + AsyncioStreamProtocol, +) + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def anyio_backend() -> str: + return "asyncio" + + +Handler = Callable[[asyncio.StreamReader, asyncio.StreamWriter], Awaitable[None]] + + +async def echo(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + # The client may drop the connection mid-way, for example when a TLS handshake fails. + with contextlib.suppress(ConnectionError, ssl.SSLError): + while data := await reader.read(65536): + writer.write(data) + await writer.drain() + writer.close() + + +async def silent(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + # Hold the connection open, without answering, until the client goes away. + await reader.read() + writer.close() + + +@pytest.fixture +async def serve() -> AsyncIterator[Callable[..., Awaitable[int]]]: + servers: list[asyncio.base_events.Server] = [] + + async def start(handler: Handler = echo, **kwargs: typing.Any) -> int: + server = await asyncio.start_server(handler, "127.0.0.1", 0, **kwargs) + servers.append(server) + port: int = server.sockets[0].getsockname()[1] + return port + + yield start + for server in servers: + server.close() + await server.wait_closed() + + +async def test_tcp_roundtrip_and_extra_info(serve: Callable[..., Awaitable[int]]) -> None: + port = await serve() + backend = httpcore2.AsyncioBackend() + stream = await backend.connect_tcp("127.0.0.1", port, socket_options=[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]) + assert stream.get_extra_info("socket").getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) == 1 + assert stream.get_extra_info("client_addr")[0] == "127.0.0.1" + assert stream.get_extra_info("server_addr") == ("127.0.0.1", port) + assert stream.get_extra_info("ssl_object") is None + assert stream.get_extra_info("invalid") is None + assert not stream.get_extra_info("is_readable") + + await stream.write(b"hello") + assert await stream.read(1024) == b"hello" + assert not stream.get_extra_info("is_readable") + + await stream.aclose() + assert stream.get_extra_info("is_readable") + + +async def test_read_at_eof(serve: Callable[..., Awaitable[int]]) -> None: + async def close_after_hello(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + writer.write(b"hello") + await writer.drain() + writer.close() + await writer.wait_closed() + + port = await serve(close_after_hello) + stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port) + assert await stream.read(1024) == b"hello" + assert await stream.read(1024) == b"" + assert await stream.read(1024) == b"" + await stream.aclose() + + +async def test_read_timeout(serve: Callable[..., Awaitable[int]]) -> None: + port = await serve(silent) + stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port) + with pytest.raises(httpcore2.ReadTimeout): + await stream.read(1024, timeout=0.05) + await stream.aclose() + + +async def test_invalid_socket_option(serve: Callable[..., Awaitable[int]]) -> None: + port = await serve() + with pytest.raises(httpcore2.ConnectError): + await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port, socket_options=[(socket.SOL_SOCKET, -1, 1)]) + + +async def test_connect_refused() -> None: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + with pytest.raises(httpcore2.ConnectError): + await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port) + + +async def test_connect_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + async def never_connects(*args: typing.Any, **kwargs: typing.Any) -> typing.Any: + await asyncio.sleep(10) + + monkeypatch.setattr(asyncio.get_running_loop(), "create_connection", never_connects) + with pytest.raises(httpcore2.ConnectTimeout): + await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", 1, timeout=0.05) + + +@pytest.mark.skipif(sys.platform == "win32", reason="Unix sockets are not available on Windows.") +async def test_unix_socket(tmp_path: typing.Any) -> None: + path = str(tmp_path / "socket") + server = await asyncio.start_unix_server(echo, path) + try: + stream = await httpcore2.AsyncioBackend().connect_unix_socket(path) + await stream.write(b"hello") + assert await stream.read(1024) == b"hello" + await stream.aclose() + finally: + server.close() + await server.wait_closed() + + with pytest.raises(httpcore2.ConnectError): + await httpcore2.AsyncioBackend().connect_unix_socket(str(tmp_path / "missing")) + + +async def test_start_tls(serve: Callable[..., Awaitable[int]]) -> None: + ca = trustme.CA() + server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + ca.issue_cert("localhost").configure_cert(server_context) + client_context = ssl.create_default_context() + ca.configure_trust(client_context) + + port = await serve(ssl=server_context) + stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port) + tls_stream = await stream.start_tls(client_context, server_hostname="localhost", timeout=5.0) + assert tls_stream.get_extra_info("ssl_object").version() == "TLSv1.3" + await tls_stream.write(b"hello") + assert await tls_stream.read(1024) == b"hello" + await tls_stream.aclose() + + +async def test_start_tls_rejects_data_received_before_the_handshake(serve: Callable[..., Awaitable[int]]) -> None: + async def banner(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + writer.write(b"220 hello\r\n") + await writer.drain() + await reader.read() + writer.close() + + port = await serve(banner) + stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port) + # Give the banner time to arrive and be buffered. + await asyncio.sleep(0.05) + with pytest.raises(httpcore2.ConnectError): + await stream.start_tls(ssl.create_default_context(), server_hostname="localhost", timeout=5.0) + + +async def test_happy_eyeballs_is_used_where_supported() -> None: + from httpcore2._backends.asyncio import HAPPY_EYEBALLS_DELAY, _connection_kwargs + + assert _connection_kwargs(asyncio.get_running_loop()) == {"happy_eyeballs_delay": HAPPY_EYEBALLS_DELAY} + + class MinimalLoop: + async def create_connection(self, protocol_factory: typing.Any, host: str, port: int) -> None: ... + + loop = typing.cast(asyncio.AbstractEventLoop, MinimalLoop()) + assert _connection_kwargs(loop) == {} + assert _connection_kwargs(loop) == {} + + +async def test_happy_eyeballs_detection_tolerates_loops_without_signatures(monkeypatch: pytest.MonkeyPatch) -> None: + import inspect + + from httpcore2._backends.asyncio import _connection_kwargs + + class OpaqueLoop: + def create_connection(self, *args: typing.Any, **kwargs: typing.Any) -> None: ... + + def no_signature(callable: typing.Any) -> typing.Any: + raise ValueError("no signature found") + + monkeypatch.setattr(inspect, "signature", no_signature) + assert _connection_kwargs(typing.cast(asyncio.AbstractEventLoop, OpaqueLoop())) == {} + + +async def test_start_tls_failure(serve: Callable[..., Awaitable[int]]) -> None: + # A plain echo server answers the ClientHello with the ClientHello. + port = await serve() + stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port) + with pytest.raises(httpcore2.ConnectError): + await stream.start_tls(ssl.create_default_context(), server_hostname="localhost", timeout=5.0) + + +async def test_start_tls_timeout(serve: Callable[..., Awaitable[int]]) -> None: + port = await serve(silent) + stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port) + with pytest.raises(httpcore2.ConnectTimeout): + await stream.start_tls(ssl.create_default_context(), server_hostname="localhost", timeout=0.05) + + +async def test_sleep() -> None: + await httpcore2.AsyncioBackend().sleep(0) + + +# The remaining behaviour is driven through a fake transport, so the timing +# of the real network is not involved. + + +class FakeTransport(asyncio.Transport): + def __init__(self) -> None: + super().__init__() + self.written: list[bytes] = [] + self.reading_paused = False + self.closed = False + + def write(self, data: typing.Any) -> None: + self.written.append(bytes(data)) + + def is_closing(self) -> bool: + return self.closed + + def close(self) -> None: + self.closed = True + + def abort(self) -> None: + self.closed = True + + def pause_reading(self) -> None: + self.reading_paused = True + + def resume_reading(self) -> None: + self.reading_paused = False + + +def fake_stream() -> tuple[AsyncioStream, AsyncioStreamProtocol, FakeTransport]: + transport = FakeTransport() + protocol = AsyncioStreamProtocol() + protocol.connection_made(transport) + return AsyncioStream(transport, protocol), protocol, transport + + +async def test_read_splits_large_chunks() -> None: + stream, protocol, _ = fake_stream() + protocol.data_received(b"abcdef") + assert await stream.read(4) == b"abcd" + assert await stream.read(4) == b"ef" + assert protocol.buffered == 0 + + +async def test_backpressure_pauses_and_resumes_reading() -> None: + stream, protocol, transport = fake_stream() + protocol.data_received(b"x" * RECEIVE_HIGH_WATER) + assert transport.reading_paused + protocol.data_received(b"y") + assert transport.reading_paused + + await stream.read(RECEIVE_HIGH_WATER - RECEIVE_LOW_WATER - 1) + assert protocol.buffered == RECEIVE_LOW_WATER + 2 + assert transport.reading_paused + await stream.read(2) + assert not transport.reading_paused + + +async def test_pending_read_wakes_on_eof() -> None: + stream, protocol, _ = fake_stream() + read = asyncio.ensure_future(stream.read(1024)) + await asyncio.sleep(0) + assert protocol.eof_received() is False + assert await read == b"" + + +async def test_pending_read_wakes_on_connection_lost_with_error() -> None: + stream, protocol, _ = fake_stream() + read = asyncio.ensure_future(stream.read(1024)) + await asyncio.sleep(0) + protocol.connection_lost(OSError("reset")) + with pytest.raises(httpcore2.ReadError): + await read + with pytest.raises(httpcore2.ReadError): + await stream.read(1024) + + +async def test_connection_lost_without_error_reads_as_eof() -> None: + stream, protocol, _ = fake_stream() + protocol.data_received(b"tail") + protocol.connection_lost(None) + assert await stream.read(1024) == b"tail" + assert await stream.read(1024) == b"" + assert stream.get_extra_info("is_readable") + + +async def test_write() -> None: + stream, protocol, transport = fake_stream() + await stream.write(b"") + await stream.write(b"hello") + assert transport.written == [b"hello"] + + transport.close() + with pytest.raises(httpcore2.WriteError): + await stream.write(b"more") + + protocol.connection_lost(None) + with pytest.raises(httpcore2.WriteError): + await stream.write(b"more") + + +async def test_aclose_forces_a_stuck_close() -> None: + stream, protocol, transport = fake_stream() + # The fake transport never reports the connection as lost by itself. + await stream.aclose() + assert transport.closed + protocol.connection_lost(None) + await stream.aclose() + + +async def test_write_waits_for_the_transport_to_drain() -> None: + stream, protocol, transport = fake_stream() + protocol.pause_writing() + write = asyncio.ensure_future(stream.write(b"hello")) + await asyncio.sleep(0) + assert not write.done() + protocol.resume_writing() + await write + assert transport.written == [b"hello"] + + +async def test_write_timeout_while_draining() -> None: + stream, protocol, _ = fake_stream() + protocol.pause_writing() + with pytest.raises(httpcore2.WriteTimeout): + await stream.write(b"hello", timeout=0.01) + + +async def test_write_fails_when_the_connection_is_lost_while_draining() -> None: + stream, protocol, _ = fake_stream() + protocol.pause_writing() + write = asyncio.ensure_future(stream.write(b"hello")) + await asyncio.sleep(0) + protocol.connection_lost(OSError("reset")) + with pytest.raises(httpcore2.WriteError): + await write