diff --git a/test/ably/realtime/realtimeannotations_test.py b/test/ably/realtime/realtimeannotations_test.py index a82b6b2b..2c5e0658 100644 --- a/test/ably/realtime/realtimeannotations_test.py +++ b/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', diff --git a/test/ably/realtime/realtimeauth_test.py b/test/ably/realtime/realtimeauth_test.py index 6ec53356..a2620489 100644 --- a/test/ably/realtime/realtimeauth_test.py +++ b/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 diff --git a/test/ably/realtime/realtimepresence_test.py b/test/ably/realtime/realtimepresence_test.py index 86a073c7..06067f14 100644 --- a/test/ably/realtime/realtimepresence_test.py +++ b/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" diff --git a/test/ably/rest/restchannelhistory_test.py b/test/ably/rest/restchannelhistory_test.py index a9a2245b..2b15ae0d 100644 --- a/test/ably/rest/restchannelhistory_test.py +++ b/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() diff --git a/test/ably/rest/restpush_test.py b/test/ably/rest/restpush_test.py index 867e8b90..30c4798d 100644 --- a/test/ably/rest/restpush_test.py +++ b/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'} diff --git a/test/ably/rest/restrequest_test.py b/test/ably/rest/restrequest_test.py index 967da19e..484c1bd6 100644 --- a/test/ably/rest/restrequest_test.py +++ b/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) diff --git a/test/ably/testapp.py b/test/ably/testapp.py index f657fdd4..744fd2c7 100644 --- a/test/ably/testapp.py +++ b/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 diff --git a/test/ably/utils.py b/test/ably/utils.py index ae19e0b5..a72fdd5c 100644 --- a/test/ably/utils.py +++ b/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"