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) <noreply@anthropic.com>

Author: owenpearson

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()

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

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

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)