Merge pull request #717 from ably/uts/realtime-integration

test: derive the realtime integration UTS tier

Author: owenpearson

.claude/skills/uts-to-python/SKILL.md

@@ -1,2 +1,22 @@
 - after making any code changes, run `uv ruff check` to make sure linting passes
 - use `uv` to run any other necessary tasks such as `pytest`
+
+# Time
+
+Production code in `ably/` reads the current time and schedules delayed callbacks through
+the clock, never through `time`, `datetime.now` or `asyncio.sleep`:
+
+- `select_clock(options)` in the constructor, from `ably.util.clock`
+- `now_ms()` for the time of day
+- `monotonic_ms()` for a duration
+- `timer(timeout_ms, callback)` for a delayed callback
+
+A test replaces it with `TestOptions(clock=...)`. Converting a caller-supplied value, and
+arithmetic on the unix epoch, are not clock readings and stay as they are.
+
+`asyncio.wait_for` in `ConnectionManager.ping` is the one delay that runs on the event
+loop's clock rather than this one; a test which waits it out sets a short real
+`realtime_request_timeout`.
+
+`Auth._timestamp` signs a token request the server validates, so a client which reads a
+fake clock talks to a mock rather than to the sandbox.

CLAUDE.md

@@ -19,6 +19,7 @@
 from ably.types.mixins import DecodingContext
 from ably.types.operations import MessageOperation, PublishResult, UpdateDeleteResult
 from ably.types.presence import PresenceMessage
+from ably.util.clock import select_clock
 from ably.util.eventemitter import EventEmitter
 from ably.util.exceptions import AblyException, IncompatibleClientIdException
 from ably.util.helper import Timer, is_callable_or_coroutine, validate_message_size
@@ -58,6 +59,7 @@ def __init__(self, realtime: AblyRealtime, name: str, channel_options: ChannelOp
         EventEmitter.__init__(self)
         self.__name = name
         self.__realtime = realtime
+        self.__clock = select_clock(realtime.options)
         self.__state = ChannelState.INITIALIZED
         self.__message_emitter = EventEmitter()
         self.__state_timer: Timer | None = None
@@ -846,7 +848,7 @@ def on_timeout() -> None:
                 self.__state_timer = None
                 self.__timeout_pending_state()
 
-            self.__state_timer = Timer(self.__realtime.options.realtime_request_timeout, on_timeout)
+            self.__state_timer = self.__clock.timer(self.__realtime.options.realtime_request_timeout, on_timeout)
 
     def __clear_state_timer(self) -> None:
         if self.__state_timer:
@@ -866,7 +868,8 @@ def __start_retry_timer(self) -> None:
         if self.__retry_timer:
             return
 
-        self.__retry_timer = Timer(self.ably.options.channel_retry_timeout, self.__on_retry_timer_expire)
+        self.__retry_timer = self.__clock.timer(
+            self.ably.options.channel_retry_timeout, self.__on_retry_timer_expire)
 
     def __cancel_retry_timer(self) -> None:
         if self.__retry_timer:

ably/realtime/channel.py

@@ -2,7 +2,6 @@
 
 import asyncio
 import logging
-import time
 from collections import deque
 from itertools import zip_longest
 from typing import TYPE_CHECKING
@@ -16,6 +15,7 @@
 from ably.types.connectionstate import ConnectionEvent, ConnectionState, ConnectionStateChange
 from ably.types.operations import PublishResult
 from ably.types.tokendetails import TokenDetails
+from ably.util.clock import Clock, select_clock
 from ably.util.eventemitter import EventEmitter
 from ably.util.exceptions import AblyException, IncompatibleClientIdException
 from ably.util.helper import Timer, get_random_id, is_token_error
@@ -123,23 +123,25 @@ def clear(self) -> None:
 class PendingPing:
     """Represents a ping awaiting its heartbeat echo from the server"""
 
-    def __init__(self, id: str):
+    def __init__(self, id: str, clock: Clock):
         self.id = id
+        self.clock = clock
         self.future: asyncio.Future[None] = asyncio.get_running_loop().create_future()
-        self.start_time: float = time.monotonic()
+        self.start_time: float = clock.monotonic_ms()
 
     @property
     def message(self) -> dict:
         return {"action": ProtocolMessageAction.HEARTBEAT, "id": self.id}
 
     @property
     def response_time_ms(self) -> float:
-        return round((time.monotonic() - self.start_time) * 1000, 2)
+        return round(self.clock.monotonic_ms() - self.start_time, 2)
 
 
 class ConnectionManager(EventEmitter):
     def __init__(self, realtime: AblyRealtime, initial_state):
         self.options = realtime.options
+        self.clock = select_clock(self.options)
         self.__ably = realtime
         self.__state: ConnectionState = initial_state
         self.__pending_pings: dict[str, PendingPing] = {}
@@ -364,7 +366,7 @@ async def ping(self) -> float:
             raise AblyException("Cannot send ping request. Calling ping in invalid state", 400, 40000)
 
         # RTN13e: the id tells this ping's echo apart from server heartbeats and other pings
-        pending_ping = PendingPing(get_random_id())
+        pending_ping = PendingPing(get_random_id(), self.clock)
         self.__pending_pings[pending_ping.id] = pending_ping
         try:
             # RTN13d: while connecting, the heartbeat goes out once the connection is established
@@ -379,7 +381,7 @@ async def ping(self) -> float:
         return pending_ping.response_time_ms
 
     async def __send_heartbeat(self, pending_ping: PendingPing) -> None:
-        pending_ping.start_time = time.monotonic()
+        pending_ping.start_time = self.clock.monotonic_ms()
         try:
             await self.send_protocol_message(pending_ping.message)
         except Exception as error:
@@ -718,7 +720,7 @@ def on_transition_timer_expire():
 
         log.debug(f'ConnectionManager.start_transition_timer(): setting timer for {timeout}ms')
 
-        self.transition_timer = Timer(timeout, on_transition_timer_expire)
+        self.transition_timer = self.clock.timer(timeout, on_transition_timer_expire)
 
     def cancel_transition_timer(self):
         log.debug('ConnectionManager.cancel_transition_timer()')
@@ -741,7 +743,7 @@ def on_suspend_timer_expire() -> None:
                 )
                 self.__fail_state = ConnectionState.SUSPENDED
 
-        self.suspend_timer = Timer(Defaults.connection_state_ttl, on_suspend_timer_expire)
+        self.suspend_timer = self.clock.timer(Defaults.connection_state_ttl, on_suspend_timer_expire)
 
     def check_suspend_timer(self, state: ConnectionState) -> None:
         if state not in (
@@ -764,7 +766,7 @@ def on_retry_timeout():
             self.retry_timer = None
             self.request_state(ConnectionState.CONNECTING)
 
-        self.retry_timer = Timer(interval, on_retry_timeout)
+        self.retry_timer = self.clock.timer(interval, on_retry_timeout)
 
     def cancel_retry_timer(self) -> None:
         if self.retry_timer:

ably/realtime/connectionmanager.py

@@ -14,9 +14,9 @@
 from ably.transport.defaults import Defaults
 from ably.types.connectiondetails import ConnectionDetails
 from ably.types.operations import PublishResult
+from ably.util.clock import select_clock
 from ably.util.eventemitter import EventEmitter
 from ably.util.exceptions import AblyException
-from ably.util.helper import Timer, unix_time_ms
 
 try:
     # websockets 15+ preferred imports
@@ -68,6 +68,8 @@ def __init__(self, connection_manager: ConnectionManager, host: str, params: dic
         self.ws_connect_task: asyncio.Task | None = None
         self.connection_manager = connection_manager
         self.options = self.connection_manager.options
+        self.connect_func = self.__select_connect_func(self.options)
+        self.clock = select_clock(self.options)
         self.is_connected = False
         self.idle_timer = None
         self.last_activity = None
@@ -78,6 +80,13 @@ def __init__(self, connection_manager: ConnectionManager, host: str, params: dic
         self.format = params.get('format', 'json')
         super().__init__()
 
+    @staticmethod
+    def __select_connect_func(options):
+        test_options = getattr(options, '_test_options', None)
+        if test_options is not None and test_options.websocket_connect is not None:
+            return test_options.websocket_connect
+        return ws_connect
+
     def connect(self):
         headers = HttpUtils.default_headers()
         query_params = urllib.parse.urlencode(self.params)
@@ -103,11 +112,11 @@ async def ws_connect(self, ws_url, headers):
         try:
             # Use additional_headers for websockets 15+, fallback to extra_headers for older versions
             try:
-                async with ws_connect(ws_url, additional_headers=headers) as websocket:
+                async with self.connect_func(ws_url, additional_headers=headers) as websocket:
                     await self._handle_websocket_connection(ws_url, websocket)
             except TypeError:
                 # Fallback for websockets 14 and earlier
-                async with ws_connect(ws_url, extra_headers=headers) as websocket:
+                async with self.connect_func(ws_url, extra_headers=headers) as websocket:
                     await self._handle_websocket_connection(ws_url, websocket)
         except (WebSocketException, socket.gaierror) as e:
             exception = AblyException(f'Error opening websocket connection: {e}', 400, 40000)
@@ -294,11 +303,11 @@ async def send(self, message: dict):
     def set_idle_timer(self, timeout: float):
         if self.idle_timer:
             self.idle_timer.cancel()
-        self.idle_timer = Timer(timeout, self.on_idle_timer_expire)
+        self.idle_timer = self.clock.timer(timeout, self.on_idle_timer_expire)
 
     async def on_idle_timer_expire(self):
         self.idle_timer = None
-        since_last = unix_time_ms() - self.last_activity
+        since_last = self.clock.now_ms() - self.last_activity
         time_remaining = self.max_idle_interval - since_last
         msg = f"No activity seen from realtime in {since_last} ms; assuming connection has dropped"
         if time_remaining <= 0:
@@ -310,7 +319,7 @@ async def on_idle_timer_expire(self):
     def on_activity(self):
         if not self.max_idle_interval:
             return
-        self.last_activity = unix_time_ms()
+        self.last_activity = self.clock.now_ms()
         self.set_idle_timer(self.max_idle_interval + 100)
 
     async def disconnect(self, reason=None):

ably/transport/websockettransport.py

@@ -6,6 +6,12 @@ class TestOptions:
     :Parameters:
       - `http_transport`: an `httpx.AsyncBaseTransport` which handles every
         HTTP request the client makes, in place of the network.
+      - `websocket_connect`: a callable which opens every realtime websocket
+        connection the client makes, in place of `websockets.connect`. It is
+        called as `websocket_connect(url, additional_headers=headers)`, or with
+        `extra_headers=headers` if that raises `TypeError`, and returns an async
+        context manager yielding an object supporting `__aiter__`, `send` and
+        `close`.
       - `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
@@ -17,6 +23,7 @@ class TestOptions:
     # any module importing it as declaring a test suite.
     __test__ = False
 
-    def __init__(self, http_transport=None, clock=None):
+    def __init__(self, http_transport=None, websocket_connect=None, clock=None):
         self.http_transport = http_transport
+        self.websocket_connect = websocket_connect
         self.clock = clock

ably/types/testoptions.py

@@ -7,9 +7,9 @@
 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.
+    Token expiry, the server-time offset, the fallback-host cache and the
+    transport's idle detection all read the clock, and every delayed callback
+    is 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.

ably/util/clock.py

@@ -75,6 +75,7 @@ async def _job(self):
     def cancel(self):
         self._task.cancel()
 
+
 def validate_message_size(encoded_messages: list, use_binary_protocol: bool, max_message_size: int) -> None:
     """Validate that encoded messages don't exceed the maximum size limit.