From 8bf3dbfc52037ac4251e1abb04903222821a2e2b Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Thu, 22 Oct 2026 16:44:14 +0200 Subject: [PATCH 01/10] Add experimental asyncio client for CrateDB --- pyproject.toml | 17 +++++++++++++++- src/crate/client/_response.py | 1 + src/crate/client/aio/__init__.py | 35 ++++++++++++++++++++++++++++++++ 3 files changed, 52 insertions(+), 1 deletion(-) create mode 100644 src/crate/client/aio/__init__.py diff --git a/pyproject.toml b/pyproject.toml index 13bacc48..1dcc3794 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,10 +56,13 @@ dependencies = [ [dependency-groups] dev = [ + "aiohttp>=3.12,<4", "certifi", "coverage", "mypy<2.4", "pytest<10", + "pytest-asyncio<2", + "pytest-timeout<3", "pytz", "ruff<0.17", ] @@ -102,9 +105,16 @@ install_types = true namespace_packages = true non_interactive = true +[[tool.mypy.overrides]] +module = ["crate.client.aio.*", "crate.client._tls"] +disallow_untyped_defs = true +disallow_incomplete_defs = true + [tool.pytest.ini_options] addopts = "-rA --verbosity=3 --doctest-modules" +asyncio_mode = "strict" +asyncio_default_fixture_loop_scope = "function" doctest_optionflags = "NORMALIZE_WHITESPACE ELLIPSIS" minversion = "2.0" testpaths = [ @@ -148,6 +158,11 @@ lint.select = [ "YTT", ] +lint.extend-select = [ + # flake8-async: blocking calls in async functions + "ASYNC", +] + lint.extend-ignore = [ # Unnecessary variable assignment before `return` statement "RET504", @@ -174,7 +189,7 @@ lint.per-file-ignores."tests/*" = [ "S310", # False positive; it's a https url "S311", # Standard pseudo-random generators are not suitable for cryptographic purposes ] -lint.per-file-ignores."src/crate/client/{connection.py,http.py,_pool.py,_response.py}" = [ +lint.per-file-ignores."src/crate/client/{connection.py,http.py,_pool.py,_response.py,aio/__init__.py,aio/_http.py}" = [ "A004", # Import `ConnectionError` is shadowing a Python builtin "A005", # Import `ConnectionError` is shadowing a Python builtin ] diff --git a/src/crate/client/_response.py b/src/crate/client/_response.py index b6dc6e95..b01dc550 100644 --- a/src/crate/client/_response.py +++ b/src/crate/client/_response.py @@ -32,6 +32,7 @@ ) SRV_UNAVAILABLE_STATUSES = {502, 503, 504, 509} +MAX_REDIRECTS = 10 PRESERVE_ACTIVE_SERVER_EXCEPTIONS = {ConnectionResetError, BrokenPipeError} diff --git a/src/crate/client/aio/__init__.py b/src/crate/client/aio/__init__.py new file mode 100644 index 00000000..30da9ed9 --- /dev/null +++ b/src/crate/client/aio/__init__.py @@ -0,0 +1,35 @@ +# -*- coding: utf-8; -*- +# +# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor +# license agreements. See the NOTICE file distributed with this work for +# additional information regarding copyright ownership. Crate licenses +# this file to you under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. You may +# obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# However, if you have executed another commercial license agreement +# with Crate these terms will supersede the license and you may use the +# software solely pursuant to the terms of the relevant commercial agreement. + +""" +Experimental asyncio client for CrateDB, over CrateDB's HTTP endpoint. + +It is not public yet: the API may change without notice until it is +documented. +""" + +try: + import aiohttp # noqa: F401 +except ImportError as ex: # pragma: no cover + raise ImportError( + "crate.client.aio requires aiohttp. " + 'Install it with: pip install "aiohttp>=3.12,<4"' + ) from ex From 55f628c5f0a0937f2ec0fedc4b599b9e149ed12d Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Thu, 22 Oct 2026 17:11:31 +0200 Subject: [PATCH 02/10] Add workflow job to test CrateDB client without aiohttp dependency --- .github/workflows/tests.yml | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 3af3583c..1b5a29d3 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -82,3 +82,33 @@ jobs: CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }} with: fail_ci_if_error: true + + without-aiohttp: + name: "Without aiohttp" + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + + - name: Set up uv + uses: astral-sh/setup-uv@v7 + with: + python-version: "3.14" + activate-environment: true + version: "latest" + + - name: Setup env + run: uv pip install -e . + + - name: Check that the async client is optional + run: | + python -c "import crate.client" + python - <<'PYTHON' + try: + import crate.client.aio + except ImportError as ex: + assert "requires aiohttp" in str(ex), ex + else: + raise AssertionError("crate.client.aio imported without aiohttp") + PYTHON From 65768a00e7456ec4d489f5ba3e3154265ad70b67 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Thu, 22 Oct 2026 17:11:55 +0200 Subject: [PATCH 03/10] Add TLS utility functions for CA certificate resolution and integrate into HTTP client --- src/crate/client/_tls.py | 65 ++++++++++++++++++++++++++++++++++++++++ src/crate/client/http.py | 7 ++--- 2 files changed, 67 insertions(+), 5 deletions(-) create mode 100644 src/crate/client/_tls.py diff --git a/src/crate/client/_tls.py b/src/crate/client/_tls.py new file mode 100644 index 00000000..78b8fbfe --- /dev/null +++ b/src/crate/client/_tls.py @@ -0,0 +1,65 @@ +# -*- coding: utf-8; -*- +# +# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor +# license agreements. See the NOTICE file distributed with this work for +# additional information regarding copyright ownership. Crate licenses +# this file to you under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. You may +# obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# However, if you have executed another commercial license agreement +# with Crate these terms will supersede the license and you may use the +# software solely pursuant to the terms of the relevant commercial agreement. + +import os +import ssl +import typing as t + +from urllib3.util.ssl_ import create_urllib3_context + + +def resolve_ca_cert(ca_cert: t.Optional[str]) -> t.Optional[str]: + """ + Return the CA bundle to use, falling back to ``REQUESTS_CA_BUNDLE``. + """ + ca_cert = ca_cert or os.environ.get("REQUESTS_CA_BUNDLE", None) + if ca_cert and not os.path.exists(ca_cert): + # Sanity check + raise IOError('CA bundle file "{}" does not exist.'.format(ca_cert)) + return ca_cert + + +def make_ssl_context( + verify_ssl_cert: bool = True, + ca_cert: t.Optional[str] = None, + cert_file: t.Optional[str] = None, + key_file: t.Optional[str] = None, + ssl_relax_minimum_version: bool = False, +) -> ssl.SSLContext: + """ + Build the `SSLContext` that urllib3 builds for the same options. + """ + context = create_urllib3_context( + cert_reqs=ssl.CERT_REQUIRED if verify_ssl_cert else ssl.CERT_NONE, + ssl_minimum_version=( + ssl.TLSVersion.MINIMUM_SUPPORTED + if ssl_relax_minimum_version + else None + ), + ) + ca_cert = resolve_ca_cert(ca_cert) + if ca_cert: + context.load_verify_locations(cafile=ca_cert) + else: + context.load_default_certs() + if cert_file: + context.load_cert_chain(cert_file, key_file) + return context diff --git a/src/crate/client/http.py b/src/crate/client/http.py index 4259628f..74f3cfeb 100644 --- a/src/crate/client/http.py +++ b/src/crate/client/http.py @@ -21,7 +21,6 @@ import logging -import os import ssl import threading import typing as t @@ -74,6 +73,7 @@ redirect_target, should_preserve_server, ) +from crate.client._tls import resolve_ca_cert from crate.client.exceptions import ( ConnectionError, DigestNotFoundException, @@ -194,10 +194,7 @@ def _pool_kw_args( timeout=None, pool_size=None, ): - ca_cert = ca_cert or os.environ.get("REQUESTS_CA_BUNDLE", None) - if ca_cert and not os.path.exists(ca_cert): - # Sanity check - raise IOError('CA bundle file "{}" does not exist.'.format(ca_cert)) + ca_cert = resolve_ca_cert(ca_cert) kw = { "ca_certs": ca_cert, From af9a98bf0400f0442d6685a35167a66c505c0746 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 26 Oct 2026 12:12:38 +0100 Subject: [PATCH 04/10] Add asynchronous HTTP client implementation --- src/crate/client/aio/_http.py | 711 ++++++++++++++++++++++++++++++++++ 1 file changed, 711 insertions(+) create mode 100644 src/crate/client/aio/_http.py diff --git a/src/crate/client/aio/_http.py b/src/crate/client/aio/_http.py new file mode 100644 index 00000000..ee0539d7 --- /dev/null +++ b/src/crate/client/aio/_http.py @@ -0,0 +1,711 @@ +# -*- coding: utf-8; -*- +# +# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor +# license agreements. See the NOTICE file distributed with this work for +# additional information regarding copyright ownership. Crate licenses +# this file to you under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. You may +# obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# However, if you have executed another commercial license agreement +# with Crate these terms will supersede the license and you may use the +# software solely pursuant to the terms of the relevant commercial agreement. + +import asyncio +import errno +import logging +import socket +import ssl +import sys +import typing as t +import warnings +from dataclasses import dataclass +from urllib.parse import SplitResult, urlsplit + +import aiohttp +import urllib3 +from yarl import URL + +from crate.client._options import ( + SQL_PATH, + credentials_from_url, + keepalive_socket_options, + sql_path, + to_server_list, + validate_compress, +) +from crate.client._pool import _ServerPool +from crate.client._request import ( + build_headers, + compress_body, + create_sql_payload, +) +from crate.client._response import ( + MAX_REDIRECTS, + SRV_UNAVAILABLE_STATUSES, + parse_server_infos, + raise_for_status, + redirect_target, +) +from crate.client._tls import make_ssl_context +from crate.client.exceptions import ConnectionError, ProgrammingError +from crate.client.http import _ex_to_message, _json_from_response + +logger = logging.getLogger("crate.client.aio") + +DEFAULT_POOL_SIZE = 20 +CONNECT_RETRIES = 10 +BACKOFF_MAX = 120.0 + +_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308}) + +# Work around the CPython SSL transport leak, as aiohttp recommends. +_ENABLE_CLEANUP_CLOSED = sys.version_info < (3, 12, 7) or sys.version_info[ + :3 +] == (3, 13, 0) + +_CONNECT_ERRORS = (aiohttp.ClientConnectorError, aiohttp.ConnectionTimeoutError) +_TRANSPORT_ERRORS = ( + aiohttp.ClientError, + TimeoutError, + asyncio.TimeoutError, + ssl.SSLError, + ConnectionResetError, + BrokenPipeError, +) + + +def _is_stale_connection(ex: BaseException) -> bool: + """ + Whether the server closed a keep-alive connection while it was idle. + """ + if isinstance( + ex, + ( + aiohttp.ServerDisconnectedError, + ConnectionResetError, + BrokenPipeError, + ), + ): + return True + return isinstance(ex, aiohttp.ClientOSError) and ex.errno in ( + errno.ECONNRESET, + errno.EPIPE, + ) + + +def backoff_time(backoff_factor: float, errors: int) -> float: + """ + Seconds to wait before the next attempt after `errors` + """ + if errors <= 1: + return 0.0 + return min(BACKOFF_MAX, backoff_factor * (2 ** (errors - 1))) + + +def client_timeout(timeout: t.Any) -> aiohttp.ClientTimeout: + """ + Map the `timeout` option to aiohttp + + """ + if isinstance(timeout, urllib3.Timeout): + raise TypeError( + "The async client does not support urllib3.Timeout; " + "pass the timeout in seconds as a number" + ) + if timeout is None: + # aiohttp's default is `total=300, sock_connect=30`. + return aiohttp.ClientTimeout( + total=None, connect=None, sock_connect=None, sock_read=None + ) + if isinstance(timeout, str): + timeout = float(timeout) + if not isinstance(timeout, (int, float)) or isinstance(timeout, bool): + raise TypeError(f"timeout must be a number of seconds, got {timeout!r}") + return aiohttp.ClientTimeout( + total=None, connect=None, sock_connect=timeout, sock_read=timeout + ) + + +def _socket_factory( + options: t.List[t.Tuple[int, int, int]], +) -> t.Callable[[t.Any], socket.socket]: + """ + Create sockets with `TCP_NODELAY` and the keepalive options. + """ + + def factory(addr_info: t.Any) -> socket.socket: + family, type_, proto, _, _ = addr_info + sock = socket.socket(family=family, type=type_, proto=proto) + try: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + for level, option, value in options: + sock.setsockopt(level, option, value) + except OSError: + sock.close() + raise + return sock + + return factory + + +@dataclass(frozen=True) +class _Response: + """ + A fully read response. + """ + + status: int + reason: t.Optional[str] + headers: t.Mapping[str, str] + data: bytes + location: t.Optional[str] + + async def aclose(self) -> None: + pass + + +class _StreamedResponse: + """ + A response whose body has not been read yet. + """ + + def __init__( + self, + response: aiohttp.ClientResponse, + location: t.Optional[str], + release: t.Callable[[], None], + ) -> None: + self.status = response.status + self.reason = response.reason + self.headers = response.headers + self.location = location + self._response = response + self._release = release + self._closed = False + + async def read(self) -> bytes: + return await self._response.read() + + def iter_chunked(self, size: int) -> t.AsyncIterator[bytes]: + return self._response.content.iter_chunked(size) + + def close(self) -> None: + """ + Close the response without awaiting anything. + """ + if self._closed: + return + self._closed = True + try: + self._response.close() + finally: + self._release() + + async def aclose(self) -> None: + self.close() + + +class AsyncServer: + """ + One CrateDB server: an aiohttp session with its own connection pool. + """ + + def __init__( + self, + server: str, + *, + ssl_context: t.Optional[ssl.SSLContext], + timeout: aiohttp.ClientTimeout, + pool_size: int, + socket_options: t.List[t.Tuple[int, int, int]], + ) -> None: + self.server = server + self.path_prefix = "" + try: + parsed_url = urlsplit(server) + except Exception as e: + parsed_url = SplitResult("", "", "", "", "") + logger.warning( + "Unable to extract path prefix from server url: %s", e + ) + if parsed_url.path: + self.path_prefix = parsed_url.path.strip("/") + # Userinfo is sent as headers by `build_headers`. aiohttp would turn + # it into a second, conflicting `Authorization` header. + self._origin = str(URL(server).origin()) + # TLS settings are ignored for `http://` servers, as in sync. + self._ssl = ssl_context if server.lower().startswith("https") else None + self._timeout = timeout + self._pool_size = pool_size + self._socket_options = socket_options + self._session: t.Optional[aiohttp.ClientSession] = None + self._loop: t.Optional[asyncio.AbstractEventLoop] = None + self._slots: t.Optional[asyncio.Semaphore] = None + self._in_flight = 0 + self._idle: t.Optional[asyncio.Event] = None + self._closed = False + + def _get_session(self) -> aiohttp.ClientSession: + loop = asyncio.get_running_loop() + if self._closed: + raise ProgrammingError("Client closed") + if self._session is None: + connector = aiohttp.TCPConnector( + limit=self._pool_size, + limit_per_host=self._pool_size, + ssl=self._ssl if self._ssl is not None else True, + enable_cleanup_closed=_ENABLE_CLEANUP_CLOSED, + socket_factory=( + _socket_factory(self._socket_options) + if self._socket_options + else None + ), + ) + self._session = aiohttp.ClientSession( + connector=connector, + timeout=self._timeout, + auto_decompress=True, + trust_env=False, + cookie_jar=aiohttp.DummyCookieJar(), + ) + self._loop = loop + self._slots = asyncio.Semaphore(self._pool_size) + self._idle = asyncio.Event() + self._idle.set() + elif loop is not self._loop: + raise ProgrammingError( + "AsyncConnection used from a different event loop than it " + "was created in" + ) + return self._session + + async def _acquire(self) -> None: + """ + Wait for a free connection, in arrival order. + """ + if self._slots is None or self._idle is None: + raise RuntimeError("No session; call _get_session() first") + self._in_flight += 1 + self._idle.clear() + try: + await self._slots.acquire() + except BaseException: + self._done() + raise + + def _release(self) -> None: + if self._slots is None: + return + self._slots.release() + self._done() + + def _done(self) -> None: + self._in_flight -= 1 + if not self._in_flight and self._idle is not None: + self._idle.set() + + def _url(self, path: str) -> URL: + if self.path_prefix: + path = "/{path_prefix}/{path}".format( + path_prefix=self.path_prefix, path=path.strip("/") + ) + return URL(self._origin + path, encoded=True) + + async def open( + self, + method: str, + path: str, + data: t.Any = None, + headers: t.Optional[t.Dict[str, str]] = None, + username: t.Optional[str] = None, + password: t.Optional[str] = None, + schema: t.Optional[str] = None, + jwt_token: t.Optional[str] = None, + ) -> _StreamedResponse: + """ + Send a request and return the response before reading its body. + """ + headers = build_headers( + headers, + data, + username=username, + password=password, + schema=schema, + jwt_token=jwt_token, + ) + session = self._get_session() + await self._acquire() + try: + resp = await session.request( + method, + self._url(path), + data=data, + headers=headers, + allow_redirects=False, + ) + except BaseException: + self._release() + raise + location = ( + resp.headers.get("Location") + if resp.status in _REDIRECT_STATUSES + else None + ) + return _StreamedResponse(resp, location, self._release) + + async def request( + self, + method: str, + path: str, + data: t.Any = None, + headers: t.Optional[t.Dict[str, str]] = None, + username: t.Optional[str] = None, + password: t.Optional[str] = None, + schema: t.Optional[str] = None, + jwt_token: t.Optional[str] = None, + ) -> _Response: + """ + Send a request and read its body. + """ + response = await self.open( + method, + path, + data=data, + headers=headers, + username=username, + password=password, + schema=schema, + jwt_token=jwt_token, + ) + try: + body = await response.read() + finally: + response.close() + return _Response( + status=response.status, + reason=response.reason, + headers=response.headers, + data=body, + location=response.location, + ) + + async def close(self) -> None: + """ + Wait for in-flight requests, then close the connection pool. + """ + if self._closed: + return + self._closed = True + try: + if self._idle is not None: + await self._idle.wait() + if self._session is not None: + await self._session.close() + except BaseException: + self.terminate() + raise + + def terminate(self) -> None: + """ + Close all connections at once, without awaiting anything. + """ + self._closed = True + session = self._session + if session is None or session.closed: + return + connector = session.connector + if connector is not None: + # The public `close()` returns an awaitable that warns when it + # is not awaited. `_close()` closes the transports synchronously. + connector._close(abort_ssl=True) + + @property + def closed(self) -> bool: + return self._closed + + @property + def is_open(self) -> bool: + """ + Whether the server has a session that was not closed yet. + """ + return ( + self._session is not None + and not self._closed + and not self._session.closed + ) + + +class AsyncClient(_ServerPool): + """ + Asynchronous CrateDB HTTP client. Experimental. + + It takes the same arguments as `crate.client.http.Client`, except that + `timeout` must be a number of seconds. + """ + + SQL_PATH = SQL_PATH + """Crate URI path for issuing SQL statements.""" + + default_server = "http://127.0.0.1:4200" + """Default server to use if no servers are given on instantiation.""" + + connect_retries = CONNECT_RETRIES + """Connection attempts per server and request after the first one.""" + + _logger = logger + + def __init__( + self, + servers: t.Union[str, t.Iterable[str], None] = None, + timeout: t.Union[int, float, str, None] = None, + backoff_factor: float = 0, + verify_ssl_cert: bool = True, + ca_cert: t.Optional[str] = None, + error_trace: bool = False, + cert_file: t.Optional[str] = None, + key_file: t.Optional[str] = None, + ssl_relax_minimum_version: bool = False, + username: t.Optional[str] = None, + password: t.Optional[str] = None, + schema: t.Optional[str] = None, + pool_size: t.Optional[int] = None, + socket_keepalive: bool = True, + socket_tcp_keepidle: t.Optional[int] = None, + socket_tcp_keepintvl: t.Optional[int] = None, + socket_tcp_keepcnt: t.Optional[int] = None, + jwt_token: t.Optional[str] = None, + compress: t.Union[int, bool] = 8192, + ) -> None: + server_list = ( + to_server_list(servers) if servers else [self.default_server] + ) + + if server_list and not username: + try: + username, password = credentials_from_url( + server_list[0], username, password + ) + except Exception as ex: + logger.warning( + "Unable to decode credentials from database " + "URI, so connecting to CrateDB without " + "authentication: {ex}".format(ex=ex) + ) + + super().__init__(server_list) + self._timeout = client_timeout(timeout) + self._ssl_context = make_ssl_context( + verify_ssl_cert, + ca_cert, + cert_file, + key_file, + ssl_relax_minimum_version, + ) + self._pool_size = ( + int(pool_size) if pool_size is not None else DEFAULT_POOL_SIZE + ) + self._socket_options = keepalive_socket_options( + socket_keepalive, + socket_tcp_keepidle, + socket_tcp_keepintvl, + socket_tcp_keepcnt, + ) + self.backoff_factor = backoff_factor + self.server_pool: t.Dict[str, AsyncServer] = {} + for server in server_list: + self._create_server(server) + self.username = username + self.password = password + self.jwt_token = jwt_token + self.schema = schema + self.compress = validate_compress(compress) + self.path = sql_path(error_trace, self.SQL_PATH) + + def _create_server(self, server: str) -> None: + self.server_pool[server] = AsyncServer( + server, + ssl_context=self._ssl_context, + timeout=self._timeout, + pool_size=self._pool_size, + socket_options=self._socket_options, + ) + + def _add_server(self, server: str) -> None: + with self._lock: + if server not in self.server_pool: + self._create_server(server) + + async def close(self) -> None: + """ + Wait for in-flight requests, then close all connections. + """ + await asyncio.gather( + *(server.close() for server in self.server_pool.values()) + ) + + def terminate(self) -> None: + """ + Close all connections at once, without awaiting anything. Safe to + call from `__del__`. + """ + for server in list(self.server_pool.values()): + server.terminate() + + @property + def closed(self) -> bool: + return all(server.closed for server in self.server_pool.values()) + + async def sql( + self, + stmt: t.Optional[str], + parameters: t.Any = None, + bulk_parameters: t.Any = None, + ) -> t.Any: + """ + Execute SQL stmt against the crate server. + """ + if stmt is None: + return None + + data = create_sql_payload(stmt, parameters, bulk_parameters) + logger.debug("Sending request to %s with payload: %s", self.path, data) + content = await self._json_request("POST", self.path, data=data) + logger.debug("JSON response for stmt(%s): %s", stmt, content) + return content + + async def server_infos(self, server: str) -> t.Tuple[str, t.Any, str]: + response = await self._request("GET", "/", server=server) + raise_for_status( + response.status, response.reason, response.headers, response.data + ) + return parse_server_infos(server, _json_from_response(response)) + + async def _json_request(self, method: str, path: str, data: bytes) -> t.Any: + """ + Issue request against the crate HTTP API. + """ + headers = {"Accept-Encoding": "gzip, deflate"} + data, compressed = compress_body(data, self.compress) + if compressed: + headers["Content-Encoding"] = "gzip" + + response = await self._request(method, path, data=data, headers=headers) + raise_for_status( + response.status, response.reason, response.headers, response.data + ) + if len(response.data) > 0: + return _json_from_response(response) + return response.data + + async def _send( + self, + server: str, + method: str, + path: str, + stream: bool = False, + **kwargs: t.Any, + ) -> t.Union[_Response, _StreamedResponse]: + """ + Send one request to `server`. + """ + target = self.server_pool[server] + send = target.open if stream else target.request + errors = 0 + stale_retried = False + while True: + try: + return await send( + method, + path, + username=self.username, + password=self.password, + schema=self.schema, + jwt_token=self.jwt_token, + **kwargs, + ) + except _CONNECT_ERRORS: + errors += 1 + if errors > self.connect_retries: + raise + await asyncio.sleep(backoff_time(self.backoff_factor, errors)) + except _TRANSPORT_ERRORS as ex: + if stale_retried or not _is_stale_connection(ex): + raise + stale_retried = True + + async def _request( + self, + method: str, + path: str, + server: t.Optional[str] = None, + _redirects: int = 0, + **kwargs: t.Any, + ) -> t.Any: + """Execute a request to the cluster + + A server is selected from the server pool. + """ + while True: + next_server = server or self._get_server() + try: + response = await self._send(next_server, method, path, **kwargs) + redirect_server = redirect_target( + response.status, response.location + ) + if redirect_server is not None: + await response.aclose() + if _redirects >= MAX_REDIRECTS: + raise ProgrammingError( + f"Exceeded {MAX_REDIRECTS} redirects, last " + f"redirect to {redirect_server}" + ) + self._add_server(redirect_server) + return await self._request( + method, + path, + server=redirect_server, + _redirects=_redirects + 1, + **kwargs, + ) + if not server and response.status in SRV_UNAVAILABLE_STATUSES: + await response.aclose() + with self._lock: + # drop server from active ones + self._drop_server(next_server, response.reason or "") + else: + return response + except _TRANSPORT_ERRORS as ex: + ex_message = _ex_to_message(ex) + if server: + raise ConnectionError( + "Server not available, exception: %s" % ex_message + ) from ex + with self._lock: + # drop server from active ones + self._drop_server(next_server, ex_message) + except Exception as e: + raise ProgrammingError(_ex_to_message(e)) from e + + def __repr__(self) -> str: + return "".format(str(self._active_servers)) + + def __del__(self) -> None: + servers = getattr(self, "server_pool", None) + if not servers or not any(s.is_open for s in servers.values()): + return + warnings.warn( + f"unclosed {self!r}; close it with `await client.close()`", + ResourceWarning, + stacklevel=2, + source=self, + ) + self.terminate() From 874be65c1558cc0fd93fa8cb5824e57a16c90abb Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 26 Oct 2026 12:19:01 +0100 Subject: [PATCH 05/10] Add unit tests for CA certificate resolution and SSL context handling --- tests/client/test_tls.py | 156 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 tests/client/test_tls.py diff --git a/tests/client/test_tls.py b/tests/client/test_tls.py new file mode 100644 index 00000000..37aa31d3 --- /dev/null +++ b/tests/client/test_tls.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8; -*- +# +# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor +# license agreements. See the NOTICE file distributed with this work for +# additional information regarding copyright ownership. Crate licenses +# this file to you under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. You may +# obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# However, if you have executed another commercial license agreement +# with Crate these terms will supersede the license and you may use the +# software solely pursuant to the terms of the relevant commercial agreement. + +import os +import socket +import ssl +from unittest.mock import patch + +import pytest +from urllib3.util.ssl_ import create_urllib3_context + +from crate.client import exceptions +from crate.client._tls import make_ssl_context, resolve_ca_cert +from crate.client.http import Client + +from .settings import assets_path + +CA_VALID = assets_path("pki/cacert_valid.pem") +CA_INVALID = assets_path("pki/cacert_invalid.pem") +CLIENT_CERT = assets_path("pki/client_valid.pem") + + +def test_resolve_ca_cert(): + with patch.dict(os.environ, {}, clear=True): + assert resolve_ca_cert(None) is None + assert resolve_ca_cert(CA_VALID) == CA_VALID + + +def test_resolve_ca_cert_from_environment(): + with patch.dict(os.environ, {"REQUESTS_CA_BUNDLE": CA_VALID}, clear=True): + assert resolve_ca_cert(None) == CA_VALID + # An explicit bundle wins. + assert resolve_ca_cert(CA_INVALID) == CA_INVALID + + +def test_resolve_ca_cert_missing_file(): + with patch.dict(os.environ, {}, clear=True): + with pytest.raises(IOError, match="does not exist"): + resolve_ca_cert("/nonexistent") + with pytest.raises(IOError, match="does not exist"): + Client("https://localhost:4200", ca_cert="/nonexistent") + env = {"REQUESTS_CA_BUNDLE": "/nonexistent"} + with patch.dict(os.environ, env, clear=True): + with pytest.raises(IOError, match="does not exist"): + resolve_ca_cert(None) + + +def test_ssl_context_matches_urllib3(): + """ + The context has the settings that urllib3 uses for the sync client. + """ + expected = create_urllib3_context(cert_reqs=ssl.CERT_REQUIRED) + context = make_ssl_context(True, CA_VALID) + assert context.minimum_version == expected.minimum_version + assert context.minimum_version == ssl.TLSVersion.TLSv1_2 + assert context.maximum_version == expected.maximum_version + assert context.options == expected.options + assert context.verify_flags == expected.verify_flags + assert context.check_hostname is True + assert context.verify_mode == ssl.CERT_REQUIRED + + +def test_ssl_context_without_verification(): + expected = create_urllib3_context(cert_reqs=ssl.CERT_NONE) + context = make_ssl_context(False, CA_VALID) + assert context.options == expected.options + assert context.check_hostname is False + assert context.verify_mode == ssl.CERT_NONE + + +def test_ssl_context_relaxed_minimum_version(): + """ + The same as the sync client's `ssl_minimum_version` override. + """ + context = make_ssl_context(True, CA_VALID, ssl_relax_minimum_version=True) + assert context.minimum_version == ssl.TLSVersion.MINIMUM_SUPPORTED + assert context.check_hostname is True + + +def test_ssl_context_default_certs(): + with patch.dict(os.environ, {}, clear=True): + context = make_ssl_context(True) + assert context.verify_mode == ssl.CERT_REQUIRED + + +def test_ssl_context_client_certificate(): + """ + The test client certificate file holds the key too. + """ + make_ssl_context(True, CA_VALID, cert_file=CLIENT_CERT) + make_ssl_context( + True, CA_VALID, cert_file=CLIENT_CERT, key_file=CLIENT_CERT + ) + with pytest.raises(ssl.SSLError): + make_ssl_context( + True, CA_VALID, cert_file=assets_path("pki/readme.rst") + ) + + +def _handshake(context: ssl.SSLContext, host: str, port: int) -> None: + with socket.create_connection((host, port), timeout=5) as sock: + with context.wrap_socket(sock, server_hostname=host): + pass + + +def test_handshake_valid_ca(https_server): + """ + The sync client and the shared context both accept the test server's + certificate with the test CA. + """ + host, port = "localhost", https_server.server_address[1] + url = f"https://{host}:{port}" + client = Client(servers=url, ca_cert=CA_VALID) + try: + assert client.server_infos(url)[1] == "test" + finally: + client.close() + _handshake(make_ssl_context(True, CA_VALID), host, port) + + +def test_handshake_invalid_ca(https_server): + """ + The sync client and the shared context both reject the test server's + certificate with another CA, and both accept it without verification. + """ + host, port = "localhost", https_server.server_address[1] + url = f"https://{host}:{port}" + client = Client(servers=url, ca_cert=CA_INVALID) + try: + with pytest.raises( + exceptions.ConnectionError, match="certificate verify failed" + ): + client.server_infos(url) + finally: + client.close() + with pytest.raises(ssl.SSLCertVerificationError): + _handshake(make_ssl_context(True, CA_INVALID), host, port) + _handshake(make_ssl_context(False, CA_INVALID), host, port) From b2032df733e3cc356f2861152e755b6e30e2ba7f Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 26 Oct 2026 14:06:43 +0100 Subject: [PATCH 06/10] Add aiohttp support and fake CrateDB server for testing --- tests/client/aio_support.py | 184 ++++++++++++++++++++++++++++++++++++ tests/client/conftest.py | 25 +++++ 2 files changed, 209 insertions(+) create mode 100644 tests/client/aio_support.py create mode 100644 tests/client/conftest.py diff --git a/tests/client/aio_support.py b/tests/client/aio_support.py new file mode 100644 index 00000000..f8a50856 --- /dev/null +++ b/tests/client/aio_support.py @@ -0,0 +1,184 @@ +# -*- coding: utf-8; -*- +# +# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor +# license agreements. See the NOTICE file distributed with this work for +# additional information regarding copyright ownership. Crate licenses +# this file to you under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. You may +# obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# However, if you have executed another commercial license agreement +# with Crate these terms will supersede the license and you may use the +# software solely pursuant to the terms of the relevant commercial agreement. + + +import asyncio +import socket +import ssl +import typing as t + +import pytest + +pytest.importorskip("aiohttp") + +import pytest_asyncio # noqa: E402 +from aiohttp import web # noqa: E402 + +from .settings import assets_path # noqa: E402 + +SQL_RESULT = {"cols": ["name"], "rows": [["crate"]], "rowcount": 1} + +Handler = t.Callable[[web.Request], t.Awaitable[web.StreamResponse]] + + +def server_ssl_context() -> ssl.SSLContext: + """ + The TLS context of the test server certificate. + """ + context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + context.load_cert_chain(assets_path("pki/server_valid.pem")) + return context + + +class FakeCrate: + """ + An in-process HTTP server. + """ + + def __init__( + self, + handler: t.Optional[Handler] = None, + ssl_context: t.Optional[ssl.SSLContext] = None, + ): + self.handler = handler or self.ok + self.ssl_context = ssl_context + self.requests: t.List[t.Tuple[web.Request, bytes]] = [] + self.runner: t.Optional[web.AppRunner] = None + self.port = 0 + self.url = "" + + @staticmethod + async def ok(request: web.Request) -> web.StreamResponse: + if request.path == "/": + return web.json_response( + {"name": "crate-1", "version": {"number": "6.1.2"}} + ) + return web.json_response(SQL_RESULT) + + async def _dispatch(self, request: web.Request) -> web.StreamResponse: + body = await request.read() + self.requests.append((request, body)) + return await self.handler(request) + + async def start(self) -> "FakeCrate": + app = web.Application() + app.router.add_route("*", "/{tail:.*}", self._dispatch) + self.runner = web.AppRunner( + app, handle_signals=False, shutdown_timeout=0.1 + ) + await self.runner.setup() + site = web.TCPSite( + self.runner, "127.0.0.1", 0, ssl_context=self.ssl_context + ) + await site.start() + self.port = site._server.sockets[0].getsockname()[1] # type: ignore[union-attr] + if self.ssl_context: + self.url = f"https://localhost:{self.port}" + else: + self.url = f"http://127.0.0.1:{self.port}" + return self + + async def stop(self) -> None: + if self.runner is not None: + await self.runner.cleanup() + + +class SilentFirstProxy: + """ + A TCP proxy that accepts the first connection and never answers it. + """ + + def __init__(self, target_port: int): + self.target_port = target_port + self.connections = 0 + self.port = 0 + self._server: t.Optional[asyncio.Server] = None + self._writers: t.List[asyncio.StreamWriter] = [] + self._tasks: t.Set[asyncio.Task] = set() + + async def start(self) -> "SilentFirstProxy": + self._server = await asyncio.start_server(self._handle, "127.0.0.1", 0) + self.port = self._server.sockets[0].getsockname()[1] + return self + + async def _handle( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + self.connections += 1 + self._writers.append(writer) + if self.connections == 1: + return + upstream_reader, upstream_writer = await asyncio.open_connection( + "127.0.0.1", self.target_port + ) + self._writers.append(upstream_writer) + for src, dst in ((reader, upstream_writer), (upstream_reader, writer)): + task = asyncio.ensure_future(self._pipe(src, dst)) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + @staticmethod + async def _pipe( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + try: + while data := await reader.read(65536): + writer.write(data) + await writer.drain() + except (ConnectionError, OSError): + pass + finally: + writer.close() + + async def stop(self) -> None: + for task in list(self._tasks): + task.cancel() + for writer in self._writers: + writer.close() + if self._server is not None: + self._server.close() + await self._server.wait_closed() + + +@pytest_asyncio.fixture +async def fake_crate(): + """ + Start in-process CrateDB stand-ins. + """ + servers: t.List[FakeCrate] = [] + + async def start( + handler: t.Optional[Handler] = None, + ssl_context: t.Optional[ssl.SSLContext] = None, + ) -> FakeCrate: + server = await FakeCrate(handler, ssl_context).start() + servers.append(server) + return server + + yield start + for server in servers: + await server.stop() + + +def unused_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] diff --git a/tests/client/conftest.py b/tests/client/conftest.py new file mode 100644 index 00000000..89e9b779 --- /dev/null +++ b/tests/client/conftest.py @@ -0,0 +1,25 @@ +# -*- coding: utf-8; -*- +# +# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor +# license agreements. See the NOTICE file distributed with this work for +# additional information regarding copyright ownership. Crate licenses +# this file to you under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. You may +# obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# However, if you have executed another commercial license agreement +# with Crate these terms will supersede the license and you may use the +# software solely pursuant to the terms of the relevant commercial agreement. + +import importlib.util + +if importlib.util.find_spec("aiohttp") is not None: + from .aio_support import fake_crate # noqa: F401 From 709b8020e9ccb82f53f5e6c0fd8544a48fe8ff96 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 26 Oct 2026 14:15:13 +0100 Subject: [PATCH 07/10] Add comprehensive unit tests for asynchronous HTTP client functionality --- tests/client/test_aio_http.py | 1130 +++++++++++++++++++++++++++++++++ 1 file changed, 1130 insertions(+) create mode 100644 tests/client/test_aio_http.py diff --git a/tests/client/test_aio_http.py b/tests/client/test_aio_http.py new file mode 100644 index 00000000..46a438f6 --- /dev/null +++ b/tests/client/test_aio_http.py @@ -0,0 +1,1130 @@ +# -*- coding: utf-8; -*- +# +# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor +# license agreements. See the NOTICE file distributed with this work for +# additional information regarding copyright ownership. Crate licenses +# this file to you under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. You may +# obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# However, if you have executed another commercial license agreement +# with Crate these terms will supersede the license and you may use the +# software solely pursuant to the terms of the relevant commercial agreement. + +import asyncio +import errno +import gc +import json +import logging +import socket +import ssl +import time +import warnings + +import pytest + +pytest.importorskip("aiohttp") + +import aiohttp # noqa: E402 +import urllib3 # noqa: E402 +from aiohttp import web # noqa: E402 + +from crate.client import exceptions # noqa: E402 +from crate.client._options import keepalive_socket_options # noqa: E402 +from crate.client.aio import _http # noqa: E402 +from crate.client.aio._http import ( # noqa: E402 + AsyncClient, + AsyncServer, + backoff_time, + client_timeout, +) +from crate.client.http import Client # noqa: E402 + +from .aio_support import ( # noqa: E402 + SQL_RESULT, + FakeCrate, + SilentFirstProxy, + server_ssl_context, + unused_port, +) +from .settings import assets_path # noqa: E402 + +pytestmark = [ + pytest.mark.timeout(30, method="thread"), + pytest.mark.filterwarnings("error::ResourceWarning"), + pytest.mark.filterwarnings( + "error::pytest.PytestUnraisableExceptionWarning" + ), +] + +CA_VALID = assets_path("pki/cacert_valid.pem") +CA_INVALID = assets_path("pki/cacert_invalid.pem") + + +async def unavailable(request): + return web.Response(status=503, reason="Service Unavailable") + + +@pytest.mark.parametrize( + "errors, expected", [(0, 0), (1, 0), (2, 1.0), (3, 2.0), (12, 120.0)] +) +def test_backoff_time_matches_urllib3(errors, expected): + assert backoff_time(0.5, errors) == expected + retry = urllib3.Retry(total=20, backoff_factor=0.5) + for _ in range(errors): + retry = retry.increment( + "POST", "/", error=urllib3.exceptions.ConnectTimeoutError() + ) + assert retry.get_backoff_time() == expected + + +@pytest.mark.parametrize( + "timeout, sock_timeout", [(None, None), (2, 2), (1.5, 1.5), ("7", 7.0)] +) +def test_client_timeout(timeout, sock_timeout): + result = client_timeout(timeout) + # aiohttp's defaults would be `total=300, sock_connect=30`. + assert result.total is None + assert result.connect is None + assert result.sock_connect == sock_timeout + assert result.sock_read == sock_timeout + + +@pytest.mark.parametrize("timeout", [urllib3.Timeout(connect=1), [1], True]) +def test_client_timeout_invalid(timeout): + with pytest.raises(TypeError): + client_timeout(timeout) + with pytest.raises(TypeError): + AsyncClient(timeout=timeout) + + +def test_missing_ca_bundle(): + with pytest.raises(IOError, match="does not exist"): + AsyncClient(ca_cert="/nonexistent/ca.pem") + + +def test_creates_no_aiohttp_objects(): + client = AsyncClient(servers=["localhost:4200", "localhost:4201"]) + assert all(s._session is None for s in client.server_pool.values()) + assert client.active_servers == [ + "http://localhost:4200", + "http://localhost:4201", + ] + + +def test_url_strips_userinfo_and_keeps_path_prefix(): + server = AsyncServer( + "https://crate:s3cr3t@localhost:4200/prefix/", + ssl_context=None, + timeout=client_timeout(None), + pool_size=1, + socket_options=[], + ) + assert str(server._url("/_sql")) == "https://localhost:4200/prefix/_sql" + + +def test_tls_context_only_for_https(): + client = AsyncClient(servers=["http://localhost:4200", "https://h:4200"]) + assert client.server_pool["http://localhost:4200"]._ssl is None + assert client.server_pool["https://h:4200"]._ssl is client._ssl_context + + +@pytest.mark.asyncio +async def test_sql(fake_crate): + crate = await fake_crate() + client = AsyncClient( + servers=crate.url, username="crate", password="s3cr3t", schema="doc" + ) + try: + assert await client.sql("SELECT name FROM sys.cluster") == SQL_RESULT + assert await client.sql(None) is None + finally: + await client.close() + + assert len(crate.requests) == 1 + request, body = crate.requests[0] + assert request.method == "POST" + assert request.path_qs == "/_sql?types=true" + assert json.loads(body) == {"stmt": "SELECT name FROM sys.cluster"} + assert request.headers["Authorization"] == "Basic Y3JhdGU6czNjcjN0" + assert request.headers["X-User"] == "crate" + assert request.headers["Default-Schema"] == "doc" + assert request.headers["Content-Type"] == "application/json" + assert request.headers["Accept"] == "application/json" + assert request.headers["Accept-Encoding"] == "gzip, deflate" + assert "Content-Encoding" not in request.headers + assert "Cookie" not in request.headers + + +@pytest.mark.asyncio +async def test_sql_parameters(fake_crate): + crate = await fake_crate() + client = AsyncClient(servers=crate.url) + try: + await client.sql("SELECT ?", parameters=[1]) + await client.sql("INSERT ?", bulk_parameters=[[1], [2]]) + finally: + await client.close() + assert json.loads(crate.requests[0][1]) == {"stmt": "SELECT ?", "args": [1]} + assert json.loads(crate.requests[1][1]) == { + "stmt": "INSERT ?", + "bulk_args": [[1], [2]], + } + + +@pytest.mark.asyncio +async def test_jwt_token(fake_crate): + crate = await fake_crate() + client = AsyncClient(servers=crate.url, jwt_token="t0k3n") + try: + await client.sql("SELECT 1") + finally: + await client.close() + request, _ = crate.requests[0] + assert request.headers["Authorization"] == "Bearer t0k3n" + + +@pytest.mark.asyncio +async def test_jwt_and_username_rejected(fake_crate): + crate = await fake_crate() + client = AsyncClient(servers=crate.url, username="crate", jwt_token="t") + try: + with pytest.raises(exceptions.ProgrammingError, match="JWT"): + await client.sql("SELECT 1") + finally: + await client.close() + assert crate.requests == [] + + +@pytest.mark.asyncio +async def test_sql_compressed_and_error_trace(fake_crate): + crate = await fake_crate() + client = AsyncClient(servers=crate.url, compress=True, error_trace=True) + try: + await client.sql("SELECT 1") + finally: + await client.close() + request, body = crate.requests[0] + assert request.path_qs == "/_sql?types=true&error_trace=true" + assert request.headers["Content-Encoding"] == "gzip" + assert json.loads(body) == {"stmt": "SELECT 1"} + + +@pytest.mark.asyncio +async def test_gzip_response(fake_crate): + async def compressed(request): + response = web.json_response(SQL_RESULT) + response.enable_compression(web.ContentCoding.gzip) + return response + + crate = await fake_crate(compressed) + client = AsyncClient(servers=crate.url) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_userinfo_and_path_prefix(fake_crate): + """ + Credentials in the server URL become headers, and a path in the URL + prefixes every request path. + """ + crate = await fake_crate() + url = crate.url.replace("http://", "http://crate:s3cr3t@") + "/crate/" + client = AsyncClient(servers=url) + try: + await client.sql("SELECT 1") + finally: + await client.close() + request, _ = crate.requests[0] + assert request.path_qs == "/crate/_sql?types=true" + assert request.headers.getall("Authorization") == ["Basic Y3JhdGU6czNjcjN0"] + + +@pytest.mark.asyncio +async def test_proxy_environment_ignored(fake_crate, monkeypatch): + """ + As in sync, the proxy environment variables are ignored. + """ + for name in ("NO_PROXY", "no_proxy"): + monkeypatch.delenv(name, raising=False) + dead_proxy = f"http://127.0.0.1:{unused_port()}" + monkeypatch.setenv("HTTP_PROXY", dead_proxy) + monkeypatch.setenv("http_proxy", dead_proxy) + crate = await fake_crate() + client = AsyncClient(servers=crate.url) + client.connect_retries = 0 + try: + assert await client.sql("SELECT 1") == SQL_RESULT + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_server_infos(fake_crate): + crate = await fake_crate() + client = AsyncClient(servers=crate.url) + try: + assert await client.server_infos(crate.url) == ( + crate.url, + "crate-1", + "6.1.2", + ) + finally: + await client.close() + request, _ = crate.requests[0] + assert request.method == "GET" + assert request.path == "/" + + +@pytest.mark.asyncio +async def test_invalid_json_response(fake_crate): + async def html(request): + return web.Response(text="", content_type="text/html") + + crate = await fake_crate(html) + client = AsyncClient(servers=crate.url) + try: + with pytest.raises( + exceptions.ProgrammingError, match="Invalid server response" + ): + await client.sql("SELECT 1") + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_duplicate_key_maps_to_integrity_error(fake_crate): + async def duplicate(request): + return web.json_response( + { + "error": { + "message": "DuplicateKeyException[A document with the " + "same primary key exists already]", + "code": 4091, + } + }, + status=409, + ) + + crate = await fake_crate(duplicate) + client = AsyncClient(servers=crate.url) + try: + with pytest.raises(exceptions.IntegrityError): + await client.sql("INSERT INTO t (id) VALUES (1)") + finally: + await client.close() + + +async def relation_unknown(request): + return web.json_response( + { + "error": { + "message": "RelationUnknown[Relation 'foo' unknown]", + "code": 4041, + } + }, + status=404, + ) + + +async def slow(request): + await asyncio.sleep(2) + return web.json_response(SQL_RESULT) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "handler, options, exception, match", + [ + (relation_unknown, {}, exceptions.ProgrammingError, "RelationUnknown"), + (unavailable, {}, exceptions.ProgrammingError, "No more Servers"), + (slow, {"timeout": 0.2}, exceptions.ConnectionError, "No more Servers"), + (None, {}, exceptions.ConnectionError, "No more Servers"), + ], + ids=["error-response", "503-on-all-servers", "read-timeout", "refused"], +) +async def test_same_exception_as_sync( + fake_crate, handler, options, exception, match +): + if handler is None: + url = f"http://127.0.0.1:{unused_port()}" + else: + url = (await fake_crate(handler)).url + + def sync_sql(): + client = Client(servers=url, **options) + try: + client.sql("SELECT 1") + finally: + client.close() + + with pytest.raises(exception, match=match): + await asyncio.to_thread(sync_sql) + + client = AsyncClient(servers=url, **options) + try: + with pytest.raises(exception, match=match): + await client.sql("SELECT 1") + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_unavailable_fails_over(fake_crate, caplog): + down = await fake_crate(unavailable) + up = await fake_crate() + client = AsyncClient(servers=[down.url, up.url]) + try: + with caplog.at_level(logging.WARNING, logger="crate.client.aio"): + assert await client.sql("SELECT 1") == SQL_RESULT + assert client.active_servers == [up.url] + finally: + await client.close() + assert len(down.requests) == 1 + assert [r.name for r in caplog.records] == ["crate.client.aio"] + assert f"Removed server {down.url}" in caplog.text + + +@pytest.mark.asyncio +async def test_unavailable_on_given_server(fake_crate): + crate = await fake_crate(unavailable) + client = AsyncClient(servers=crate.url) + try: + with pytest.raises(exceptions.ConnectionError, match="503 Server"): + await client.server_infos(crate.url) + assert client.active_servers == [crate.url] + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_read_timeout_is_not_retried(fake_crate): + crate = await fake_crate(slow) + client = AsyncClient(servers=crate.url, timeout=0.2) + try: + with pytest.raises(exceptions.ConnectionError): + await client.sql("SELECT 1") + finally: + await client.close() + assert len(crate.requests) == 1 + + +@pytest.mark.asyncio +async def test_read_timeout_fails_over(fake_crate): + slow_server = await fake_crate(slow) + fast_server = await fake_crate() + client = AsyncClient( + servers=[slow_server.url, fast_server.url], timeout=0.2 + ) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + assert client.active_servers == [fast_server.url] + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_no_timeout_waits(fake_crate): + async def handler(request): + await asyncio.sleep(0.3) + return web.json_response(SQL_RESULT) + + crate = await fake_crate(handler) + client = AsyncClient(servers=crate.url, timeout=None) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + session = client.server_pool[crate.url]._session + # Not aiohttp's default of `total=300, sock_connect=30`. + assert session.timeout == aiohttp.ClientTimeout( + total=None, connect=None, sock_connect=None, sock_read=None + ) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_connection_refused_fails_over(fake_crate): + up = await fake_crate() + down = f"http://127.0.0.1:{unused_port()}" + client = AsyncClient(servers=[down, up.url]) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + assert client.active_servers == [up.url] + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_server_infos_unreachable(): + url = f"http://127.0.0.1:{unused_port()}" + client = AsyncClient(servers=url) + client.connect_retries = 0 + try: + with pytest.raises( + exceptions.ConnectionError, match="Server not available" + ): + await client.server_infos(url) + assert client.active_servers == [url] + finally: + await client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + asyncio.TimeoutError(), + ssl.SSLError(1, "[SSL: DECRYPTION_FAILED_OR_BAD_RECORD_MAC]"), + ], + ids=["asyncio-timeout", "ssl-error"], +) +async def test_unwrapped_transport_error_fails_over( + fake_crate, monkeypatch, error +): + """ + Errors that aiohttp does not wrap still fail over with + `ConnectionError`, instead of surfacing as `ProgrammingError`. + """ + first = await fake_crate() + second = await fake_crate() + original = AsyncServer.request + + async def fail_first(self, *args, **kwargs): + if self.server == first.url: + raise error + return await original(self, *args, **kwargs) + + monkeypatch.setattr(AsyncServer, "request", fail_first) + client = AsyncClient(servers=[first.url, second.url]) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + assert client.active_servers == [second.url] + with pytest.raises(exceptions.ConnectionError): + await client.server_infos(first.url) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_connect_retries_with_backoff(fake_crate, monkeypatch): + """ + A connection failure is retried `connect_retries` times on the same + server, with urllib3's backoff, before failing over. + """ + crate = await fake_crate() + attempts = [] + sleeps = [] + + async def refuse(self, *args, **kwargs): + attempts.append(self.server) + raise aiohttp.ConnectionTimeoutError("connect timed out") + + async def record_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(AsyncServer, "request", refuse) + monkeypatch.setattr(_http.asyncio, "sleep", record_sleep) + client = AsyncClient(servers=crate.url, backoff_factor=0.1) + try: + with pytest.raises(exceptions.ConnectionError): + await client.sql("SELECT 1") + finally: + await client.close() + # urllib3's default is `Retry(total=10)`: 11 attempts in all. + assert attempts == [crate.url] * 11 + assert sleeps == [backoff_time(0.1, n) for n in range(1, 11)] + assert sleeps[:4] == [0.0, 0.2, 0.4, 0.8] + + +@pytest.mark.asyncio +async def test_connect_retry_succeeds(fake_crate, monkeypatch): + crate = await fake_crate() + original = AsyncServer.request + calls = [] + + async def refuse_twice(self, *args, **kwargs): + calls.append(1) + if len(calls) <= 2: + raise aiohttp.ConnectionTimeoutError("connect timed out") + return await original(self, *args, **kwargs) + + monkeypatch.setattr(AsyncServer, "request", refuse_twice) + client = AsyncClient(servers=crate.url) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + assert client.active_servers == [crate.url] + finally: + await client.close() + assert len(calls) == 3 + + +@pytest.mark.asyncio +async def test_tls_failure_is_a_connect_error(): + """ + TLS handshake and certificate errors count as connect errors and are + retried, as urllib3 does. + """ + assert issubclass( + aiohttp.ClientConnectorCertificateError, _http._CONNECT_ERRORS + ) + assert issubclass(aiohttp.ClientConnectorSSLError, _http._CONNECT_ERRORS) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + aiohttp.ServerDisconnectedError(), + aiohttp.ClientOSError(errno.ECONNRESET, "Connection reset by peer"), + aiohttp.ClientOSError(errno.EPIPE, "Broken pipe"), + ConnectionResetError(errno.ECONNRESET, "Connection reset by peer"), + BrokenPipeError(errno.EPIPE, "Broken pipe"), + ], + ids=["disconnected", "oserror-reset", "oserror-pipe", "reset", "pipe"], +) +async def test_stale_connection_retried_once(fake_crate, monkeypatch, error): + crate = await fake_crate() + original = AsyncServer.request + calls = [] + + async def flaky(self, *args, **kwargs): + calls.append(1) + if len(calls) == 1: + raise error + return await original(self, *args, **kwargs) + + monkeypatch.setattr(AsyncServer, "request", flaky) + client = AsyncClient(servers=crate.url) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + assert client.active_servers == [crate.url] + finally: + await client.close() + assert len(calls) == 2 + + +@pytest.mark.asyncio +async def test_stale_connection_twice_fails_over(fake_crate, monkeypatch): + """ + A second reset in the same request counts as a read failure. (The sync + client keeps retrying a server that resets every connection.) + """ + first = await fake_crate() + second = await fake_crate() + original = AsyncServer.request + calls = [] + + async def reset_first(self, *args, **kwargs): + if self.server == first.url: + calls.append(1) + raise aiohttp.ServerDisconnectedError() + return await original(self, *args, **kwargs) + + monkeypatch.setattr(AsyncServer, "request", reset_first) + client = AsyncClient(servers=[first.url, second.url]) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + assert client.active_servers == [second.url] + finally: + await client.close() + assert len(calls) == 2 + + +@pytest.mark.asyncio +async def test_reset_by_server(fake_crate): + """ + A server that closes every connection without answering ends with + `ConnectionError` instead of a retry loop. + """ + + async def reset(request): + request.transport.close() + await asyncio.sleep(1) + return web.Response() + + crate = await fake_crate(reset) + client = AsyncClient(servers=crate.url) + try: + with pytest.raises(exceptions.ConnectionError): + await client.sql("SELECT 1") + finally: + await client.close() + assert len(crate.requests) == 2 + + +@pytest.mark.asyncio +async def test_redirect(fake_crate): + target = await fake_crate() + + async def redirect(request): + raise web.HTTPTemporaryRedirect(target.url + request.path_qs) + + origin = await fake_crate(redirect) + client = AsyncClient(servers=origin.url) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + assert target.url in client.server_pool + assert client.active_servers == [origin.url] + finally: + await client.close() + assert json.loads(target.requests[0][1]) == {"stmt": "SELECT 1"} + + +@pytest.mark.asyncio +async def test_location_ignored_without_redirect_status(fake_crate): + + target = await fake_crate() + + async def multiple_choices(request): + return web.json_response( + SQL_RESULT, status=300, headers={"Location": target.url + "/x"} + ) + + origin = await fake_crate(multiple_choices) + + def sync_sql(): + client = Client(servers=origin.url) + try: + return client.sql("SELECT 1") + finally: + client.close() + + assert await asyncio.to_thread(sync_sql) == SQL_RESULT + client = AsyncClient(servers=origin.url) + try: + assert await client.sql("SELECT 1") == SQL_RESULT + finally: + await client.close() + assert target.requests == [] + + +@pytest.mark.asyncio +async def test_redirect_hop_limit(fake_crate): + loop_server = None + + async def redirect_to_self(request): + raise web.HTTPTemporaryRedirect(loop_server.url + request.path_qs) + + loop_server = await fake_crate(redirect_to_self) + client = AsyncClient(servers=loop_server.url) + try: + with pytest.raises(exceptions.ProgrammingError, match="10 redirects"): + await client.sql("SELECT 1") + finally: + await client.close() + assert len(loop_server.requests) == 11 + + +@pytest.mark.asyncio +async def test_redirect_target_unavailable(fake_crate): + """ + As in sync, an error on the redirect target is a `ProgrammingError`. + """ + down = f"http://127.0.0.1:{unused_port()}" + + async def redirect(request): + raise web.HTTPTemporaryRedirect(down + request.path_qs) + + origin = await fake_crate(redirect) + client = AsyncClient(servers=origin.url) + client.connect_retries = 0 + try: + with pytest.raises( + exceptions.ProgrammingError, match="Server not available" + ): + await client.sql("SELECT 1") + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_tls_handshake(https_server): + + url = "https://localhost:{}".format(https_server.server_address[1]) + + client = AsyncClient(servers=url, ca_cert=CA_VALID) + try: + assert (await client.server_infos(url))[1] == "test" + finally: + await client.close() + + client = AsyncClient(servers=url, ca_cert=CA_INVALID) + client.connect_retries = 0 + try: + with pytest.raises( + exceptions.ConnectionError, match="certificate verify failed" + ): + await client.server_infos(url) + finally: + await client.close() + + client = AsyncClient(servers=url, verify_ssl_cert=False) + try: + assert (await client.server_infos(url))[1] == "test" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_pool_size(fake_crate): + running = 0 + most = 0 + + async def handler(request): + nonlocal running, most + running += 1 + most = max(most, running) + await asyncio.sleep(0.05) + running -= 1 + return web.json_response(SQL_RESULT) + + crate = await fake_crate(handler) + client = AsyncClient(servers=crate.url, pool_size=3) + try: + await asyncio.gather(*(client.sql("SELECT 1") for _ in range(12))) + finally: + await client.close() + assert most == 3 + assert client.server_pool[crate.url]._pool_size == 3 + assert AsyncClient()._pool_size == 20 + + +@pytest.mark.asyncio +async def test_fair_queuing(fake_crate): + + async def handler(request): + await asyncio.sleep(0.001) + return web.json_response(SQL_RESULT) + + crate = await fake_crate(handler) + client = AsyncClient(servers=crate.url, pool_size=5) + run_for = 1.0 + stop = time.monotonic() + run_for + waits = [] + + async def worker(): + while time.monotonic() < stop: + start = time.monotonic() + await client.sql("SELECT 1") + waits.append(time.monotonic() - start) + + try: + await asyncio.gather(*(worker() for _ in range(50))) + finally: + await client.close() + # Fair queuing keeps the longest wait at a few tens of milliseconds + assert max(waits) < run_for / 2 + + +@pytest.mark.asyncio +async def test_cancelled_waiter_frees_its_place(fake_crate): + release = asyncio.Event() + + async def blocked(request): + await release.wait() + return web.json_response(SQL_RESULT) + + crate = await fake_crate(blocked) + client = AsyncClient(servers=crate.url, pool_size=1) + try: + busy = asyncio.ensure_future(client.sql("SELECT 1")) + await asyncio.sleep(0.05) + waiter = asyncio.ensure_future(client.sql("SELECT 2")) + await asyncio.sleep(0.05) + waiter.cancel() + with pytest.raises(asyncio.CancelledError): + await waiter + release.set() + assert await busy == SQL_RESULT + assert await asyncio.wait_for(client.sql("SELECT 3"), 5) == SQL_RESULT + server = client.server_pool[crate.url] + assert server._in_flight == 0 + finally: + await client.close() + assert len(crate.requests) == 2 + + +async def _cancel_after(coro, delay): + task = asyncio.ensure_future(coro) + await asyncio.sleep(delay) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.asyncio +async def test_cancel_during_tls_handshake(fake_crate): + """ + The first connection hangs in the TLS handshake, which is part of + connecting. The next request opens a new connection. + """ + crate = await fake_crate(ssl_context=server_ssl_context()) + proxy = await SilentFirstProxy(crate.port).start() + url = f"https://localhost:{proxy.port}" + client = AsyncClient(servers=url, ca_cert=CA_VALID) + try: + await _cancel_after(client.sql("SELECT 1"), 0.2) + assert crate.requests == [] + assert client.active_servers == [url] + assert await client.sql("SELECT 2") == SQL_RESULT + assert client.active_servers == [url] + finally: + await client.close() + await proxy.stop() + assert proxy.connections >= 2 + + +@pytest.mark.asyncio +async def test_cancel_during_send(fake_crate): + calls = [] + + async def slow_first(request): + calls.append(1) + if len(calls) == 1: + await asyncio.sleep(5) + return web.json_response(SQL_RESULT) + + crate = await fake_crate(slow_first) + client = AsyncClient(servers=crate.url) + try: + await _cancel_after(client.sql("SELECT 1"), 0.1) + assert client.active_servers == [crate.url] + assert await client.sql("SELECT 2") == SQL_RESULT + assert client.server_pool[crate.url]._in_flight == 0 + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_cancel_mid_body(fake_crate): + calls = [] + + async def stream_slowly(request): + calls.append(1) + if len(calls) > 1: + return web.json_response(SQL_RESULT) + response = web.StreamResponse() + response.content_type = "application/json" + await response.prepare(request) + await response.write(b'{"cols": ["name"], "rows": [') + await asyncio.sleep(5) + return response + + crate = await fake_crate(stream_slowly) + client = AsyncClient(servers=crate.url, pool_size=1) + try: + await _cancel_after(client.sql("SELECT 1"), 0.1) + assert client.active_servers == [crate.url] + # With a pool of one, this only works if the slot was released. + assert await client.sql("SELECT 2") == SQL_RESULT + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_cancel_during_backoff(fake_crate, monkeypatch): + crate = await fake_crate() + + async def refuse(self, *args, **kwargs): + raise aiohttp.ConnectionTimeoutError("connect timed out") + + monkeypatch.setattr(AsyncServer, "request", refuse) + client = AsyncClient(servers=crate.url, backoff_factor=10) + try: + await _cancel_after(client.sql("SELECT 1"), 0.1) + assert client.active_servers == [crate.url] + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_close_waits_for_in_flight_requests(fake_crate): + async def handler(request): + await asyncio.sleep(0.2) + return web.json_response(SQL_RESULT) + + crate = await fake_crate(handler) + client = AsyncClient(servers=crate.url, pool_size=1) + running = asyncio.ensure_future(client.sql("SELECT 1")) + queued = asyncio.ensure_future(client.sql("SELECT 2")) + await asyncio.sleep(0.05) + await client.close() + assert running.done() and running.result() == SQL_RESULT + assert queued.done() and queued.result() == SQL_RESULT + assert client.closed + with pytest.raises(exceptions.ProgrammingError, match="Client closed"): + await client.sql("SELECT 3") + # Idempotent. + await client.close() + + +@pytest.mark.asyncio +async def test_close_unused_client(): + client = AsyncClient(servers=["localhost:4200", "localhost:4201"]) + await client.close() + assert client.closed + with pytest.raises(exceptions.ProgrammingError, match="Client closed"): + await client.sql("SELECT 1") + + +@pytest.mark.asyncio +async def test_cancelled_close_terminates(fake_crate): + async def blocked(request): + await asyncio.sleep(5) + return web.json_response(SQL_RESULT) + + crate = await fake_crate(blocked) + client = AsyncClient(servers=crate.url) + request = asyncio.ensure_future(client.sql("SELECT 1")) + await asyncio.sleep(0.05) + await _cancel_after(client.close(), 0.05) + session = client.server_pool[crate.url]._session + assert session is not None and session.closed + with pytest.raises(exceptions.Error): + await request + + +@pytest.mark.asyncio +async def test_terminate(fake_crate): + """ + `terminate()` relies on aiohttp's private `BaseConnector._close()`; this + fails if an aiohttp release changes it. + """ + crate = await fake_crate() + client = AsyncClient(servers=crate.url) + await client.sql("SELECT 1") + session = client.server_pool[crate.url]._session + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + client.terminate() + assert session.closed + assert client.closed + del client, session + gc.collect() + await asyncio.sleep(0) + assert [str(w.message) for w in caught] == [] + + +@pytest.mark.asyncio +async def test_unclosed_client_warns(fake_crate): + crate = await fake_crate() + client = AsyncClient(servers=crate.url) + await client.sql("SELECT 1") + session = client.server_pool[crate.url]._session + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + del client + gc.collect() + await asyncio.sleep(0) + assert [w.category for w in caught] == [ResourceWarning] + assert "unclosed Date: Mon, 26 Oct 2026 14:45:55 +0100 Subject: [PATCH 08/10] Remove deprecated tests --- tests/client/test_aio_http.py | 215 +++------------------------------- 1 file changed, 18 insertions(+), 197 deletions(-) diff --git a/tests/client/test_aio_http.py b/tests/client/test_aio_http.py index 46a438f6..01b814ce 100644 --- a/tests/client/test_aio_http.py +++ b/tests/client/test_aio_http.py @@ -106,11 +106,6 @@ def test_client_timeout_invalid(timeout): AsyncClient(timeout=timeout) -def test_missing_ca_bundle(): - with pytest.raises(IOError, match="does not exist"): - AsyncClient(ca_cert="/nonexistent/ca.pem") - - def test_creates_no_aiohttp_objects(): client = AsyncClient(servers=["localhost:4200", "localhost:4201"]) assert all(s._session is None for s in client.server_pool.values()) @@ -120,17 +115,6 @@ def test_creates_no_aiohttp_objects(): ] -def test_url_strips_userinfo_and_keeps_path_prefix(): - server = AsyncServer( - "https://crate:s3cr3t@localhost:4200/prefix/", - ssl_context=None, - timeout=client_timeout(None), - pool_size=1, - socket_options=[], - ) - assert str(server._url("/_sql")) == "https://localhost:4200/prefix/_sql" - - def test_tls_context_only_for_https(): client = AsyncClient(servers=["http://localhost:4200", "https://h:4200"]) assert client.server_pool["http://localhost:4200"]._ssl is None @@ -164,46 +148,6 @@ async def test_sql(fake_crate): assert "Cookie" not in request.headers -@pytest.mark.asyncio -async def test_sql_parameters(fake_crate): - crate = await fake_crate() - client = AsyncClient(servers=crate.url) - try: - await client.sql("SELECT ?", parameters=[1]) - await client.sql("INSERT ?", bulk_parameters=[[1], [2]]) - finally: - await client.close() - assert json.loads(crate.requests[0][1]) == {"stmt": "SELECT ?", "args": [1]} - assert json.loads(crate.requests[1][1]) == { - "stmt": "INSERT ?", - "bulk_args": [[1], [2]], - } - - -@pytest.mark.asyncio -async def test_jwt_token(fake_crate): - crate = await fake_crate() - client = AsyncClient(servers=crate.url, jwt_token="t0k3n") - try: - await client.sql("SELECT 1") - finally: - await client.close() - request, _ = crate.requests[0] - assert request.headers["Authorization"] == "Bearer t0k3n" - - -@pytest.mark.asyncio -async def test_jwt_and_username_rejected(fake_crate): - crate = await fake_crate() - client = AsyncClient(servers=crate.url, username="crate", jwt_token="t") - try: - with pytest.raises(exceptions.ProgrammingError, match="JWT"): - await client.sql("SELECT 1") - finally: - await client.close() - assert crate.requests == [] - - @pytest.mark.asyncio async def test_sql_compressed_and_error_trace(fake_crate): crate = await fake_crate() @@ -287,45 +231,6 @@ async def test_server_infos(fake_crate): assert request.path == "/" -@pytest.mark.asyncio -async def test_invalid_json_response(fake_crate): - async def html(request): - return web.Response(text="", content_type="text/html") - - crate = await fake_crate(html) - client = AsyncClient(servers=crate.url) - try: - with pytest.raises( - exceptions.ProgrammingError, match="Invalid server response" - ): - await client.sql("SELECT 1") - finally: - await client.close() - - -@pytest.mark.asyncio -async def test_duplicate_key_maps_to_integrity_error(fake_crate): - async def duplicate(request): - return web.json_response( - { - "error": { - "message": "DuplicateKeyException[A document with the " - "same primary key exists already]", - "code": 4091, - } - }, - status=409, - ) - - crate = await fake_crate(duplicate) - client = AsyncClient(servers=crate.url) - try: - with pytest.raises(exceptions.IntegrityError): - await client.sql("INSERT INTO t (id) VALUES (1)") - finally: - await client.close() - - async def relation_unknown(request): return web.json_response( { @@ -357,10 +262,12 @@ async def slow(request): async def test_same_exception_as_sync( fake_crate, handler, options, exception, match ): + crate = None if handler is None: url = f"http://127.0.0.1:{unused_port()}" else: - url = (await fake_crate(handler)).url + crate = await fake_crate(handler) + url = crate.url def sync_sql(): client = Client(servers=url, **options) @@ -372,12 +279,16 @@ def sync_sql(): with pytest.raises(exception, match=match): await asyncio.to_thread(sync_sql) + sent = len(crate.requests) if crate else 0 client = AsyncClient(servers=url, **options) try: with pytest.raises(exception, match=match): await client.sql("SELECT 1") finally: await client.close() + if crate: + # Error responses, 503 and read timeouts are not retried. + assert len(crate.requests) - sent == 1 @pytest.mark.asyncio @@ -408,18 +319,6 @@ async def test_unavailable_on_given_server(fake_crate): await client.close() -@pytest.mark.asyncio -async def test_read_timeout_is_not_retried(fake_crate): - crate = await fake_crate(slow) - client = AsyncClient(servers=crate.url, timeout=0.2) - try: - with pytest.raises(exceptions.ConnectionError): - await client.sql("SELECT 1") - finally: - await client.close() - assert len(crate.requests) == 1 - - @pytest.mark.asyncio async def test_read_timeout_fails_over(fake_crate): slow_server = await fake_crate(slow) @@ -547,51 +446,15 @@ async def record_sleep(delay): assert sleeps[:4] == [0.0, 0.2, 0.4, 0.8] -@pytest.mark.asyncio -async def test_connect_retry_succeeds(fake_crate, monkeypatch): - crate = await fake_crate() - original = AsyncServer.request - calls = [] - - async def refuse_twice(self, *args, **kwargs): - calls.append(1) - if len(calls) <= 2: - raise aiohttp.ConnectionTimeoutError("connect timed out") - return await original(self, *args, **kwargs) - - monkeypatch.setattr(AsyncServer, "request", refuse_twice) - client = AsyncClient(servers=crate.url) - try: - assert await client.sql("SELECT 1") == SQL_RESULT - assert client.active_servers == [crate.url] - finally: - await client.close() - assert len(calls) == 3 - - -@pytest.mark.asyncio -async def test_tls_failure_is_a_connect_error(): - """ - TLS handshake and certificate errors count as connect errors and are - retried, as urllib3 does. - """ - assert issubclass( - aiohttp.ClientConnectorCertificateError, _http._CONNECT_ERRORS - ) - assert issubclass(aiohttp.ClientConnectorSSLError, _http._CONNECT_ERRORS) - - @pytest.mark.asyncio @pytest.mark.parametrize( "error", [ aiohttp.ServerDisconnectedError(), aiohttp.ClientOSError(errno.ECONNRESET, "Connection reset by peer"), - aiohttp.ClientOSError(errno.EPIPE, "Broken pipe"), ConnectionResetError(errno.ECONNRESET, "Connection reset by peer"), - BrokenPipeError(errno.EPIPE, "Broken pipe"), ], - ids=["disconnected", "oserror-reset", "oserror-pipe", "reset", "pipe"], + ids=["disconnected", "oserror", "reset"], ) async def test_stale_connection_retried_once(fake_crate, monkeypatch, error): crate = await fake_crate() @@ -726,28 +589,6 @@ async def redirect_to_self(request): assert len(loop_server.requests) == 11 -@pytest.mark.asyncio -async def test_redirect_target_unavailable(fake_crate): - """ - As in sync, an error on the redirect target is a `ProgrammingError`. - """ - down = f"http://127.0.0.1:{unused_port()}" - - async def redirect(request): - raise web.HTTPTemporaryRedirect(down + request.path_qs) - - origin = await fake_crate(redirect) - client = AsyncClient(servers=origin.url) - client.connect_retries = 0 - try: - with pytest.raises( - exceptions.ProgrammingError, match="Server not available" - ): - await client.sql("SELECT 1") - finally: - await client.close() - - @pytest.mark.asyncio async def test_tls_handshake(https_server): @@ -969,15 +810,6 @@ async def handler(request): await client.close() -@pytest.mark.asyncio -async def test_close_unused_client(): - client = AsyncClient(servers=["localhost:4200", "localhost:4201"]) - await client.close() - assert client.closed - with pytest.raises(exceptions.ProgrammingError, match="Client closed"): - await client.sql("SELECT 1") - - @pytest.mark.asyncio async def test_cancelled_close_terminates(fake_crate): async def blocked(request): @@ -1018,6 +850,16 @@ async def test_terminate(fake_crate): @pytest.mark.asyncio async def test_unclosed_client_warns(fake_crate): + """ + Only a client that opened connections warns. + """ + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + unused = AsyncClient() + del unused + gc.collect() + assert caught == [] + crate = await fake_crate() client = AsyncClient(servers=crate.url) await client.sql("SELECT 1") @@ -1033,15 +875,6 @@ async def test_unclosed_client_warns(fake_crate): assert session.closed -def test_unused_client_does_not_warn(): - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - client = AsyncClient() - del client - gc.collect() - assert caught == [] - - def test_different_event_loop(): async def start(): crate = await FakeCrate().start() @@ -1116,15 +949,3 @@ async def test_socket_keepalive(fake_crate): assert sock.getsockopt(level, option) == value finally: await client.close() - - -@pytest.mark.asyncio -async def test_socket_keepalive_off(fake_crate): - crate = await fake_crate() - client = AsyncClient(servers=crate.url, socket_keepalive=False) - try: - await client.sql("SELECT 1") - sock = _pooled_socket(client, crate.url) - assert not sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE) - finally: - await client.close() From cecc420dad025039a312c8010244c5043b41728c Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 26 Oct 2026 14:58:09 +0100 Subject: [PATCH 09/10] Fix SSL pipeline failure --- tests/conftest.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index f6325b2c..a92ec32c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,7 @@ import os import platform import socket +import socketserver import ssl import sys import tarfile @@ -232,6 +233,12 @@ class HttpsServer(HTTPServer): CERT_FILE = assets_path("pki/server_valid.pem") CACERT_FILE = assets_path("pki/cacert_valid.pem") + def server_bind(self): + socketserver.TCPServer.server_bind(self) + host, port = self.server_address[:2] + self.server_name = host + self.server_port = port + def get_request(self): # Prepare SSL context. context = ssl._create_unverified_context( # noqa: S323 From 6a66d28777ecbf2b80afc7d4848d9d72ba6508e6 Mon Sep 17 00:00:00 2001 From: Bilal Tonga Date: Mon, 26 Oct 2026 17:11:29 +0100 Subject: [PATCH 10/10] fix duplicated code --- tests/client/test_aio_http.py | 22 +++++++++------------- 1 file changed, 9 insertions(+), 13 deletions(-) diff --git a/tests/client/test_aio_http.py b/tests/client/test_aio_http.py index 01b814ce..0ae5256e 100644 --- a/tests/client/test_aio_http.py +++ b/tests/client/test_aio_http.py @@ -668,6 +668,14 @@ async def worker(): assert max(waits) < run_for / 2 +async def _cancel_after(coro, delay): + task = asyncio.ensure_future(coro) + await asyncio.sleep(delay) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + @pytest.mark.asyncio async def test_cancelled_waiter_frees_its_place(fake_crate): release = asyncio.Event() @@ -681,11 +689,7 @@ async def blocked(request): try: busy = asyncio.ensure_future(client.sql("SELECT 1")) await asyncio.sleep(0.05) - waiter = asyncio.ensure_future(client.sql("SELECT 2")) - await asyncio.sleep(0.05) - waiter.cancel() - with pytest.raises(asyncio.CancelledError): - await waiter + await _cancel_after(client.sql("SELECT 2"), 0.05) release.set() assert await busy == SQL_RESULT assert await asyncio.wait_for(client.sql("SELECT 3"), 5) == SQL_RESULT @@ -696,14 +700,6 @@ async def blocked(request): assert len(crate.requests) == 2 -async def _cancel_after(coro, delay): - task = asyncio.ensure_future(coro) - await asyncio.sleep(delay) - task.cancel() - with pytest.raises(asyncio.CancelledError): - await task - - @pytest.mark.asyncio async def test_cancel_during_tls_handshake(fake_crate): """