fix: auth_url handling and add timeout to once_async calls in realtime tests

- Replaced all `connection.once_async` calls with `asyncio.wait_for` to include a 5-second timeout.
- Ensures tests fail gracefully if connection isn't established within the specified timeframe.

Author: ttypic

ably/rest/auth.py

@@ -1,13 +1,17 @@
 from __future__ import annotations
+
 import base64
-from datetime import timedelta
 import logging
 import time
-from typing import Optional, TYPE_CHECKING, Union
 import uuid
+from datetime import timedelta
+from typing import Optional, TYPE_CHECKING, Union
+from urllib.parse import urlparse, parse_qs
+
 import httpx
 
 from ably.types.options import Options
+
 if TYPE_CHECKING:
     from ably.rest.rest import AblyRest
     from ably.realtime.realtime import AblyRealtime
@@ -23,7 +27,6 @@
 
 
 class Auth:
-
     class Method:
         BASIC = "BASIC"
         TOKEN = "TOKEN"
@@ -271,8 +274,7 @@ async def create_token_request(self, token_params: Optional[dict | str] = None,
         if capability is not None:
             token_request['capability'] = str(Capability(capability))
 
-        token_request["client_id"] = (
-            token_params.get('client_id') or self.client_id)
+        token_request["client_id"] = token_params.get('client_id') or self.client_id
 
         # Note: There is no expectation that the client
         # specifies the nonce; this is done by the library
@@ -388,17 +390,39 @@ def _random_nonce(self):
 
     async def token_request_from_auth_url(self, method: str, url: str, token_params,
                                           headers, auth_params):
+        # Parse URL to extract existing query parameters
+        parsed_url = urlparse(url)
+        url_params = {}
+        if parsed_url.query:
+            # Convert query parameters to a flat dictionary
+            query_params = parse_qs(parsed_url.query)
+            for key, values in query_params.items():
+                # Take the last value if multiple values exist for the same key
+                url_params[key] = values[-1]
+
+        # Reconstruct clean URL without query parameters
+        clean_url = f"{parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}"
+        if parsed_url.fragment:
+            clean_url += f"#{parsed_url.fragment}"
+
         body = None
         params = None
         if method == 'GET':
             body = {}
-            params = dict(auth_params, **token_params)
+            # Merge URL params, auth_params, and token_params (later params override earlier ones)
+            # we do this because httpx version has inconsistency and some versions override query params
+            # that are specified in url string
+            params = dict(url_params, **auth_params, **token_params)
         elif method == 'POST':
             if isinstance(auth_params, TokenDetails):
                 auth_params = auth_params.to_dict()
-            params = {}
+            # For POST, URL params go in query string, auth_params and token_params go in body
+            params = url_params
             body = dict(auth_params, **token_params)
 
+        # Use clean URL for the request
+        url = clean_url
+
         from ably.http.http import Response
         async with httpx.AsyncClient(http2=True) as client:
             resp = await client.request(method=method, url=url, headers=headers, params=params, data=body)
@@ -420,6 +444,6 @@ async def token_request_from_auth_url(self, method: str, url: str, token_params,
             token_request = response.text
         else:
             msg = 'auth_url responded with unacceptable content-type ' + content_type + \
-                ', should be either text/plain, application/jwt or application/json',
+                  ', should be either text/plain, application/jwt or application/json',
             raise AblyAuthException(msg, 401, 40170)
         return token_request

test/ably/realtime/realtimeauth_test.py

@@ -34,7 +34,7 @@ async def auth_callback_failure(options, expect_failure=False):
 class TestRealtimeAuth(BaseAsyncTestCase):
     async def test_auth_valid_api_key(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         assert ably.connection.error_reason is None
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
@@ -53,7 +53,7 @@ async def test_auth_with_token_string(self):
         rest = await TestApp.get_ably_rest()
         token_details = await rest.auth.request_token()
         ably = await TestApp.get_ably_realtime(token=token_details.token)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -71,7 +71,7 @@ async def test_auth_with_token_details(self):
         rest = await TestApp.get_ably_rest()
         token_details = await rest.auth.request_token()
         ably = await TestApp.get_ably_realtime(token_details=token_details)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -93,7 +93,7 @@ async def callback(params):
             return token_details
 
         ably = await TestApp.get_ably_realtime(auth_callback=callback)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -107,7 +107,7 @@ async def callback(params):
             return token_details
 
         ably = await TestApp.get_ably_realtime(auth_callback=callback)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -121,7 +121,7 @@ async def callback(params):
             return token_details.token
 
         ably = await TestApp.get_ably_realtime(auth_callback=callback)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -144,7 +144,10 @@ async def test_auth_with_auth_url_json(self):
         url_path = f"{echo_url}/?type=json&body={urllib.parse.quote_plus(token_details_json)}"
 
         ably = await TestApp.get_ably_realtime(auth_url=url_path)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(
+            ably.connection.once_async(ConnectionState.CONNECTED),
+            timeout=5,
+        )
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -156,7 +159,7 @@ async def test_auth_with_auth_url_text_plain(self):
         url_path = f"{echo_url}/?type=text&body={token_details.token}"
 
         ably = await TestApp.get_ably_realtime(auth_url=url_path)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -169,7 +172,7 @@ async def test_auth_with_auth_url_post(self):
 
         ably = await TestApp.get_ably_realtime(auth_url=url_path, auth_method='POST',
                                                auth_params=token_details)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         response_time_ms = await ably.connection.ping()
         assert response_time_ms is not None
         assert ably.connection.error_reason is None
@@ -183,7 +186,7 @@ async def callback(params):
             return token_details.token
 
         ably = await TestApp.get_ably_realtime(auth_callback=callback)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
 
         assert ably.connection.connection_manager.transport
         original_access_token = ably.connection.connection_manager.transport.params.get('accessToken')
@@ -307,7 +310,7 @@ async def callback(params):
             "action": ProtocolMessageAction.AUTH,
         }
 
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         auth_future = asyncio.Future()
 
         def on_update(state_change):
@@ -334,7 +337,7 @@ async def auth_callback(_):
 
         ably = await TestApp.get_ably_realtime(auth_callback=auth_callback)
 
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         original_token_details = ably.auth.token_details
         await ably.connection.once_async(ConnectionEvent.UPDATE)
         assert ably.auth.token_details is not original_token_details
@@ -496,7 +499,7 @@ async def callback(params):
             }
         }
 
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         original_token_details = ably.auth.token_details
         assert ably.connection.connection_manager.transport
         await ably.connection.connection_manager.transport.on_protocol_message(msg)
@@ -511,7 +514,7 @@ async def test_renew_token_no_renew_means_provided_upon_disconnection(self):
 
         ably = await TestApp.get_ably_realtime(token_details=token_details)
 
-        state_change = await ably.connection.once_async(ConnectionState.CONNECTED)
+        state_change = await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         msg = {
             "action": ProtocolMessageAction.DISCONNECTED,
             "error": {
@@ -544,7 +547,7 @@ async def callback(params):
             }
         }
 
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         connection_key = ably.connection.connection_details.connection_key
         await ably.connection.connection_manager.transport.dispose()
         ably.connection.connection_manager.notify_state(ConnectionState.DISCONNECTED)
@@ -572,12 +575,12 @@ async def test_renew_token_no_renew_means_provided_on_resume(self):
             }
         }
 
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         connection_key = ably.connection.connection_details.connection_key
         await ably.connection.connection_manager.transport.dispose()
         ably.connection.connection_manager.notify_state(ConnectionState.DISCONNECTED)
 
-        state_change = await ably.connection.once_async(ConnectionState.CONNECTED)
+        state_change = await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         assert ably.connection.connection_manager.transport.params["resume"] == connection_key
 
         assert ably.connection.connection_manager.transport

test/ably/realtime/realtimechannel_test.py

@@ -33,7 +33,7 @@ async def test_channels_release(self):
 
     async def test_channel_attach(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         assert channel.state == ChannelState.INITIALIZED
         await channel.attach()
@@ -42,7 +42,7 @@ async def test_channel_attach(self):
 
     async def test_channel_detach(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         await channel.attach()
         await channel.detach()
@@ -62,7 +62,7 @@ def listener(message):
             else:
                 second_message_future.set_result(message)
 
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         await channel.attach()
         await channel.subscribe('event', listener)
@@ -91,7 +91,7 @@ def listener(msg: Message):
             if not message_future.done():
                 message_future.set_result(msg)
 
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         await channel.attach()
         await channel.subscribe('event', listener)
@@ -110,7 +110,7 @@ def listener(msg: Message):
 
     async def test_subscribe_coroutine(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         await channel.attach()
 
@@ -138,7 +138,7 @@ async def listener(msg):
     # RTL7a
     async def test_subscribe_all_events(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         await channel.attach()
 
@@ -165,7 +165,7 @@ def listener(msg):
     # RTL7c
     async def test_subscribe_auto_attach(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         assert channel.state == ChannelState.INITIALIZED
 
@@ -181,7 +181,7 @@ def listener(_):
     # RTL8b
     async def test_unsubscribe(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         await channel.attach()
 
@@ -216,7 +216,7 @@ def listener(msg):
     # RTL8c
     async def test_unsubscribe_all(self):
         ably = await TestApp.get_ably_realtime()
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get('my_channel')
         await channel.attach()
 
@@ -250,7 +250,7 @@ def listener(msg):
 
     async def test_realtime_request_timeout_attach(self):
         ably = await TestApp.get_ably_realtime(realtime_request_timeout=2000)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         original_send_protocol_message = ably.connection.connection_manager.send_protocol_message
 
         async def new_send_protocol_message(msg):
@@ -268,7 +268,7 @@ async def new_send_protocol_message(msg):
 
     async def test_realtime_request_timeout_detach(self):
         ably = await TestApp.get_ably_realtime(realtime_request_timeout=2000)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         original_send_protocol_message = ably.connection.connection_manager.send_protocol_message
 
         async def new_send_protocol_message(msg):
@@ -287,7 +287,7 @@ async def new_send_protocol_message(msg):
 
     async def test_channel_detached_once_connection_closed(self):
         ably = await TestApp.get_ably_realtime(realtime_request_timeout=2000)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get(random_string(5))
         await channel.attach()
 
@@ -296,7 +296,7 @@ async def test_channel_detached_once_connection_closed(self):
 
     async def test_channel_failed_once_connection_failed(self):
         ably = await TestApp.get_ably_realtime(realtime_request_timeout=2000)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get(random_string(5))
         await channel.attach()
 
@@ -307,7 +307,7 @@ async def test_channel_failed_once_connection_failed(self):
 
     async def test_channel_suspended_once_connection_suspended(self):
         ably = await TestApp.get_ably_realtime(realtime_request_timeout=2000)
-        await ably.connection.once_async(ConnectionState.CONNECTED)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
         channel = ably.channels.get(random_string(5))
         await channel.attach()