diff --git a/ably/http/http.py b/ably/http/http.py index d21a9386..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,7 +129,15 @@ def __init__(self, ably, options): # Cached fallback host (RSC15f) self.__host = None self.__host_expires = None - self.__client = httpx.AsyncClient(http2=True) + self.__clock = select_clock(options) + 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() @@ -147,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 @@ -184,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 @@ -231,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/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..2cb5010c --- /dev/null +++ b/ably/types/testoptions.py @@ -0,0 +1,22 @@ +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. + - `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, 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/submodules b/submodules index dd700951..18c920d4 160000 --- a/submodules +++ b/submodules @@ -1 +1 @@ -Subproject commit dd70095146dba8126e1f27e0407fa453304fb659 +Subproject commit 18c920d4bba7458100c18df81a894f7b069168e5 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() 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)