diff --git a/.claude/skills/uts-to-python/SKILL.md b/.claude/skills/uts-to-python/SKILL.md
new file mode 100644
index 00000000..3dceac07
--- /dev/null
+++ b/.claude/skills/uts-to-python/SKILL.md
@@ -0,0 +1,190 @@
+---
+description: Translate Universal Test Specifications from ably/specification into ably-python tests. Use when deriving, updating or evaluating tests under test/uts.
+allowed-tools: Bash, Read, Edit, Write, WebFetch
+---
+
+# Translating UTS specs into ably-python tests
+
+## Sources
+
+Fetch both fresh at the start of every run; do not work from memory.
+
+```bash
+gh api repos/ably/specification/contents/uts/docs/writing-derived-tests.md --jq '.content' | base64 -d
+gh api repos/ably/specification/contents/uts/rest/unit/<spec>.md --jq '.content' | base64 -d
+```
+
+`writing-derived-tests.md` governs. This file covers only what is particular to ably-python.
+
+## Layout
+
+A spec at `uts/<tier path>/<name>.md` becomes `test/uts/<tier path>/<name>_test.py`, so
+`uts/rest/unit/auth/token_renewal.md` becomes `test/uts/rest/unit/auth/token_renewal_test.py`.
+Every directory needs an `__init__.py`, as `test` is a package.
+
+`test/uts/rest/unit/time_test.py` is the reference example. Follow its shape.
+
+## Anatomy of a derived test
+
+```python
+"""Derived from uts/rest/unit/time.md in ably/specification.
+
+Spec points: RSC16
+"""
+
+# UTS: rest/unit/RSC16/returns-server-time-0
+async def test_rsc16_time_returns_server_time():
+    captured_requests = []
+
+    def on_request(request):
+        captured_requests.append(request)
+        request.respond_with(200, [SERVER_TIME_MS])
+
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=on_request,
+    )
+    client = rest_client(mock_http)
+
+    result = await client.time()
+
+    assert result == SERVER_TIME_MS
+```
+
+- The `# UTS:` comment carries the spec's Test ID verbatim, immediately above the function.
+- The function name is the spec point plus the Test ID slug: `rest/unit/RSC16/returns-server-time-0`
+  becomes `test_rsc16_time_returns_server_time`. Drop the trailing index.
+- `pytest` runs with `asyncio_mode = "auto"`, so async tests take no decorator.
+- Keep `captured_requests` a local list appended from the handler, as the specs do. Do not
+  reach for `mock_http.captured_requests`.
+- Always pass `on_connection_attempt`, even where it only succeeds.
+- Do not close clients. `test/uts/conftest.py` closes them after every test.
+
+## Mapping pseudocode to ably-python
+
+| Pseudocode | ably-python |
+|---|---|
+| `install_mock(m)` + `Rest(options: ClientOptions(...))` | `rest_client(m, ...)` from `test.uts.helpers.client` |
+| `uninstall_mock()` | nothing; the fixture closes clients |
+| `ClientOptions(key: "app.key:secret")` | the default in `rest_client`; pass nothing |
+| `useBinaryProtocol` / `useTokenAuth` / `tls` | `use_binary_protocol` / `use_token_auth` / `tls` |
+| `AWAIT client.time()` | `await client.time()` |
+| `... FAILS WITH error` | `with pytest.raises(AblyException) as excinfo:` |
+| `error.code` / `error.statusCode` / `error.message` | `excinfo.value.code` / `.status_code` / `.message` |
+| `request.url.queryParams` / `queryParameters` | `request.url.query_params` (spec drift; one concept) |
+| `parse_json(request.body)` | `json.loads(request.body)` |
+| `msgpack_decode(x)` / `msgpack_encode(x)` | `msgpack.unpackb(x)` / `msgpack.packb(x, use_bin_type=False)` |
+| `process_pending_events()` | `await asyncio.sleep(0)` |
+| `enable_fake_timers()` / `ADVANCE_TIME(ms)` | no equivalent; see Timers below |
+
+Client options are snake_case throughout. Check the actual signature in
+`ably/types/options.py` before assuming an option exists.
+
+## The mock
+
+`test/uts/helpers/mock_http.py`, matching `uts/rest/unit/helpers/mock_http.md`. Read it.
+Names match the pseudocode exactly.
+
+Every call raises a `PendingConnection`, then a `PendingRequest` only if the connection
+succeeded. A failed connection records nothing in `captured_requests`.
+
+`PendingConnection`: `host`, `port`, `tls`, `timestamp`; `respond_with_success()`,
+`respond_with_refused()`, `respond_with_timeout()`, `respond_with_dns_error()`.
+
+`PendingRequest`: `method`, `url` (`scheme`, `host`, `port`, `path`, `query_params`),
+`path`, `headers` (case-insensitive), `body` (raw bytes), `timestamp`;
+`respond_with(status, body, headers)`, `respond_with_delay(ms, status, body, headers)`,
+`respond_with_timeout()`.
+
+`MockHttpClient` also carries `captured_requests`, `await_request(timeout)`,
+`await_connection_attempt(timeout)`, `reset()`, and the queue family
+(`queue_response`, `queue_responses`, `queue_timeout`, `queue_delayed_response`,
+`queue_response_for_host`, `queue_response_for_url`). Handlers are reassignable.
+
+A response body given as a dict or list is encoded to match what the client asked for,
+so specs that exercise the binary protocol need no special handling. Pass `bytes` to
+control the encoding yourself, alongside an explicit `Content-Type`.
+
+## ably-python traits that catch translations out
+
+- **The binary protocol is the default.** `use_binary_protocol` defaults to `True`, so
+  requests carry `Accept: application/x-msgpack` and bodies are msgpack. Decode request
+  bodies with `msgpack.unpackb` unless the spec sets `use_binary_protocol=False`.
+- **5xx and CloudFront responses retry across every host.** A handler that always answers
+  500 is called once per host, not once. Count requests accordingly, or branch on a counter.
+- **`client.request()` requires `version`.** It is not optional.
+- **`client.time()` returns milliseconds as a number**, not a datetime. The spec requirement
+  admits "a DateTime or timestamp", so this is idiomatic, not a deviation.
+- **`AblyException` carries `code`, `status_code` and `message`**, and `@catch_all` wraps
+  several public methods, so a transport error surfaces as `AblyException` with code 50000.
+- **A response with no `Content-Type` raises** `ValueError` from `Response.to_native()`, and
+  `PaginatedResult` reads the header unguarded. Set `Content-Type` on any body passed as
+  `bytes` or `str` that the client is meant to decode.
+
+## Traps found while deriving the REST unit specs
+
+- **`/time` returns an array.** Several specs stub it as `{"time": N}`; the endpoint
+  and `time.md` both use `[N]`, and `AblyRest.time()` indexes it. Stub `[N]`.
+- **A single queued response is consumed by the first host.** A 5xx or CloudFront
+  response sends the client to the next fallback, so queue one response per host
+  (`queue_responses(3, ...)`) or answer from a handler.
+- **`"encoding": null` crashes decoding** in `Message`, `PresenceMessage` and
+  `Annotation` — `obj.get('encoding', '')` returns `None` when the key is present
+  and null. Expect `AttributeError: 'NoneType' object has no attribute 'strip'`.
+- **`msgpack.packb(..., use_bin_type=True)`** is required for a payload that must
+  arrive as msgpack `bin` rather than `str`. The mock's automatic encoding uses
+  `use_bin_type=False`, so encode such a body yourself with an explicit
+  `Content-Type`.
+- **`auth_url` requests bypass the injected transport.** `Auth.token_request_from_auth_url`
+  builds its own `httpx.AsyncClient`, so a spec driving `auth_url` cannot be observed
+  through the mock and its test has to be skipped outright.
+- **`PaginatedResult` reads `Content-Type` unguarded**, so anything it paginates over
+  needs one. A native dict or list body gets one automatically.
+- **The mock enforces the client's read timeout**, so `respond_with_delay` beyond
+  `http_request_timeout` raises rather than arriving late.
+- **Several client options are missing entirely** — `max_message_size` and
+  `log_handler` among them — and raise `TypeError` rather than being ignored. Check
+  `ably/types/options.py` first.
+
+## Timers
+
+There is no clock seam. `ably/http/http.py` calls `time.time()` directly. Where a spec calls
+`enable_fake_timers()` / `ADVANCE_TIME(ms)`, prefer short real timeouts driven by client
+options (`fallback_retry_timeout=100`), which is what the specs themselves do. The global
+pytest timeout is 30 seconds, so keep waits well under it.
+
+## Deviations
+
+Diagnose per the decision tree in `writing-derived-tests.md`, then apply one of:
+
+- **Env-gated skip**, for non-compliance expected to be fixed:
+  ```python
+  from test.uts.helpers.deviations import deviation
+
+  @deviation
+  async def test_rsa7b_client_id_from_token_details():
+      ...
+  ```
+  Reproduce with `RUN_DEVIATIONS=1 uv run --extra crypto pytest -k rsa7b`.
+- **Adapted assertion**, preferred where the behaviour is stable: assert what the SDK does,
+  with the spec's expectation in a comment above.
+- **Spec-error fail-fast**, only where the spec contradicts the features spec:
+  `pytest.fail('UTS spec error <point> - fix the spec first; see deviations.md')`.
+
+Never write a test that passes under either behaviour.
+
+Record every one in `test/uts/deviations.md` under its heading, keeping all four headings
+present and in order: UTS Spec Errors, Failing Tests, Adapted Tests, Mock Infrastructure
+Limitations. Each entry needs the spec point, what the spec says, what the SDK does, root
+cause where known, which tests are affected, and status.
+
+A differently spelled API is not a deviation. Record only wrong behaviour.
+
+## Checks
+
+```bash
+uv run ruff check ably/ test/
+uv run --extra crypto pytest test/uts -q
+```
+
+Both must pass. Line length is 115.
diff --git a/test/uts/README.md b/test/uts/README.md
new file mode 100644
index 00000000..22acd263
--- /dev/null
+++ b/test/uts/README.md
@@ -0,0 +1,45 @@
+# Universal Test Specifications
+
+Tests here are derived from the pseudocode specifications in the
+[ably/specification](https://github.com/ably/specification) repository under `uts/`.
+They are mechanical translations: each one names the spec point it covers and
+carries a `# UTS: <id>` comment identifying the specification it came from.
+
+Read `uts/docs/writing-derived-tests.md` in the specification repository before
+adding or changing tests here, alongside `.claude/skills/uts-to-python/SKILL.md`,
+which covers what is particular to this SDK. Record anything that departs from a
+specification in [deviations.md](deviations.md), which also covers how the
+specifications are adopted here and why.
+
+## Layout
+
+```
+helpers/     shared infrastructure the specifications assume
+rest/        specifications under uts/rest
+realtime/    specifications under uts/realtime
+```
+
+Unit tests serve every request from a mock and reach no network. Integration
+tests run against a sandbox app.
+
+## Installing the mock
+
+The specifications express mock installation as a global `install_mock(mock_http)`.
+Here a mock is passed to the client it serves:
+
+```python
+mock_http = MockHttpClient(
+    on_connection_attempt=lambda conn: conn.respond_with_success(),
+    on_request=lambda req: req.respond_with(200, {'result': 'ok'}),
+)
+ably = AblyRest(key=key, test_options=TestOptions(http_transport=mock_http.as_transport()))
+```
+
+The client builds its HTTP client once, so construct the mock first. Teardown is
+`await ably.close()`, which stands in for `uninstall_mock()`.
+
+## Running
+
+```
+uv run --extra crypto pytest test/uts
+```
diff --git a/test/uts/__init__.py b/test/uts/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/test/uts/conftest.py b/test/uts/conftest.py
new file mode 100644
index 00000000..d0dc371b
--- /dev/null
+++ b/test/uts/conftest.py
@@ -0,0 +1,10 @@
+import pytest
+
+from test.uts.helpers.client import close_open_clients
+
+
+@pytest.fixture(autouse=True)
+async def close_clients():
+    """Closes the clients a test built, whether or not its assertions held."""
+    yield
+    await close_open_clients()
diff --git a/test/uts/deviations.md b/test/uts/deviations.md
new file mode 100644
index 00000000..32d1d878
--- /dev/null
+++ b/test/uts/deviations.md
@@ -0,0 +1,174 @@
+# Deviations
+
+Where the derived tests depart from the Universal Test Specifications, and why.
+The closing section covers how the specifications are adopted here; everything
+before it records behaviour.
+
+Entries are grouped by root cause rather than by test, so one entry covers every
+test it affects. Headings are fixed and appear even when they hold nothing.
+
+Entries closed by a fix are removed rather than kept as history; `git log` holds that.
+
+Run the gated tests with:
+
+```
+RUN_DEVIATIONS=1 uv run --extra crypto pytest test/uts
+```
+
+## UTS Spec Errors
+
+Faults in the specifications themselves, found while deriving. A fault here is not an
+SDK deviation, so it is recorded against the specification and not adapted to what the
+SDK happens to do.
+
+Where a specification *asserts* something `features.md` or `protocol.md` contradicts,
+there is no spec-correct assertion left to write. The test is still derived faithfully
+from the specification text, so that correcting the specification is all it takes to make
+it pass, and it is marked `@spec_error` — a skip gated on `RUN_DEVIATIONS`, the same gate
+`@deviation` uses, with a reason naming the specification rather than the SDK. The suite
+stays green, a real regression still shows, and the failure is one environment variable
+away. Each is filed upstream, in the issues named below.
+
+Where instead only a specification's *fixture*, *setup* or *label* is at fault, the
+assertion it carries still stands. Those tests keep the corrected fixture (or the
+corrected label in a comment), pass, and carry a `# UTS SPEC ERROR:` comment at the
+site. The entries below cover both kinds and say which applies.
+
+The three sections that follow this one record SDK behaviour rather than specification
+faults.
+
+Raised upstream:
+
+| Issue | Covers |
+|---|---|
+| [#523](https://github.com/ably/specification/issues/523) | `fallback.md` written against a mock API the contract does not define and the guide bans |
+| [#524](https://github.com/ably/specification/issues/524) | `/time` stubbed as an object rather than an array |
+| [#525](https://github.com/ably/specification/issues/525) | Spec points mislabelled across four specs |
+| [#526](https://github.com/ably/specification/issues/526) | Presence actions written as strings, and wire action 4 asserted to be LEAVE |
+| [#527](https://github.com/ably/specification/issues/527) | Wire encodings that contradict the features spec, and eight specs that read a body without pinning the protocol |
+| [#528](https://github.com/ably/specification/issues/528) | Fixtures that cannot hold their stated values |
+| [#529](https://github.com/ably/specification/issues/529) | Token expiry tests that demand optional behaviour |
+| [#530](https://github.com/ably/specification/issues/530) | Token renewal driven through `/time` |
+| [#531](https://github.com/ably/specification/issues/531) | `RSA10i` asserting that an API key survives `authorize()`, with no assertions |
+| [#532](https://github.com/ably/specification/issues/532) | Housekeeping: a leaked local path, sections carrying no Test ID, a duplicate, misfiled tests |
+
+Not every entry has an issue of its own: the URL-safe base64 alphabet is recorded below
+and not filed, because ably-python's own encoding settles the tests either way. Line
+references in these entries are against `ably/specification@d9a04ca`.
+
+
+## Failing Tests
+
+The specification's assertion is preserved and gated behind `@deviation`. Removing
+the mark is the only change needed once the SDK behaviour lands.
+
+## Adapted Tests
+
+The test asserts what the SDK does, with the specification's expectation in a
+comment above. These run, so they guard against regression.
+
+## Mock Infrastructure Limitations
+
+## How the specifications are adopted here
+
+Choices about the approach, as against the behaviour recorded above.
+
+### Tests are derived against the async API only
+
+`ably/sync` is generated from `ably/` by `ably/scripts/unasync.py`, a token
+rewriter driven by hand-maintained literal-string maps. Those maps name mock
+targets individually, so every patch target in a derived test would need an
+entry, and a missed entry produces a test that runs against the async class
+while appearing to cover the sync one.
+
+The specifications describe client-side behaviour — request formation, response
+parsing, state transitions — which is the same code in both variants. Running
+the suite twice would establish that the rewriter works, which is a different
+question and deserves its own much smaller test.
+
+Derived tests therefore live at `test/uts/`, outside both unasync source globs.
+
+### A mock is an httpx transport, supplied as a client option
+
+`mock_http.md` leaves injection open: "The mechanism for injecting the mock is
+implementation-specific and not part of the public API."
+
+The seam is `TestOptions(http_transport=...)`, consumed in `Http.__create_client`.
+Serving requests at the transport layer keeps URL construction, header merging,
+the host fallback loop and response decoding in the path, which is what the
+specifications assert on. It also carries the distinction between a failed
+connection and a failed request directly, as httpx raises `ConnectError`,
+`ConnectTimeout` and `ReadTimeout` separately.
+
+The alternative, replacing the whole client, would stub out the code under test.
+
+### A mock serves one client rather than being installed globally
+
+The specifications write `install_mock(mock_http)` and warn against passing a
+mock to a client, because the SDKs they were first written against hold HTTP
+behind a platform singleton. This client builds its HTTP client during
+construction and holds it, so a mock is passed as a client option instead.
+
+Derived tests construct the mock first, and `uninstall_mock()` has no
+counterpart — `test/uts/conftest.py` closes clients after each test, so a test
+whose assertions fail still releases its client.
+
+### A native response body is encoded to match the request
+
+`respond_with(200, {...})` leaves the encoding open. Encoding it as JSON
+unconditionally, as the reference implementation does, breaks against this SDK,
+where `use_binary_protocol` defaults to `True` and `Response.to_native()`
+dispatches on the response content type.
+
+A body given as a dict or list is therefore encoded to match the request's
+`Accept` header, and carries the matching `Content-Type`. A body given as
+`bytes` or `str` is passed through untouched, so a specification can still pin
+the encoding itself.
+
+This keeps the binary protocol on its default setting for the specifications
+that exercise it, rather than turning it off across the suite.
+
+### The superseded `queue_*` mock API is implemented
+
+`writing-test-specs.md` lists `mock_http.queue_response()` under "Common
+Mistakes to Avoid" in favour of the `onRequest` handler. The specifications have
+not followed: `fallback.md`, `request.md` and `rest_client.md` use it heavily.
+
+Implementing it costs little and keeps those three translations literal.
+Rewriting each call into a handler and a counter would be the kind of invention
+that derived tests exist to avoid.
+
+Stubs are matched by URL, then by host, then consumed first-in-first-out, and a
+URL is normalised before matching so that a default port may be given or
+omitted.
+
+### Test names carry the spec point, and the Test ID sits above them
+
+A specification's Test ID, `rest/unit/RSC16/returns-server-time-0`, becomes
+`test_rsc16_time_returns_server_time`, with the full ID in a `# UTS:` comment
+immediately above the function. The comment is the identifier that survives
+renaming; the function name makes failures readable without it.
+
+### Timing is driven by client options rather than a fake clock
+
+`enable_fake_timers()` and `ADVANCE_TIME(ms)` have no counterpart here, as there
+is no clock seam — `ably/http/http.py` calls `time.time()` directly. Where a
+specification advances time, the derived test shortens the interval through a
+client option instead, which is what the specifications themselves do for
+`fallback_retry_timeout`. Adding a clock seam is left until the realtime specs,
+which need one for reconnection timing.
+
+### A TokenDetails payload is recognised by its `token`, not only by `issued`
+
+RSA8c admits "a `TokenRequest` or `TokenDetails` object" from an `authUrl` without
+saying how to tell them apart. ably-python discriminated on `issued`, as ably-js
+still does, so the `{"token": ..., "expires": ...}` that seven specifications
+return was read as a `TokenRequest` and rejected as 40170.
+
+`token` is now accepted as a second discriminator. A `TokenRequest` never carries
+one — TE2 makes `keyName`, `nonce` and `mac` its required fields — so this only
+widens what is accepted, and the existing `issued` branch is untouched.
+
+It is a deliberate divergence from ably-js, whose derived suite avoids the
+question by returning a `text/plain` token string in place of the specification's
+JSON body.
diff --git a/test/uts/helpers/__init__.py b/test/uts/helpers/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/test/uts/helpers/client.py b/test/uts/helpers/client.py
new file mode 100644
index 00000000..a20014f8
--- /dev/null
+++ b/test/uts/helpers/client.py
@@ -0,0 +1,29 @@
+"""Construction and teardown for the clients that derived tests drive."""
+
+from ably import AblyRest
+from ably.types.testoptions import TestOptions
+
+DEFAULT_KEY = 'app.key:secret'
+
+CREDENTIAL_OPTIONS = ('key', 'token', 'token_details', 'auth_callback', 'auth_url', 'key_name')
+
+__open_clients = []
+
+
+def rest_client(mock_http, **kwargs):
+    """A REST client whose HTTP calls `mock_http` serves.
+
+    Stands in for the specifications' `install_mock(mock_http)` followed by
+    `Rest(options: ...)`. Credentials default to a key where a specification
+    does not name any. The client is closed when the test ends.
+    """
+    if not any(option in kwargs for option in CREDENTIAL_OPTIONS):
+        kwargs['key'] = DEFAULT_KEY
+    client = AblyRest(test_options=TestOptions(http_transport=mock_http.as_transport()), **kwargs)
+    __open_clients.append(client)
+    return client
+
+
+async def close_open_clients():
+    while __open_clients:
+        await __open_clients.pop().close()
diff --git a/test/uts/helpers/deviations.py b/test/uts/helpers/deviations.py
new file mode 100644
index 00000000..7d91a9f4
--- /dev/null
+++ b/test/uts/helpers/deviations.py
@@ -0,0 +1,20 @@
+"""Gating for tests that record where the SDK departs from a specification."""
+
+import os
+
+import pytest
+
+RUN_DEVIATIONS = 'RUN_DEVIATIONS'
+
+deviation = pytest.mark.skipif(
+    not os.environ.get(RUN_DEVIATIONS),
+    reason=f'Departs from the specification; see test/uts/deviations.md. Set {RUN_DEVIATIONS}=1 to run.',
+)
+
+spec_error = pytest.mark.skipif(
+    not os.environ.get(RUN_DEVIATIONS),
+    reason=(
+        'The specification is at fault, not the SDK; see test/uts/deviations.md. '
+        f'Set {RUN_DEVIATIONS}=1 to run.'
+    ),
+)
diff --git a/test/uts/helpers/mock_http.py b/test/uts/helpers/mock_http.py
new file mode 100644
index 00000000..3e7e5611
--- /dev/null
+++ b/test/uts/helpers/mock_http.py
@@ -0,0 +1,282 @@
+"""A stand-in for the HTTP layer, implementing the ``mock_http`` helper that the
+Universal Test Specifications are written against.
+
+The contract lives in ``uts/rest/unit/helpers/mock_http.md`` in the
+ably/specification repository. Names here match the pseudocode, which reserves
+snake_case for test-harness constructs.
+
+Every HTTP call surfaces as two events: a `PendingConnection`, and then, only
+once that connection succeeds, a `PendingRequest`. Connection failures raise
+transport exceptions and leave `captured_requests` untouched.
+"""
+
+import asyncio
+import json
+import time
+
+import httpx
+import msgpack
+
+DEFAULT_PORTS = {'https': 443, 'http': 80}
+
+MSGPACK_CONTENT_TYPE = 'application/x-msgpack'
+JSON_CONTENT_TYPE = 'application/json'
+
+
+class RecordedUrl:
+    """The parts of a request URL that the specifications assert on."""
+
+    def __init__(self, url):
+        self.__url = url
+        self.scheme = url.scheme
+        # Hostname only, so that assertions on url.host ignore the port
+        self.host = url.host
+        self.port = url.port or DEFAULT_PORTS.get(url.scheme, 80)
+        self.path = url.path
+        # The path as it goes on the wire, for assertions about encoding, which
+        # url.path cannot carry because it is decoded
+        self.raw_path = url.raw_path.decode('ascii').split('?')[0]
+        self.query_params = dict(url.params.items())
+
+    def __str__(self):
+        return str(self.__url)
+
+    def __repr__(self):
+        return f'RecordedUrl({str(self.__url)!r})'
+
+
+class PendingConnection:
+    """A connection attempt awaiting an outcome from the test."""
+
+    def __init__(self, host, port, tls):
+        self.host = host
+        self.port = port
+        self.tls = tls
+        self.timestamp = time.time()
+        self._outcome = asyncio.get_running_loop().create_future()
+
+    def __settle(self, failure):
+        if not self._outcome.done():
+            self._outcome.set_result(failure)
+
+    def respond_with_success(self):
+        self.__settle(None)
+
+    def respond_with_refused(self):
+        self.__settle((httpx.ConnectError, f'Connection refused to {self.host}:{self.port}'))
+
+    def respond_with_timeout(self):
+        self.__settle((httpx.ConnectTimeout, f'Connection to {self.host}:{self.port} timed out'))
+
+    def respond_with_dns_error(self):
+        self.__settle((httpx.ConnectError, f'Name resolution failed for {self.host}'))
+
+
+class PendingRequest:
+    """A request awaiting a response from the test.
+
+    Resolving it is independent of the handler that received it, so a handler
+    may hold on to a request and respond to it later in the test.
+    """
+
+    def __init__(self, request):
+        self._request = request
+        self.method = request.method.upper()
+        self.url = RecordedUrl(request.url)
+        self.path = self.url.path
+        # httpx.Headers looks up case-insensitively, as the specifications expect
+        self.headers = request.headers
+        self.body = request.content
+        self.timestamp = time.time()
+        self._outcome = asyncio.get_running_loop().create_future()
+
+    def __settle(self, delay, response, error=None):
+        if not self._outcome.done():
+            self._outcome.set_result((delay, response, error))
+
+    def respond_with(self, status, body=None, headers=None):
+        self.__settle(0, self.__build_response(status, body, headers))
+
+    def respond_with_delay(self, delay, status, body=None, headers=None):
+        """Respond after `delay` milliseconds have passed."""
+        self.__settle(delay / 1000.0, self.__build_response(status, body, headers))
+
+    def respond_with_timeout(self):
+        self.__settle(0, None, httpx.ReadTimeout(f'Request to {self.url} timed out',
+                                                 request=self._request))
+
+    def __build_response(self, status, body, headers):
+        headers = dict(headers or {})
+        content_type = next(
+            (value for name, value in headers.items() if name.lower() == 'content-type'), None)
+
+        if body is None:
+            content = b''
+        elif isinstance(body, (bytes, str)):
+            # Bodies the test encoded itself are passed through as given
+            content = body.encode('utf-8') if isinstance(body, str) else body
+        else:
+            if content_type is None:
+                # Serialise a native body the way the client asked to receive it
+                accept = self.headers.get('accept', '')
+                content_type = MSGPACK_CONTENT_TYPE if 'x-msgpack' in accept else JSON_CONTENT_TYPE
+                headers['Content-Type'] = content_type
+            if 'x-msgpack' in content_type:
+                content = msgpack.packb(body, use_bin_type=False)
+            else:
+                content = json.dumps(body, separators=(',', ':')).encode('utf-8')
+
+        return httpx.Response(status, headers=headers, content=content)
+
+
+class MockHttpClient:
+    """Serves the HTTP requests a client makes, in place of the network.
+
+    Requests reach the test one of three ways, in order of precedence: a
+    pending `await_request` future, the `on_request` handler, or a queued
+    response. A request none of those covers is answered with a 404.
+    """
+
+    def __init__(self, on_connection_attempt=None, on_request=None):
+        # Handlers are reassignable, as some specifications set them per-phase
+        self.on_connection_attempt = on_connection_attempt
+        self.on_request = on_request
+        self.captured_requests = []
+        self.__connection_waiters = []
+        self.__request_waiters = []
+        self.__queued = []
+        self.__queued_by_host = {}
+        self.__queued_by_url = {}
+
+    def as_transport(self):
+        """The transport to pass as `TestOptions(http_transport=...)`."""
+        return _MockTransport(self)
+
+    def reset(self):
+        self.captured_requests.clear()
+        self.__connection_waiters.clear()
+        self.__request_waiters.clear()
+        self.__queued.clear()
+        self.__queued_by_host.clear()
+        self.__queued_by_url.clear()
+
+    async def await_connection_attempt(self, timeout=None):
+        return await self.__await_event(
+            self.__connection_waiters, timeout, 'Timeout waiting for connection attempt')
+
+    async def await_request(self, timeout=None):
+        return await self.__await_event(
+            self.__request_waiters, timeout, 'Timeout waiting for request')
+
+    def queue_response(self, status, body=None, headers=None):
+        self.__queued.append(lambda request: request.respond_with(status, body, headers))
+
+    def queue_responses(self, count, status, body=None, headers=None):
+        for _ in range(count):
+            self.queue_response(status, body, headers)
+
+    def queue_timeout(self):
+        self.__queued.append(lambda request: request.respond_with_timeout())
+
+    def queue_delayed_response(self, delay, status, body=None, headers=None):
+        self.__queued.append(
+            lambda request: request.respond_with_delay(delay, status, body, headers))
+
+    def queue_response_for_host(self, host, status, body=None, headers=None):
+        self.__queued_by_host.setdefault(host, []).append(
+            lambda request: request.respond_with(status, body, headers))
+
+    def queue_response_for_url(self, url, status, body=None, headers=None):
+        # Matched against the normalised URL, so a default port may be given or omitted
+        self.__queued_by_url.setdefault(str(httpx.URL(str(url))), []).append(
+            lambda request: request.respond_with(status, body, headers))
+
+    async def _handle(self, request):
+        connection = PendingConnection(
+            host=request.url.host,
+            port=request.url.port or DEFAULT_PORTS.get(request.url.scheme, 80),
+            tls=request.url.scheme == 'https',
+        )
+        self.__dispatch_connection(connection)
+        failure = await connection._outcome
+        if failure is not None:
+            error_class, message = failure
+            raise error_class(message, request=request)
+
+        pending = PendingRequest(request)
+        self.captured_requests.append(pending)
+        self.__dispatch_request(pending)
+
+        delay, response, error = await pending._outcome
+        if delay:
+            # Timeouts are enforced by the transport, so a delay beyond the
+            # client's read budget has to surface as one
+            read_timeout = (request.extensions.get('timeout') or {}).get('read')
+            if read_timeout is not None and delay > read_timeout:
+                await asyncio.sleep(read_timeout)
+                raise httpx.ReadTimeout(f'Request to {pending.url} timed out', request=request)
+            await asyncio.sleep(delay)
+        if error is not None:
+            raise error
+        return response
+
+    def __dispatch_connection(self, connection):
+        waiter = self.__take_waiter(self.__connection_waiters)
+        if waiter is not None:
+            waiter.set_result(connection)
+        elif self.on_connection_attempt is not None:
+            self.on_connection_attempt(connection)
+        else:
+            connection.respond_with_success()
+
+    def __dispatch_request(self, request):
+        waiter = self.__take_waiter(self.__request_waiters)
+        if waiter is not None:
+            waiter.set_result(request)
+            return
+        if self.on_request is not None:
+            self.on_request(request)
+            return
+        stub = self.__take_queued_stub(request)
+        if stub is not None:
+            stub(request)
+            return
+        request.respond_with(404, {'error': {'message': 'No response configured', 'code': 40400}})
+
+    def __take_queued_stub(self, request):
+        for stubs, key in ((self.__queued_by_url, str(request.url)),
+                           (self.__queued_by_host, request.url.host)):
+            queued = stubs.get(key)
+            if queued:
+                return queued.pop(0)
+        if self.__queued:
+            return self.__queued.pop(0)
+        return None
+
+    @staticmethod
+    def __take_waiter(waiters):
+        while waiters:
+            waiter = waiters.pop(0)
+            if not waiter.done():
+                return waiter
+        return None
+
+    @staticmethod
+    async def __await_event(waiters, timeout, message):
+        waiter = asyncio.get_running_loop().create_future()
+        waiters.append(waiter)
+        try:
+            return await asyncio.wait_for(waiter, timeout)
+        except asyncio.TimeoutError:
+            raise AssertionError(message) from None
+
+
+class _MockTransport(httpx.AsyncBaseTransport):
+    def __init__(self, mock):
+        self.__mock = mock
+
+    async def handle_async_request(self, request):
+        return await self.__mock._handle(request)
+
+    async def aclose(self):
+        pass
diff --git a/test/uts/helpers/mock_http_test.py b/test/uts/helpers/mock_http_test.py
new file mode 100644
index 00000000..7d02097e
--- /dev/null
+++ b/test/uts/helpers/mock_http_test.py
@@ -0,0 +1,647 @@
+"""Tests for the `mock_http` helper, driven through the REST client it serves."""
+
+import asyncio
+import time
+
+import msgpack
+import pytest
+
+from ably import AblyRest
+from ably.types.testoptions import TestOptions
+from ably.util.exceptions import AblyException
+from test.uts.helpers.mock_http import MockHttpClient
+
+KEY = 'a.b:c'
+SERVER_TIME = 1234567890000
+
+MSGPACK_CONTENT_TYPE = 'application/x-msgpack'
+JSON_CONTENT_TYPE = 'application/json'
+
+
+def rest_client(mock, **kwargs):
+    """A REST client whose HTTP calls are served by `mock`."""
+    return AblyRest(key=KEY, test_options=TestOptions(http_transport=mock.as_transport()), **kwargs)
+
+
+async def raw_get(ably, path='/time'):
+    """The response to a GET as it came off the wire, before the client interprets it."""
+    return await ably.http.make_request('GET', path, skip_auth=True, raise_on_error=False)
+
+
+def fail_first_connection(failure):
+    """A connection handler which fails the first attempt with `failure` and accepts the rest."""
+    attempts = []
+
+    def connect(connection):
+        attempts.append(connection)
+        if len(attempts) == 1:
+            getattr(connection, failure)()
+        else:
+            connection.respond_with_success()
+
+    return connect, attempts
+
+
+async def wait_for(predicate, timeout=1.0):
+    deadline = time.time() + timeout
+    while not predicate():
+        assert time.time() < deadline, 'Timed out waiting for the client to reach the mock'
+        await asyncio.sleep(0.005)
+
+
+# Dispatch precedence
+
+async def test_a_queued_response_serves_a_request():
+    mock = MockHttpClient()
+    mock.queue_response(200, [SERVER_TIME])
+    ably = rest_client(mock)
+
+    assert await ably.time() == SERVER_TIME
+    await ably.close()
+
+
+async def test_the_on_request_handler_takes_precedence_over_a_queued_response():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, [1]))
+    mock.queue_response(200, [2])
+    ably = rest_client(mock)
+
+    assert await ably.time() == 1
+
+    # The queued stub was left untouched, so it serves the next request
+    mock.on_request = None
+    assert await ably.time() == 2
+    await ably.close()
+
+
+async def test_an_awaiting_test_takes_precedence_over_the_on_request_handler():
+    handled = []
+    mock = MockHttpClient(on_request=handled.append)
+    mock.queue_response(200, [2])
+    ably = rest_client(mock)
+
+    call = asyncio.ensure_future(ably.time())
+    request = await mock.await_request(timeout=0.2)
+    request.respond_with(200, [SERVER_TIME])
+
+    assert await call == SERVER_TIME
+    assert handled == []
+    await ably.close()
+
+
+async def test_an_unconfigured_request_is_answered_with_a_404():
+    mock = MockHttpClient()
+    ably = rest_client(mock)
+
+    response = await raw_get(ably)
+
+    assert response.status_code == 404
+    assert response.to_native() == {'error': {'message': 'No response configured', 'code': 40400}}
+    await ably.close()
+
+
+async def test_a_connection_succeeds_by_default():
+    mock = MockHttpClient()
+    mock.queue_response(200, [SERVER_TIME])
+    ably = rest_client(mock)
+
+    assert await ably.time() == SERVER_TIME
+    assert len(mock.captured_requests) == 1
+    await ably.close()
+
+
+async def test_the_on_connection_attempt_handler_receives_each_attempt():
+    attempts = []
+
+    def connect(connection):
+        attempts.append(connection)
+        connection.respond_with_success()
+
+    mock = MockHttpClient(on_connection_attempt=connect,
+                          on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+    host = ably.http.get_hosts()[0]
+
+    assert await ably.time() == SERVER_TIME
+    assert [connection.host for connection in attempts] == [host]
+    await ably.close()
+
+
+async def test_an_awaiting_test_takes_precedence_over_the_on_connection_attempt_handler():
+    handled = []
+    mock = MockHttpClient(on_connection_attempt=handled.append,
+                          on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+
+    call = asyncio.ensure_future(ably.time())
+    connection = await mock.await_connection_attempt(timeout=0.2)
+    connection.respond_with_success()
+
+    assert await call == SERVER_TIME
+    assert handled == []
+    await ably.close()
+
+
+# The two phases of a call
+
+async def test_a_pending_connection_describes_a_tls_target():
+    attempts = []
+
+    def connect(connection):
+        attempts.append(connection)
+        connection.respond_with_success()
+
+    mock = MockHttpClient(on_connection_attempt=connect,
+                          on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+    host = ably.http.get_hosts()[0]
+    before = time.time()
+
+    await ably.time()
+
+    connection = attempts[0]
+    assert connection.host == host
+    assert connection.port == 443
+    assert connection.tls is True
+    assert before <= connection.timestamp <= time.time()
+    await ably.close()
+
+
+async def test_a_pending_connection_describes_a_non_tls_target():
+    attempts = []
+
+    def connect(connection):
+        attempts.append(connection)
+        connection.respond_with_success()
+
+    mock = MockHttpClient(on_connection_attempt=connect,
+                          on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = AblyRest(token='foo', tls=False, test_options=TestOptions(http_transport=mock.as_transport()))
+
+    await ably.time()
+
+    assert attempts[0].port == 80
+    assert attempts[0].tls is False
+    assert mock.captured_requests[0].url.scheme == 'http'
+    assert mock.captured_requests[0].url.port == 80
+    await ably.close()
+
+
+@pytest.mark.parametrize('failure', ['respond_with_refused', 'respond_with_timeout', 'respond_with_dns_error'])
+async def test_a_failed_connection_moves_the_client_to_the_next_host(failure):
+    connect, attempts = fail_first_connection(failure)
+    mock = MockHttpClient(on_connection_attempt=connect,
+                          on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+    hosts = ably.http.get_hosts()
+
+    assert await ably.time() == SERVER_TIME
+
+    assert [connection.host for connection in attempts] == hosts[:2]
+    # A connection that failed never became a request
+    assert [request.url.host for request in mock.captured_requests] == [hosts[1]]
+    await ably.close()
+
+
+async def test_a_connection_that_never_succeeds_captures_no_requests():
+    attempts = []
+
+    def refuse(connection):
+        attempts.append(connection)
+        connection.respond_with_refused()
+
+    mock = MockHttpClient(on_connection_attempt=refuse)
+    ably = rest_client(mock)
+    hosts = ably.http.get_hosts()
+
+    with pytest.raises(AblyException):
+        await ably.time()
+
+    assert [connection.host for connection in attempts] == hosts
+    assert mock.captured_requests == []
+    await ably.close()
+
+
+# Response body serialisation
+
+async def test_a_native_body_is_msgpack_encoded_for_a_binary_protocol_client():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, {'served': 'msgpack'}))
+    ably = rest_client(mock)
+
+    response = await raw_get(ably)
+
+    assert response.headers['content-type'] == MSGPACK_CONTENT_TYPE
+    assert msgpack.unpackb(response.content) == {'served': 'msgpack'}
+    await ably.close()
+
+
+async def test_a_native_body_is_json_encoded_for_a_text_protocol_client():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, {'served': 'json'}))
+    ably = rest_client(mock, use_binary_protocol=False)
+
+    response = await raw_get(ably)
+
+    assert response.headers['content-type'] == JSON_CONTENT_TYPE
+    assert response.content == b'{"served":"json"}'
+    await ably.close()
+
+
+async def test_an_explicit_json_content_type_overrides_a_binary_protocol_client():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(
+        200, [SERVER_TIME], {'Content-Type': JSON_CONTENT_TYPE}))
+    ably = rest_client(mock)
+
+    response = await raw_get(ably)
+
+    assert response.headers['content-type'] == JSON_CONTENT_TYPE
+    assert response.content == b'[1234567890000]'
+    assert response.to_native() == [SERVER_TIME]
+    await ably.close()
+
+
+async def test_an_explicit_msgpack_content_type_overrides_a_text_protocol_client():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(
+        200, [SERVER_TIME], {'Content-Type': MSGPACK_CONTENT_TYPE}))
+    ably = rest_client(mock, use_binary_protocol=False)
+
+    response = await raw_get(ably)
+
+    assert response.headers['content-type'] == MSGPACK_CONTENT_TYPE
+    assert msgpack.unpackb(response.content) == [SERVER_TIME]
+    await ably.close()
+
+
+async def test_a_bytes_body_is_passed_through_untouched():
+    packed = msgpack.packb({'clientId': 'alice'}, use_bin_type=False)
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(
+        200, packed, {'Content-Type': MSGPACK_CONTENT_TYPE}))
+    ably = rest_client(mock)
+
+    response = await raw_get(ably)
+
+    assert response.content == packed
+    assert response.to_native() == {'clientId': 'alice'}
+    await ably.close()
+
+
+async def test_a_str_body_is_passed_through_as_utf8_without_a_content_type():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, 'grüße'))
+    ably = rest_client(mock)
+
+    response = await raw_get(ably)
+
+    assert response.content == 'grüße'.encode()
+    assert 'content-type' not in response.headers
+    await ably.close()
+
+
+async def test_a_none_body_yields_empty_content():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(204))
+    ably = rest_client(mock)
+
+    response = await raw_get(ably)
+
+    assert response.status_code == 204
+    assert response.content == b''
+    assert response.to_native() is None
+    await ably.close()
+
+
+async def test_an_error_body_surfaces_with_its_own_code_and_message():
+    # RSC8: the code and message come from the body, not from the status
+    error = msgpack.packb({'error': {'message': 'Token expired', 'code': 40140, 'statusCode': 401}},
+                          use_bin_type=False)
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(
+        401, error, {'Content-Type': MSGPACK_CONTENT_TYPE}))
+    ably = rest_client(mock)
+
+    with pytest.raises(AblyException) as exc_info:
+        await ably.time()
+
+    assert exc_info.value.code == 40140
+    assert exc_info.value.status_code == 401
+    assert exc_info.value.message == 'Token expired'
+    assert len(mock.captured_requests) == 1
+    await ably.close()
+
+
+async def test_a_serialised_body_is_decoded_by_the_client():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+
+    assert await ably.time() == SERVER_TIME
+    await ably.close()
+
+
+# The shape of a recorded request
+
+async def test_a_recorded_request_exposes_the_parts_of_its_url():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+    host = ably.http.get_hosts()[0]
+    before = time.time()
+
+    await ably.time()
+
+    request = mock.captured_requests[0]
+    assert request.method == 'GET'
+    assert request.url.scheme == 'https'
+    assert request.url.host == host
+    assert request.url.port == 443
+    assert request.url.path == '/time'
+    assert request.path == '/time'
+    assert request.url.query_params == {}
+    assert str(request.url) == f'https://{host}/time'
+    assert before <= request.timestamp <= time.time()
+    await ably.close()
+
+
+async def test_a_recorded_request_exposes_its_query_params():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, []))
+    ably = rest_client(mock)
+    host = ably.http.get_hosts()[0]
+
+    await ably.stats(limit=5, direction='forwards')
+
+    request = mock.captured_requests[0]
+    assert request.url.path == '/stats'
+    assert request.url.query_params == {'direction': 'forwards', 'limit': '5'}
+    assert str(request.url).startswith(f'https://{host}/stats?')
+    await ably.close()
+
+
+async def test_a_recorded_request_carries_the_headers_and_body_the_client_sent():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, {'ok': True}))
+    ably = rest_client(mock)
+
+    await ably.request('POST', '/echo', version='3', params={'foo': 'bar'}, body={'x': 1})
+
+    request = mock.captured_requests[0]
+    assert request.method == 'POST'
+    assert request.url.path == '/echo'
+    assert request.url.query_params == {'foo': 'bar'}
+    assert request.headers['Content-Type'] == MSGPACK_CONTENT_TYPE
+    assert request.headers['content-type'] == MSGPACK_CONTENT_TYPE
+    assert request.headers['Accept'] == MSGPACK_CONTENT_TYPE
+    assert request.headers['Authorization'].startswith('Basic ')
+    assert msgpack.unpackb(request.body) == {'x': 1}
+    await ably.close()
+
+
+# Queued responses
+
+async def test_queued_responses_are_consumed_in_order():
+    mock = MockHttpClient()
+    mock.queue_response(200, [1])
+    mock.queue_response(200, [2])
+    ably = rest_client(mock)
+
+    assert await ably.time() == 1
+    assert await ably.time() == 2
+    # The queue is empty again, so the fallback applies
+    assert (await raw_get(ably)).status_code == 404
+    await ably.close()
+
+
+async def test_queue_responses_queues_the_same_stub_several_times():
+    mock = MockHttpClient()
+    mock.queue_responses(count=3, status=200, body=[SERVER_TIME])
+    ably = rest_client(mock)
+
+    assert [await ably.time() for _ in range(3)] == [SERVER_TIME] * 3
+    assert (await raw_get(ably)).status_code == 404
+    await ably.close()
+
+
+async def test_a_queued_timeout_moves_the_client_to_the_next_host():
+    mock = MockHttpClient()
+    mock.queue_timeout()
+    mock.queue_response(200, [SERVER_TIME])
+    ably = rest_client(mock)
+    hosts = ably.http.get_hosts()
+
+    assert await ably.time() == SERVER_TIME
+    # A timed out request was still made, unlike a failed connection
+    assert [request.url.host for request in mock.captured_requests] == hosts[:2]
+    await ably.close()
+
+
+async def test_a_queued_delayed_response_arrives_after_the_given_milliseconds():
+    mock = MockHttpClient()
+    mock.queue_delayed_response(100, 200, [SERVER_TIME])
+    ably = rest_client(mock)
+
+    started = time.time()
+    assert await ably.time() == SERVER_TIME
+    elapsed = time.time() - started
+
+    assert 0.09 <= elapsed < 1
+    await ably.close()
+
+
+async def test_a_stub_queued_for_a_host_takes_precedence_over_the_queue():
+    mock = MockHttpClient()
+    ably = rest_client(mock)
+    mock.queue_response(200, [2])
+    mock.queue_response_for_host(ably.http.get_hosts()[0], 200, [1])
+
+    assert await ably.time() == 1
+    assert await ably.time() == 2
+    await ably.close()
+
+
+async def test_stubs_queued_for_one_host_are_consumed_in_order():
+    mock = MockHttpClient()
+    ably = rest_client(mock)
+    host = ably.http.get_hosts()[0]
+    mock.queue_response_for_host(host, 200, [1])
+    mock.queue_response_for_host(host, 200, [2])
+
+    assert await ably.time() == 1
+    assert await ably.time() == 2
+    await ably.close()
+
+
+async def test_a_stub_queued_for_a_url_takes_precedence_over_the_queue():
+    mock = MockHttpClient()
+    ably = rest_client(mock)
+    mock.queue_response(200, [2])
+    mock.queue_response_for_url(f'https://{ably.http.get_hosts()[0]}/time', 200, [1])
+
+    assert await ably.time() == 1
+    assert await ably.time() == 2
+    await ably.close()
+
+
+# Awaiting an event
+
+async def test_await_request_hands_the_pending_request_to_the_test():
+    mock = MockHttpClient()
+    ably = rest_client(mock)
+
+    call = asyncio.ensure_future(ably.time())
+    request = await mock.await_request(timeout=0.2)
+
+    assert request.url.path == '/time'
+    request.respond_with(200, [SERVER_TIME])
+    assert await call == SERVER_TIME
+    await ably.close()
+
+
+async def test_await_request_raises_when_no_request_arrives():
+    mock = MockHttpClient()
+
+    with pytest.raises(AssertionError, match='^Timeout waiting for request$'):
+        await mock.await_request(timeout=0.2)
+
+
+async def test_await_connection_attempt_hands_the_pending_connection_to_the_test():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+    host = ably.http.get_hosts()[0]
+
+    call = asyncio.ensure_future(ably.time())
+    connection = await mock.await_connection_attempt(timeout=0.2)
+
+    assert connection.host == host
+    # Nothing is recorded until the connection succeeds
+    assert mock.captured_requests == []
+
+    connection.respond_with_success()
+    assert await call == SERVER_TIME
+    assert len(mock.captured_requests) == 1
+    await ably.close()
+
+
+async def test_await_connection_attempt_raises_when_no_connection_is_attempted():
+    mock = MockHttpClient()
+
+    with pytest.raises(AssertionError, match='^Timeout waiting for connection attempt$'):
+        await mock.await_connection_attempt(timeout=0.2)
+
+
+async def test_a_handler_may_hold_a_request_and_respond_to_it_later():
+    held = []
+    mock = MockHttpClient(on_request=held.append)
+    ably = rest_client(mock)
+
+    call = asyncio.ensure_future(ably.time())
+    await wait_for(lambda: len(held) == 1)
+
+    # The client waits while the test gets on with something else
+    await asyncio.sleep(0.05)
+    assert not call.done()
+
+    held[0].respond_with(200, [SERVER_TIME])
+    assert await call == SERVER_TIME
+    await ably.close()
+
+
+# Resetting and reassigning
+
+async def test_reset_clears_captured_requests_and_queued_responses():
+    mock = MockHttpClient()
+    mock.queue_response(200, [SERVER_TIME])
+    ably = rest_client(mock)
+    await ably.time()
+    assert len(mock.captured_requests) == 1
+
+    mock.queue_response(200, [1])
+    mock.reset()
+
+    assert mock.captured_requests == []
+    assert (await raw_get(ably)).status_code == 404
+    await ably.close()
+
+
+async def test_reset_keeps_the_handlers_in_place():
+    attempts = []
+
+    def connect(connection):
+        attempts.append(connection)
+        connection.respond_with_success()
+
+    mock = MockHttpClient(on_connection_attempt=connect,
+                          on_request=lambda request: request.respond_with(200, [SERVER_TIME]))
+    ably = rest_client(mock)
+    await ably.time()
+
+    mock.reset()
+
+    assert await ably.time() == SERVER_TIME
+    assert len(attempts) == 2
+    assert len(mock.captured_requests) == 1
+    await ably.close()
+
+
+async def test_reset_drops_a_waiting_test():
+    mock = MockHttpClient()
+    ably = rest_client(mock)
+    waiting = asyncio.ensure_future(mock.await_request(timeout=0.2))
+    await asyncio.sleep(0)
+
+    mock.reset()
+
+    assert (await raw_get(ably)).status_code == 404
+    with pytest.raises(AssertionError, match='^Timeout waiting for request$'):
+        await waiting
+    await ably.close()
+
+
+async def test_the_handlers_can_be_assigned_after_construction():
+    mock = MockHttpClient()
+    ably = rest_client(mock)
+
+    mock.on_request = lambda request: request.respond_with(200, [SERVER_TIME])
+    assert await ably.time() == SERVER_TIME
+
+    mock.on_connection_attempt = lambda connection: connection.respond_with_refused()
+    with pytest.raises(AblyException):
+        await ably.time()
+    await ably.close()
+
+
+async def test_queue_response_for_url_matches_a_key_written_with_a_default_port():
+    mock = MockHttpClient()
+    mock.queue_response_for_url('https://main.realtime.ably.net:443/time', 200, [99])
+    ably = rest_client(mock)
+
+    assert await ably.time() == 99
+
+    await ably.close()
+
+
+async def test_a_delayed_response_beyond_the_read_budget_times_out():
+    mock = MockHttpClient()
+    mock.queue_delayed_response(5000, 200, [1])
+    ably = rest_client(mock, http_request_timeout=0.05)
+
+    started = time.monotonic()
+    with pytest.raises(AblyException):
+        await ably.time()
+    elapsed = time.monotonic() - started
+
+    # The client's budget decides when it gives up, not the queued delay
+    assert elapsed < 1
+    await ably.close()
+
+
+async def test_a_delayed_response_within_the_read_budget_arrives():
+    mock = MockHttpClient()
+    mock.queue_delayed_response(20, 200, [7])
+    ably = rest_client(mock, http_request_timeout=5)
+
+    assert await ably.time() == 7
+
+    await ably.close()
+
+
+async def test_a_recorded_url_keeps_the_encoded_path_alongside_the_decoded_one():
+    mock = MockHttpClient(on_request=lambda request: request.respond_with(200, []))
+    ably = rest_client(mock)
+
+    await ably.channels.get('a/b:c').history()
+
+    url = mock.captured_requests[0].url
+    assert url.raw_path == '/channels/a%2Fb:c/messages'
+    assert url.path == '/channels/a/b:c/messages'
+    await ably.close()
diff --git a/test/uts/rest/__init__.py b/test/uts/rest/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/test/uts/rest/unit/__init__.py b/test/uts/rest/unit/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/test/uts/rest/unit/time_test.py b/test/uts/rest/unit/time_test.py
new file mode 100644
index 00000000..ad2d4b6e
--- /dev/null
+++ b/test/uts/rest/unit/time_test.py
@@ -0,0 +1,121 @@
+"""Derived from uts/rest/unit/time.md in ably/specification.
+
+Spec points: RSC16
+"""
+
+import pytest
+
+from ably.util.exceptions import AblyException
+from test.uts.helpers.client import rest_client
+from test.uts.helpers.mock_http import MockHttpClient
+
+SERVER_TIME_MS = 1704067200000
+
+
+def capture_and_respond(captured_requests, body=None):
+    def on_request(request):
+        captured_requests.append(request)
+        request.respond_with(200, body if body is not None else [SERVER_TIME_MS])
+
+    return on_request
+
+
+# UTS: rest/unit/RSC16/returns-server-time-0
+async def test_rsc16_time_returns_server_time():
+    captured_requests = []
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    client = rest_client(mock_http)
+
+    result = await client.time()
+
+    # NOTE: the spec asserts the result IS DateTime. Its stated requirement allows
+    # "a DateTime or timestamp", and time() returns milliseconds since the epoch.
+    assert isinstance(result, (int, float))
+    assert result == SERVER_TIME_MS
+
+    assert len(captured_requests) == 1
+    request = captured_requests[0]
+    assert request.method == 'GET'
+    assert request.path == '/time'
+
+
+# UTS: rest/unit/RSC16/request-format-get-time-1
+async def test_rsc16_time_request_format():
+    captured_requests = []
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    client = rest_client(mock_http)
+
+    await client.time()
+
+    assert len(captured_requests) == 1
+    request = captured_requests[0]
+
+    assert request.method == 'GET'
+    assert request.path == '/time'
+
+    assert 'X-Ably-Version' in request.headers
+    assert 'Ably-Agent' in request.headers
+
+
+# UTS: rest/unit/RSC16/no-auth-required-2
+async def test_rsc16_time_does_not_require_authentication():
+    captured_requests = []
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    client = rest_client(mock_http)
+
+    result = await client.time()
+
+    assert isinstance(result, (int, float))
+
+    assert len(captured_requests) == 1
+    request = captured_requests[0]
+    assert 'Authorization' not in request.headers
+
+
+# UTS: rest/unit/RSC16/works-without-tls-3
+async def test_rsc16_time_works_without_tls():
+    captured_requests = []
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    client = rest_client(mock_http, tls=False, use_token_auth=True)
+
+    result = await client.time()
+
+    assert isinstance(result, (int, float))
+
+    assert len(captured_requests) == 1
+    request = captured_requests[0]
+    assert request.url.scheme == 'http'
+    assert 'Authorization' not in request.headers
+
+
+# UTS: rest/unit/RSC16/error-propagated-4
+async def test_rsc16_time_propagates_errors():
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=lambda request: request.respond_with(500, {
+            'error': {
+                'message': 'Internal server error',
+                'code': 50000,
+                'statusCode': 500,
+            },
+        }),
+    )
+    client = rest_client(mock_http)
+
+    with pytest.raises(AblyException) as excinfo:
+        await client.time()
+
+    assert excinfo.value.status_code == 500
+    assert excinfo.value.code == 50000