test: remove the races that make presence and push tests flaky

test/ably/realtime/realtimeannotations_test.py

@@ -57,18 +57,20 @@ async def on_annotation(annotation):
 
         await channel.annotations.subscribe(on_annotation)
 
-        # Publish a message
-        publish_result = await channel.publish('message', 'foobar')
-
-        # Reset for next message (summary)
+        # The summary for a message is delivered as a separate message carrying the same
+        # serial, once the server has aggregated the annotations made against it, so the
+        # message's own create echo reaches this listener first
         message_summary = asyncio.Future()
 
         def on_message(msg):
-            if not message_summary.done():
+            if msg.action == MessageAction.MESSAGE_SUMMARY and not message_summary.done():
                 message_summary.set_result(msg)
 
         await channel.subscribe('message', on_message)
 
+        # Publish a message
+        publish_result = await channel.publish('message', 'foobar')
+
         # Publish annotation using realtime
         await channel.annotations.publish(publish_result.serials[0], Annotation(
             type='reaction:distinct.v1',

test/ably/realtime/realtimeauth_test.py

@@ -228,6 +228,17 @@ async def callback(params):
 
         ably = await TestApp.get_ably_realtime(auth_callback=callback)
         original_transport = await ably.connection.connection_manager.once_async('transport.pending')
+
+        # Protocol messages from the first transport are withheld, so the connection is still
+        # CONNECTING for as long as it takes to obtain the new token
+        async def withhold_protocol_message(msg):
+            pass
+
+        original_transport.on_protocol_message = withhold_protocol_message
+        assert ably.connection.state == ConnectionState.CONNECTING
+
+        # RTC8b: a reauth issued while CONNECTING halts the in-flight connection attempt and
+        # starts a new one with the new token
         await ably.auth.authorize()
         assert ably.connection.state == ConnectionState.CONNECTED
         assert ably.connection.connection_manager.transport is not original_transport

test/ably/realtime/realtimepresence_test.py

@@ -17,6 +17,32 @@
 from test.ably.utils import BaseAsyncTestCase
 
 
+async def await_presence_sync(channel):
+    """
+    Block until the channel's presence set is in sync with the server.
+
+    An ATTACHED reply may carry the HAS_PRESENCE flag, in which case the server
+    follows it with a presence SYNC. Until that SYNC lands, a member entering
+    the channel can be delivered inside the SYNC with action PRESENT instead of
+    as a live ENTER, and the duplicate live ENTER that follows is then correctly
+    discarded by the RTP2b2 newness check (it carries the same message id), so
+    no 'enter' event is emitted at all. Waiting here for the SYNC to complete
+    means anything that happens afterwards is observed as a live event.
+    """
+    await channel.presence.get()
+
+
+async def wait_until(predicate, timeout=10.0, interval=0.05):
+    """Wait until predicate() is truthy, or raise AssertionError on timeout."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        if predicate():
+            return
+        await asyncio.sleep(interval)
+    raise AssertionError(f'condition not met within {timeout}s')
+
+
 async def force_suspended(client):
     client.connection.connection_manager.request_state(ConnectionState.DISCONNECTED)
 
@@ -70,6 +96,7 @@ def on_presence(msg):
                 presence_received.set_result(msg)
 
         await channel1.presence.subscribe(on_presence)
+        await await_presence_sync(channel1)
 
         # Client 2 enters without attaching first
         channel2 = self.client2.channels.get(channel_name)
@@ -119,6 +146,7 @@ def on_presence(msg):
             events.append((msg.action, msg.client_id))
 
         await channel1.presence.subscribe(on_presence)
+        await await_presence_sync(channel1)
 
         # Client 2 enters
         await channel2.presence.enter('enter data')
@@ -346,6 +374,7 @@ def on_presence(msg):
                 received.set_result(msg)
 
         await channel1.presence.subscribe(on_presence)
+        await await_presence_sync(channel1)
         await channel1.presence.enter()
 
         msg = await asyncio.wait_for(received, timeout=5.0)
@@ -469,6 +498,7 @@ def on_presence(msg):
                 received.set_result(msg)
 
         await listener_channel.presence.subscribe(on_presence)
+        await await_presence_sync(listener_channel)
 
         # Create client and enter before it's connected
         enterer_client = await TestApp.get_ably_realtime(
@@ -509,6 +539,7 @@ def on_presence(msg):
                 second_enter_received.set_result(msg)
 
         await listener_channel.presence.subscribe(on_presence)
+        await await_presence_sync(listener_channel)
 
         # Create enterer client
         enterer_client = await TestApp.get_ably_realtime(
@@ -660,9 +691,11 @@ def on_presence(msg):
             })
 
         await observer_channel.presence.subscribe(on_presence)
+        await await_presence_sync(observer_channel)
 
         # Create main client with remainPresentFor to control LEAVE timing
-        # This tells the server to send LEAVE for presence members 5 seconds after disconnect
+        # remainPresentFor tells the server how long to keep this connection's presence
+        # members after it disconnects before emitting a LEAVE for them
         client = await TestApp.get_ably_realtime(
             client_id='test_client',
             transport_params={'remainPresentFor': 1000},
@@ -695,14 +728,24 @@ def on_presence(msg):
         # Connection IDs should be different after suspend
         assert first_conn_id != second_conn_id
 
-        # Wait for presence events including LEAVE (which arrives after remainPresentFor timeout)
-        await asyncio.sleep(2)
+        def leaves():
+            return [e for e in events if e['action'] == PresenceAction.LEAVE
+                    and e['client_id'] == 'test_client']
+
+        def enters():
+            return [e for e in events if e['action'] == PresenceAction.ENTER
+                    and e['client_id'] == 'test_client']
+
+        # The LEAVE for the old connection is emitted by the server once
+        # remainPresentFor elapses, so its timing is not under the test's
+        # control.
+        await wait_until(
+            lambda: any(e['connection_id'] == first_conn_id for e in leaves())
+            and any(e['connection_id'] == second_conn_id for e in enters())
+        )
 
-        # Should see LEAVE for old connection and ENTER for new connection
-        leave_events = [e for e in events if e['action'] == PresenceAction.LEAVE
-                       and e['client_id'] == 'test_client']
-        enter_events = [e for e in events if e['action'] == PresenceAction.ENTER
-                       and e['client_id'] == 'test_client']
+        leave_events = leaves()
+        enter_events = enters()
 
         assert len(leave_events) >= 1, "Should have LEAVE event for old connection"
         assert len(enter_events) >= 2, "Should have ENTER event for new connection"

test/ably/rest/restchannelhistory_test.py

@@ -15,7 +15,10 @@ class TestRestChannelHistory(BaseAsyncTestCase, metaclass=VaryByProtocolTestsMet
 
     @pytest.fixture(autouse=True)
     async def setup(self):
-        self.ably = await TestApp.get_ably_rest(fallback_hosts=[])
+        # Publishing tens of messages one at a time leaves these tests unusually exposed to a
+        # connection failure part-way through, so they rely on the RSC15 retry onto one of the
+        # endpoint's fallback hosts to absorb it.
+        self.ably = await TestApp.get_ably_rest()
         self.test_vars = await TestApp.get_test_vars()
         yield
         await self.ably.close()

test/ably/rest/restpush_test.py

@@ -19,6 +19,12 @@
 DEVICE_TOKEN = '740f4707bebcf74f9b7c25d48e3358945f6aa01da5ddb387462c7eaf61bb78ad'
 
 
+# Each test in this class registers a set of devices and subscriptions in setup
+# and deletes them again in teardown, a dozen or more sequential requests per
+# phase against a live endpoint. A device deletion alone measures around 600ms,
+# so the default 30s per-phase timeout leaves too little headroom on a loaded
+# runner for teardown to finish.
+@pytest.mark.timeout(120)
 class TestPush(BaseAsyncTestCase, metaclass=VaryByProtocolTestsMetaclass):
 
     @pytest.fixture(autouse=True)
@@ -37,10 +43,12 @@ async def setup(self):
             await self.save_subscription(channel, device_id=device.id)
         assert len(list(itertools.chain(*self.channels.values()))) == len(self.devices)
         yield
-        for key, channel in zip(self.devices, itertools.cycle(self.channels)):
-            device = self.devices[key]
-            await self.remove_subscription(channel, device_id=device.id)
-            await self.ably.push.admin.device_registrations.remove(device_id=device.id)
+        # Removing a device registration also removes that device's channel
+        # subscriptions, so deleting every device cleans up both. This covers
+        # devices the test body registered as well, which matters because other
+        # tests assert on the total number of registered devices.
+        for device_id in list(self.devices):
+            await self.ably.push.admin.device_registrations.remove(device_id=device_id)
         await self.ably.close()
 
     def per_protocol_setup(self, use_binary_protocol):
@@ -121,11 +129,6 @@ async def save_subscription(self, channel, **kw):
         self.channels.setdefault(channel, []).append(subscription)
         return subscription
 
-    async def remove_subscription(self, channel, **kw):
-        subscription = PushChannelSubscription(channel, **kw)
-        subscription = await self.ably.push.admin.channel_subscriptions.remove(subscription)
-        return subscription
-
     # RSH1a
     async def test_admin_publish(self):
         recipient = {'clientId': 'ablyChannel'}

test/ably/rest/restrequest_test.py

@@ -17,15 +17,16 @@ async def setup(self):
         self.ably = await TestApp.get_ably_rest()
         self.test_vars = await TestApp.get_test_vars()
 
-        # Populate the channel (using the new api)
         self.channel = self.get_channel_name()
         self.path = f'/channels/{self.channel}/messages'
-        for i in range(20):
-            body = {'name': f'event{i}', 'data': f'lorem ipsum {i}'}
-            await self.ably.request('POST', self.path, body=body, version=Defaults.protocol_version)
         yield
         await self.ably.close()
 
+    async def publish_messages(self, count):
+        for i in range(count):
+            body = {'name': f'event{i}', 'data': f'lorem ipsum {i}'}
+            await self.ably.request('POST', self.path, body=body, version=Defaults.protocol_version)
+
     def per_protocol_setup(self, use_binary_protocol):
         self.ably.options.use_binary_protocol = use_binary_protocol
         self.use_binary_protocol = use_binary_protocol
@@ -42,6 +43,9 @@ async def test_post(self):
         assert 'messageId' in result.items[0]
 
     async def test_get(self):
+        # Paging is exercised below, so the channel needs more messages than one page holds
+        await self.publish_messages(20)
+
         params = {'limit': 10, 'direction': 'forwards'}
         result = await self.ably.request('GET', self.path, params=params, version=Defaults.protocol_version)
 

test/ably/testapp.py

@@ -71,6 +71,11 @@ async def get_ably_rest(**kw):
     async def get_ably_realtime(**kw):
         test_vars = await TestApp.get_test_vars()
         options = TestApp.get_options(test_vars, **kw)
+        # A connect attempt that fails leaves the connection DISCONNECTED until the retry
+        # timer expires. The default interval is longer than the timeout most tests allow
+        # for reaching CONNECTED, so a short one keeps a single failed attempt from using
+        # up the whole budget. Tests that assert on retry timing set their own value.
+        options.setdefault('disconnected_retry_timeout', 1000)
         return AblyRealtime(**options)
 
     @staticmethod

test/ably/utils.py

@@ -85,8 +85,12 @@ def test_decorator(fn):
         @functools.wraps(fn)
         async def test_decorated(self, *args, **kwargs):
             patcher = patch()
-            await fn(self, *args, **kwargs)
-            unpatch(patcher)
+            try:
+                await fn(self, *args, **kwargs)
+            finally:
+                # The patch is undone however the test body exits, so Http.make_request is
+                # left as it was found even when an assertion or a transport error escapes.
+                unpatch(patcher)
 
             assert len(responses) >= 1, \
                 "If your test doesn't make any requests, use the @dont_vary_protocol decorator"