Skip to content

Commit 2d4f17f

Browse files
authored
Merge pull request #695 from ably/uts/http-transport-seam
feat: allow an HTTP transport to be injected via test options
2 parents c5e4f37 + abd194d commit 2d4f17f

8 files changed

Lines changed: 260 additions & 10 deletions

File tree

‎ably/http/http.py‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import functools
22
import json
33
import logging
4-
import time
54
from urllib.parse import urljoin
65

76
import httpx
@@ -10,6 +9,7 @@
109
from ably.http.httputils import HttpUtils
1110
from ably.rest.auth import Auth
1211
from ably.transport.defaults import Defaults
12+
from ably.util.clock import select_clock
1313
from ably.util.exceptions import AblyException
1414
from ably.util.helper import extract_url_params, is_token_error
1515

@@ -129,7 +129,15 @@ def __init__(self, ably, options):
129129
# Cached fallback host (RSC15f)
130130
self.__host = None
131131
self.__host_expires = None
132-
self.__client = httpx.AsyncClient(http2=True)
132+
self.__clock = select_clock(options)
133+
self.__client = self.__create_client(options)
134+
135+
@staticmethod
136+
def __create_client(options):
137+
test_options = getattr(options, 'test_options', None)
138+
if test_options is not None and test_options.http_transport is not None:
139+
return httpx.AsyncClient(transport=test_options.http_transport)
140+
return httpx.AsyncClient(http2=True)
133141

134142
async def close(self):
135143
await self.__client.aclose()
@@ -147,7 +155,7 @@ def get_hosts(self):
147155
return hosts
148156

149157
# unstore saved fallback host after fallbackRetryTimeout (RSC15f)
150-
if self.__host_expires is not None and time.time() > self.__host_expires:
158+
if self.__host_expires is not None and self.__clock.now_ms() > self.__host_expires:
151159
self.__host = None
152160
self.__host_expires = None
153161
return hosts
@@ -184,12 +192,12 @@ async def make_request(self, method, path, version=None, headers=None, body=None
184192

185193
timeout = (self.http_open_timeout, self.http_request_timeout)
186194
http_max_retry_duration = self.http_max_retry_duration
187-
requested_at = time.time()
195+
requested_at = self.__clock.now_ms()
188196

189197
hosts = self.get_hosts()
190198
for retry_count, host in enumerate(hosts):
191199
def should_stop_retrying(retry_count=retry_count):
192-
time_passed = time.time() - requested_at
200+
time_passed = (self.__clock.now_ms() - requested_at) / 1000
193201
# if it's the last try or cumulative timeout is done, we stop retrying
194202
return retry_count == len(hosts) - 1 or time_passed > http_max_retry_duration
195203

@@ -231,7 +239,7 @@ def should_stop_retrying(retry_count=retry_count):
231239
# Keep fallback host for later (RSC15f)
232240
if retry_count > 0 and host != self.options.get_host():
233241
self.__host = host
234-
self.__host_expires = time.time() + (self.options.fallback_retry_timeout / 1000.0)
242+
self.__host_expires = self.__clock.now_ms() + self.options.fallback_retry_timeout
235243

236244
return Response(response)
237245
except AblyException as e:

‎ably/rest/auth.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
import base64
44
import logging
5-
import time
65
import uuid
76
from datetime import timedelta
87
from typing import TYPE_CHECKING
@@ -18,6 +17,7 @@
1817
from ably.types.capability import Capability
1918
from ably.types.tokendetails import TokenDetails
2019
from ably.types.tokenrequest import TokenRequest
20+
from ably.util.clock import select_clock
2121
from ably.util.exceptions import AblyAuthException, AblyException, IncompatibleClientIdException
2222
from ably.util.helper import extract_url_params
2323

@@ -34,6 +34,7 @@ class Method:
3434
def __init__(self, ably: AblyRest | AblyRealtime, options: Options):
3535
self.__ably = ably
3636
self.__auth_options = options
37+
self.__clock = select_clock(options)
3738

3839
if not self.ably._is_realtime:
3940
self.__client_id = options.client_id
@@ -383,7 +384,7 @@ async def _get_auth_headers(self):
383384

384385
def _timestamp(self):
385386
"""Returns the local time in milliseconds since the unix epoch"""
386-
return int(time.time() * 1000)
387+
return self.__clock.now_ms()
387388

388389
def _random_nonce(self):
389390
return uuid.uuid4().hex[:16]

‎ably/types/options.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,8 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti
3434
idempotent_rest_publishing=None, loop=None, auto_connect=True,
3535
suspended_retry_timeout=None, connectivity_check_url=None,
3636
channel_retry_timeout=Defaults.channel_retry_timeout, add_request_ids=False,
37-
vcdiff_decoder: VCDiffDecoder = None, transport_params=None, **kwargs):
37+
vcdiff_decoder: VCDiffDecoder = None, transport_params=None, test_options=None,
38+
**kwargs):
3839

3940
super().__init__(**kwargs)
4041

@@ -129,6 +130,7 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti
129130
self.__add_request_ids = add_request_ids
130131
self.__vcdiff_decoder = vcdiff_decoder
131132
self.__transport_params = transport_params or {}
133+
self.__test_options = test_options
132134
self.__hosts = self.__get_hosts()
133135

134136
@property
@@ -306,6 +308,10 @@ def vcdiff_decoder(self):
306308
def transport_params(self):
307309
return self.__transport_params
308310

311+
@property
312+
def test_options(self):
313+
return self.__test_options
314+
309315
def __get_hosts(self):
310316
"""
311317
Return the list of hosts as they should be tried. First comes the main

‎ably/types/testoptions.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
class TestOptions:
2+
"""Substitutes for the infrastructure a client uses to perform I/O.
3+
4+
Hooks left as None use the production implementation.
5+
6+
:Parameters:
7+
- `http_transport`: an `httpx.AsyncBaseTransport` which handles every
8+
HTTP request the client makes, in place of the network.
9+
- `clock`: the source of time the client reads, in place of
10+
`ably.util.clock.Clock`. It supplies `now_ms()`, `monotonic_ms()` and
11+
`timer(timeout_ms, callback)`, where `callback` is either a coroutine
12+
function or a plain callable and the return value has a `cancel()`
13+
method.
14+
"""
15+
16+
# Excludes the class from pytest collection, which would otherwise treat
17+
# any module importing it as declaring a test suite.
18+
__test__ = False
19+
20+
def __init__(self, http_transport=None, clock=None):
21+
self.http_transport = http_transport
22+
self.clock = clock

‎ably/util/clock.py‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
import time
2+
from typing import Callable
3+
4+
from ably.util.helper import Timer
5+
6+
7+
class Clock:
8+
"""The source of time for every decision a client makes from it.
9+
10+
Token expiry, the server-time offset and the fallback-host cache all read
11+
the clock, and delayed callbacks are scheduled through it, so replacing one
12+
moves all of them together.
13+
14+
`now_ms` is the time of day and may step; `monotonic_ms` only ever moves
15+
forward and is what a duration is measured with.
16+
"""
17+
18+
def now_ms(self) -> int:
19+
"""The milliseconds elapsed since the unix epoch."""
20+
return round(time.time_ns() / 1_000_000)
21+
22+
def monotonic_ms(self) -> float:
23+
"""The milliseconds elapsed since an arbitrary fixed point."""
24+
return time.monotonic() * 1000
25+
26+
def timer(self, timeout: float, callback: Callable) -> Timer:
27+
"""Schedules `callback` for `timeout` milliseconds from now."""
28+
return Timer(timeout, callback)
29+
30+
31+
def select_clock(options) -> Clock:
32+
"""The clock a client reads its time from.
33+
34+
`TestOptions.clock` substitutes for the real one during tests, letting them
35+
drive time-dependent behaviour without waiting for it. Clients which supply
36+
none get `Clock`.
37+
"""
38+
test_options = getattr(options, 'test_options', None)
39+
if test_options is not None and test_options.clock is not None:
40+
return test_options.clock
41+
return Clock()

‎submodules‎

Submodule submodules updated 343 files

‎test/unit/clock_test.py‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
import httpx
2+
import pytest
3+
4+
from ably import AblyRest
5+
from ably.types.testoptions import TestOptions
6+
from ably.util.clock import Clock
7+
from ably.util.exceptions import AblyException
8+
9+
# A plausible wall-clock reading for a fake to start from, so that a time taken
10+
# against it looks like a real one.
11+
EPOCH_MS = 1_700_000_000_000
12+
13+
14+
class FakeClock:
15+
"""A clock whose notional time only the test moves.
16+
17+
`advance` moves it by hand; `step_ms` moves it by that much on every read,
18+
which is how a duration measured across several reads is driven.
19+
"""
20+
21+
def __init__(self, now_ms=EPOCH_MS, step_ms=0):
22+
self.__now_ms = now_ms
23+
self.__step_ms = step_ms
24+
25+
def now_ms(self):
26+
reading = self.__now_ms
27+
self.__now_ms += self.__step_ms
28+
return reading
29+
30+
def monotonic_ms(self):
31+
return float(self.__now_ms)
32+
33+
def advance(self, milliseconds):
34+
self.__now_ms += milliseconds
35+
36+
def timer(self, timeout, callback):
37+
raise AssertionError('the REST client schedules no delayed callbacks')
38+
39+
40+
class RecordingTransport(httpx.AsyncBaseTransport):
41+
"""Records the host of every request and refuses the ones in `refusing`.
42+
43+
A host left out of `refusing` is served a server time. `refusing` is set
44+
after the client is built, since the host to refuse is one of its own.
45+
"""
46+
47+
def __init__(self):
48+
self.refusing = ()
49+
self.hosts = []
50+
51+
async def handle_async_request(self, request):
52+
self.hosts.append(request.url.host)
53+
if request.url.host in self.refusing:
54+
raise httpx.ConnectError('connection refused', request=request)
55+
return httpx.Response(200, json=[1500000000000])
56+
57+
58+
def test_auth_timestamps_from_the_injected_clock():
59+
ably = AblyRest(key='name:secret', test_options=TestOptions(clock=FakeClock(1_500_000_000_000)))
60+
assert ably.auth._timestamp() == 1_500_000_000_000
61+
62+
63+
def test_a_client_without_test_options_reads_the_real_clock():
64+
ably = AblyRest(token='foo')
65+
assert isinstance(ably.auth._Auth__clock, Clock)
66+
assert isinstance(ably.http._Http__clock, Clock)
67+
68+
69+
async def test_the_cached_fallback_host_expires_on_the_clock():
70+
# RSC15f: the cache lasts fallback_retry_timeout milliseconds
71+
clock = FakeClock()
72+
transport = RecordingTransport()
73+
ably = AblyRest(token='foo', fallback_retry_timeout=2000,
74+
test_options=TestOptions(http_transport=transport, clock=clock))
75+
primary = ably.options.get_host()
76+
transport.refusing = (primary,)
77+
78+
await ably.time()
79+
80+
fallback = transport.hosts[-1]
81+
assert fallback != primary
82+
assert ably.http._Http__host_expires == EPOCH_MS + 2000
83+
assert ably.http.get_hosts()[0] == fallback
84+
85+
clock.advance(1999)
86+
assert ably.http.get_hosts()[0] == fallback
87+
88+
clock.advance(2)
89+
assert ably.http.get_hosts()[0] == primary
90+
assert ably.http._Http__host_expires is None
91+
92+
await ably.close()
93+
94+
95+
async def test_every_host_is_tried_while_the_retry_budget_lasts():
96+
# A second of clock time per reading, against the default 15 second budget
97+
clock = FakeClock(step_ms=1000)
98+
transport = RecordingTransport()
99+
ably = AblyRest(token='foo', test_options=TestOptions(http_transport=transport, clock=clock))
100+
hosts = ably.http.get_hosts()
101+
transport.refusing = tuple(hosts)
102+
103+
with pytest.raises(AblyException):
104+
await ably.time()
105+
106+
assert transport.hosts == hosts
107+
108+
await ably.close()
109+
110+
111+
async def test_retrying_stops_once_the_retry_budget_is_spent():
112+
# RSC15l: http_max_retry_duration is seconds where the clock is milliseconds
113+
clock = FakeClock(step_ms=1000)
114+
transport = RecordingTransport()
115+
ably = AblyRest(token='foo', http_max_retry_duration=0.5,
116+
test_options=TestOptions(http_transport=transport, clock=clock))
117+
transport.refusing = tuple(ably.http.get_hosts())
118+
119+
with pytest.raises(AblyException):
120+
await ably.time()
121+
122+
assert len(transport.hosts) == 1
123+
124+
await ably.close()

‎test/unit/http_test.py‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,9 @@
1+
import httpx
2+
import pytest
3+
14
from ably import AblyRest
5+
from ably.types.testoptions import TestOptions
6+
from ably.util.exceptions import AblyException
27

38

49
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():
1722
hosts = ably.http.get_hosts()
1823
assert isinstance(hosts, list)
1924
assert all(isinstance(host, str) for host in hosts)
25+
26+
27+
class RecordingTransport(httpx.AsyncBaseTransport):
28+
def __init__(self, response_factory):
29+
self.requests = []
30+
self.__response_factory = response_factory
31+
32+
async def handle_async_request(self, request):
33+
self.requests.append(request)
34+
return self.__response_factory(request)
35+
36+
37+
async def test_http_sends_requests_through_an_injected_transport():
38+
transport = RecordingTransport(lambda request: httpx.Response(200, json=[1500000000000]))
39+
ably = AblyRest(token="foo", test_options=TestOptions(http_transport=transport))
40+
41+
server_time = await ably.time()
42+
43+
assert server_time == 1500000000000
44+
assert len(transport.requests) == 1
45+
assert transport.requests[0].method == 'GET'
46+
assert transport.requests[0].url.path == '/time'
47+
await ably.close()
48+
49+
50+
async def test_http_surfaces_transport_connection_errors():
51+
def refuse(request):
52+
raise httpx.ConnectError("connection refused", request=request)
53+
54+
transport = RecordingTransport(refuse)
55+
ably = AblyRest(token="foo", test_options=TestOptions(http_transport=transport))
56+
57+
with pytest.raises(AblyException):
58+
await ably.time()
59+
60+
# Every host is tried before the error is surfaced (RSC15l)
61+
assert len(transport.requests) == len(ably.http.get_hosts())
62+
await ably.close()
63+
64+
65+
def test_http_uses_a_network_transport_without_test_options():
66+
ably = AblyRest(token="foo")
67+
assert isinstance(ably.http._Http__client._transport, httpx.AsyncHTTPTransport)

0 commit comments

Comments
 (0)