From 937537ed1cadc51856a0db36f7cebd5dbc987d27 Mon Sep 17 00:00:00 2001 From: owenpearson Date: Wed, 16 Sep 2026 13:43:43 +0100 Subject: [PATCH 1/3] feat: allow an HTTP transport to be injected via test options The REST client builds its own httpx.AsyncClient, leaving no way for a test to observe or serve the requests it makes. A TestOptions holder passed as a client option supplies an httpx.AsyncBaseTransport which handles requests in place of the network, so request building, host fallback and response parsing all still run against the real code paths. Co-Authored-By: Claude Opus 5 (1M context) --- ably/http/http.py | 9 +++++++- ably/types/options.py | 8 ++++++- ably/types/testoptions.py | 16 +++++++++++++ test/unit/http_test.py | 48 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 79 insertions(+), 2 deletions(-) create mode 100644 ably/types/testoptions.py diff --git a/ably/http/http.py b/ably/http/http.py index d21a9386..8d4d756c 100644 --- a/ably/http/http.py +++ b/ably/http/http.py @@ -129,7 +129,14 @@ def __init__(self, ably, options): # Cached fallback host (RSC15f) self.__host = None self.__host_expires = None - self.__client = httpx.AsyncClient(http2=True) + self.__client = self.__create_client(options) + + @staticmethod + def __create_client(options): + test_options = getattr(options, 'test_options', None) + if test_options is not None and test_options.http_transport is not None: + return httpx.AsyncClient(transport=test_options.http_transport) + return httpx.AsyncClient(http2=True) async def close(self): await self.__client.aclose() diff --git a/ably/types/options.py b/ably/types/options.py index 1dad41fb..86af021f 100644 --- a/ably/types/options.py +++ b/ably/types/options.py @@ -34,7 +34,8 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti idempotent_rest_publishing=None, loop=None, auto_connect=True, suspended_retry_timeout=None, connectivity_check_url=None, channel_retry_timeout=Defaults.channel_retry_timeout, add_request_ids=False, - vcdiff_decoder: VCDiffDecoder = None, transport_params=None, **kwargs): + vcdiff_decoder: VCDiffDecoder = None, transport_params=None, test_options=None, + **kwargs): super().__init__(**kwargs) @@ -129,6 +130,7 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti self.__add_request_ids = add_request_ids self.__vcdiff_decoder = vcdiff_decoder self.__transport_params = transport_params or {} + self.__test_options = test_options self.__hosts = self.__get_hosts() @property @@ -306,6 +308,10 @@ def vcdiff_decoder(self): def transport_params(self): return self.__transport_params + @property + def test_options(self): + return self.__test_options + def __get_hosts(self): """ Return the list of hosts as they should be tried. First comes the main diff --git a/ably/types/testoptions.py b/ably/types/testoptions.py new file mode 100644 index 00000000..87ba98ef --- /dev/null +++ b/ably/types/testoptions.py @@ -0,0 +1,16 @@ +class TestOptions: + """Substitutes for the infrastructure a client uses to perform I/O. + + Hooks left as None use the production implementation. + + :Parameters: + - `http_transport`: an `httpx.AsyncBaseTransport` which handles every + HTTP request the client makes, in place of the network. + """ + + # Excludes the class from pytest collection, which would otherwise treat + # any module importing it as declaring a test suite. + __test__ = False + + def __init__(self, http_transport=None): + self.http_transport = http_transport diff --git a/test/unit/http_test.py b/test/unit/http_test.py index 61e0d35e..0f30f251 100644 --- a/test/unit/http_test.py +++ b/test/unit/http_test.py @@ -1,4 +1,9 @@ +import httpx +import pytest + from ably import AblyRest +from ably.types.testoptions import TestOptions +from ably.util.exceptions import AblyException def test_http_get_rest_hosts_works_when_fallback_realtime_host_is_set(): @@ -17,3 +22,46 @@ def test_http_get_rest_hosts_works_when_fallback_realtime_host_is_not_set(): hosts = ably.http.get_hosts() assert isinstance(hosts, list) assert all(isinstance(host, str) for host in hosts) + + +class RecordingTransport(httpx.AsyncBaseTransport): + def __init__(self, response_factory): + self.requests = [] + self.__response_factory = response_factory + + async def handle_async_request(self, request): + self.requests.append(request) + return self.__response_factory(request) + + +async def test_http_sends_requests_through_an_injected_transport(): + transport = RecordingTransport(lambda request: httpx.Response(200, json=[1500000000000])) + ably = AblyRest(token="foo", test_options=TestOptions(http_transport=transport)) + + server_time = await ably.time() + + assert server_time == 1500000000000 + assert len(transport.requests) == 1 + assert transport.requests[0].method == 'GET' + assert transport.requests[0].url.path == '/time' + await ably.close() + + +async def test_http_surfaces_transport_connection_errors(): + def refuse(request): + raise httpx.ConnectError("connection refused", request=request) + + transport = RecordingTransport(refuse) + ably = AblyRest(token="foo", test_options=TestOptions(http_transport=transport)) + + with pytest.raises(AblyException): + await ably.time() + + # Every host is tried before the error is surfaced (RSC15l) + assert len(transport.requests) == len(ably.http.get_hosts()) + await ably.close() + + +def test_http_uses_a_network_transport_without_test_options(): + ably = AblyRest(token="foo") + assert isinstance(ably.http._Http__client._transport, httpx.AsyncHTTPTransport) From 3f1776ca12314856192d883127e930e19f070151 Mon Sep 17 00:00:00 2001 From: owenpearson Date: Wed, 16 Sep 2026 16:01:19 +0100 Subject: [PATCH 2/3] chore: update ably-common to a revision carrying the msgpack fixtures The msgpack interop specifications read test-resources/msgpack_test_fixtures.json, which the pinned revision predated, so those tests were gated on the file being present. They now run. Co-Authored-By: Claude Opus 5 (1M context) --- submodules | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/submodules b/submodules index dd700951..18c920d4 160000 --- a/submodules +++ b/submodules @@ -1 +1 @@ -Subproject commit dd70095146dba8126e1f27e0407fa453304fb659 +Subproject commit 18c920d4bba7458100c18df81a894f7b069168e5 From abd194df1d9b4beafd797d9957fb1c70a2b8ef42 Mon Sep 17 00:00:00 2001 From: owenpearson Date: Fri, 25 Sep 2026 19:27:10 +0100 Subject: [PATCH 3/3] feat: read the current time through a clock which tests can replace The client reads the wall clock for the fallback-host cache, the cumulative retry budget and the timestamp a token request is signed with, so a test which drives any of them has to spend the interval in real seconds. A Clock passed as a test option supplies now_ms, monotonic_ms and timer together, so that substituting one moves every reading the client makes; clients which supply none read the system clock. The cached fallback host's expiry is held in milliseconds, matching fallbackRetryTimeout and the rest of the library, and the retry budget is converted where it is compared, since httpMaxRetryDuration is in seconds. Co-Authored-By: Claude Opus 5 (1M context) --- ably/http/http.py | 11 ++-- ably/rest/auth.py | 5 +- ably/types/testoptions.py | 8 ++- ably/util/clock.py | 41 +++++++++++++ test/unit/clock_test.py | 124 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 181 insertions(+), 8 deletions(-) create mode 100644 ably/util/clock.py create mode 100644 test/unit/clock_test.py diff --git a/ably/http/http.py b/ably/http/http.py index 8d4d756c..727c6b35 100644 --- a/ably/http/http.py +++ b/ably/http/http.py @@ -1,7 +1,6 @@ import functools import json import logging -import time from urllib.parse import urljoin import httpx @@ -10,6 +9,7 @@ from ably.http.httputils import HttpUtils from ably.rest.auth import Auth from ably.transport.defaults import Defaults +from ably.util.clock import select_clock from ably.util.exceptions import AblyException from ably.util.helper import extract_url_params, is_token_error @@ -129,6 +129,7 @@ def __init__(self, ably, options): # Cached fallback host (RSC15f) self.__host = None self.__host_expires = None + self.__clock = select_clock(options) self.__client = self.__create_client(options) @staticmethod @@ -154,7 +155,7 @@ def get_hosts(self): return hosts # unstore saved fallback host after fallbackRetryTimeout (RSC15f) - if self.__host_expires is not None and time.time() > self.__host_expires: + if self.__host_expires is not None and self.__clock.now_ms() > self.__host_expires: self.__host = None self.__host_expires = None return hosts @@ -191,12 +192,12 @@ async def make_request(self, method, path, version=None, headers=None, body=None timeout = (self.http_open_timeout, self.http_request_timeout) http_max_retry_duration = self.http_max_retry_duration - requested_at = time.time() + requested_at = self.__clock.now_ms() hosts = self.get_hosts() for retry_count, host in enumerate(hosts): def should_stop_retrying(retry_count=retry_count): - time_passed = time.time() - requested_at + time_passed = (self.__clock.now_ms() - requested_at) / 1000 # if it's the last try or cumulative timeout is done, we stop retrying return retry_count == len(hosts) - 1 or time_passed > http_max_retry_duration @@ -238,7 +239,7 @@ def should_stop_retrying(retry_count=retry_count): # Keep fallback host for later (RSC15f) if retry_count > 0 and host != self.options.get_host(): self.__host = host - self.__host_expires = time.time() + (self.options.fallback_retry_timeout / 1000.0) + self.__host_expires = self.__clock.now_ms() + self.options.fallback_retry_timeout return Response(response) except AblyException as e: diff --git a/ably/rest/auth.py b/ably/rest/auth.py index d2057533..ddf70518 100644 --- a/ably/rest/auth.py +++ b/ably/rest/auth.py @@ -2,7 +2,6 @@ import base64 import logging -import time import uuid from datetime import timedelta from typing import TYPE_CHECKING @@ -18,6 +17,7 @@ from ably.types.capability import Capability from ably.types.tokendetails import TokenDetails from ably.types.tokenrequest import TokenRequest +from ably.util.clock import select_clock from ably.util.exceptions import AblyAuthException, AblyException, IncompatibleClientIdException from ably.util.helper import extract_url_params @@ -34,6 +34,7 @@ class Method: def __init__(self, ably: AblyRest | AblyRealtime, options: Options): self.__ably = ably self.__auth_options = options + self.__clock = select_clock(options) if not self.ably._is_realtime: self.__client_id = options.client_id @@ -383,7 +384,7 @@ async def _get_auth_headers(self): def _timestamp(self): """Returns the local time in milliseconds since the unix epoch""" - return int(time.time() * 1000) + return self.__clock.now_ms() def _random_nonce(self): return uuid.uuid4().hex[:16] diff --git a/ably/types/testoptions.py b/ably/types/testoptions.py index 87ba98ef..2cb5010c 100644 --- a/ably/types/testoptions.py +++ b/ably/types/testoptions.py @@ -6,11 +6,17 @@ class TestOptions: :Parameters: - `http_transport`: an `httpx.AsyncBaseTransport` which handles every HTTP request the client makes, in place of the network. + - `clock`: the source of time the client reads, in place of + `ably.util.clock.Clock`. It supplies `now_ms()`, `monotonic_ms()` and + `timer(timeout_ms, callback)`, where `callback` is either a coroutine + function or a plain callable and the return value has a `cancel()` + method. """ # Excludes the class from pytest collection, which would otherwise treat # any module importing it as declaring a test suite. __test__ = False - def __init__(self, http_transport=None): + def __init__(self, http_transport=None, clock=None): self.http_transport = http_transport + self.clock = clock diff --git a/ably/util/clock.py b/ably/util/clock.py new file mode 100644 index 00000000..e014dc2e --- /dev/null +++ b/ably/util/clock.py @@ -0,0 +1,41 @@ +import time +from typing import Callable + +from ably.util.helper import Timer + + +class Clock: + """The source of time for every decision a client makes from it. + + Token expiry, the server-time offset and the fallback-host cache all read + the clock, and delayed callbacks are scheduled through it, so replacing one + moves all of them together. + + `now_ms` is the time of day and may step; `monotonic_ms` only ever moves + forward and is what a duration is measured with. + """ + + def now_ms(self) -> int: + """The milliseconds elapsed since the unix epoch.""" + return round(time.time_ns() / 1_000_000) + + def monotonic_ms(self) -> float: + """The milliseconds elapsed since an arbitrary fixed point.""" + return time.monotonic() * 1000 + + def timer(self, timeout: float, callback: Callable) -> Timer: + """Schedules `callback` for `timeout` milliseconds from now.""" + return Timer(timeout, callback) + + +def select_clock(options) -> Clock: + """The clock a client reads its time from. + + `TestOptions.clock` substitutes for the real one during tests, letting them + drive time-dependent behaviour without waiting for it. Clients which supply + none get `Clock`. + """ + test_options = getattr(options, 'test_options', None) + if test_options is not None and test_options.clock is not None: + return test_options.clock + return Clock() diff --git a/test/unit/clock_test.py b/test/unit/clock_test.py new file mode 100644 index 00000000..75695c7c --- /dev/null +++ b/test/unit/clock_test.py @@ -0,0 +1,124 @@ +import httpx +import pytest + +from ably import AblyRest +from ably.types.testoptions import TestOptions +from ably.util.clock import Clock +from ably.util.exceptions import AblyException + +# A plausible wall-clock reading for a fake to start from, so that a time taken +# against it looks like a real one. +EPOCH_MS = 1_700_000_000_000 + + +class FakeClock: + """A clock whose notional time only the test moves. + + `advance` moves it by hand; `step_ms` moves it by that much on every read, + which is how a duration measured across several reads is driven. + """ + + def __init__(self, now_ms=EPOCH_MS, step_ms=0): + self.__now_ms = now_ms + self.__step_ms = step_ms + + def now_ms(self): + reading = self.__now_ms + self.__now_ms += self.__step_ms + return reading + + def monotonic_ms(self): + return float(self.__now_ms) + + def advance(self, milliseconds): + self.__now_ms += milliseconds + + def timer(self, timeout, callback): + raise AssertionError('the REST client schedules no delayed callbacks') + + +class RecordingTransport(httpx.AsyncBaseTransport): + """Records the host of every request and refuses the ones in `refusing`. + + A host left out of `refusing` is served a server time. `refusing` is set + after the client is built, since the host to refuse is one of its own. + """ + + def __init__(self): + self.refusing = () + self.hosts = [] + + async def handle_async_request(self, request): + self.hosts.append(request.url.host) + if request.url.host in self.refusing: + raise httpx.ConnectError('connection refused', request=request) + return httpx.Response(200, json=[1500000000000]) + + +def test_auth_timestamps_from_the_injected_clock(): + ably = AblyRest(key='name:secret', test_options=TestOptions(clock=FakeClock(1_500_000_000_000))) + assert ably.auth._timestamp() == 1_500_000_000_000 + + +def test_a_client_without_test_options_reads_the_real_clock(): + ably = AblyRest(token='foo') + assert isinstance(ably.auth._Auth__clock, Clock) + assert isinstance(ably.http._Http__clock, Clock) + + +async def test_the_cached_fallback_host_expires_on_the_clock(): + # RSC15f: the cache lasts fallback_retry_timeout milliseconds + clock = FakeClock() + transport = RecordingTransport() + ably = AblyRest(token='foo', fallback_retry_timeout=2000, + test_options=TestOptions(http_transport=transport, clock=clock)) + primary = ably.options.get_host() + transport.refusing = (primary,) + + await ably.time() + + fallback = transport.hosts[-1] + assert fallback != primary + assert ably.http._Http__host_expires == EPOCH_MS + 2000 + assert ably.http.get_hosts()[0] == fallback + + clock.advance(1999) + assert ably.http.get_hosts()[0] == fallback + + clock.advance(2) + assert ably.http.get_hosts()[0] == primary + assert ably.http._Http__host_expires is None + + await ably.close() + + +async def test_every_host_is_tried_while_the_retry_budget_lasts(): + # A second of clock time per reading, against the default 15 second budget + clock = FakeClock(step_ms=1000) + transport = RecordingTransport() + ably = AblyRest(token='foo', test_options=TestOptions(http_transport=transport, clock=clock)) + hosts = ably.http.get_hosts() + transport.refusing = tuple(hosts) + + with pytest.raises(AblyException): + await ably.time() + + assert transport.hosts == hosts + + await ably.close() + + +async def test_retrying_stops_once_the_retry_budget_is_spent(): + # RSC15l: http_max_retry_duration is seconds where the clock is milliseconds + clock = FakeClock(step_ms=1000) + transport = RecordingTransport() + ably = AblyRest(token='foo', http_max_retry_duration=0.5, + test_options=TestOptions(http_transport=transport, clock=clock)) + transport.refusing = tuple(ably.http.get_hosts()) + + with pytest.raises(AblyException): + await ably.time() + + assert len(transport.hosts) == 1 + + await ably.close()