[ServiceBus] unskip complete message over diff recvr test for uamqp (#37192)

* unskip + fix failing

* remove whitespace

* fix failing tests

* bump asyncio sleep for complete test

* only time/asyncio sleep if uamqp

* missing await

* remove uamqp from dev reqs

Author: swathipil

sdk/servicebus/azure-servicebus/dev_requirements.txt

@@ -4,5 +4,4 @@
 azure-mgmt-servicebus~=8.0.0
 aiohttp>=3.0
 websocket-client
-azure-mgmt-resource<=16.0.0
-uamqp
+azure-mgmt-resource<=16.0.0

sdk/servicebus/azure-servicebus/tests/async_tests/test_queues_async.py

@@ -2965,29 +2965,30 @@ async def test_state_deferred_async(self, uamqp_transport, *, servicebus_namespa
     @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
     @ArgPasserAsync()
     async def test_queue_complete_message_on_different_receiver_async(self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs):
-        # Skipping if uamqp: This bug will not be fixed in uamqp.
-        if not uamqp_transport:
-            async with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string, uamqp_transport=uamqp_transport) as sb_client:
-                sender = sb_client.get_queue_sender(servicebus_queue.name)
-                receiver1 = sb_client.get_queue_receiver(servicebus_queue.name)
-                receiver2 = sb_client.get_queue_receiver(servicebus_queue.name)
-                
-                async with sender, receiver1, receiver2:
-                    await sender.send_messages([ServiceBusMessage('test') for _ in range(5)])
-                    received_msgs = []
-                    # the amount of messages returned by receive call is not stable, especially in live tests
-                    # of different os platforms, this is why a while loop is used here to receive the specific
-                    # amount of message we want to receive
-                    while len(received_msgs) < 5:
-                        # start receives on the first receiver and complete them on the other one.
-                        # the messages should settle over the management of the second receiver.
-                        for msg in await receiver1.receive_messages(max_message_count=10, max_wait_time=5):
-                            await receiver2.complete_message(msg)
-                            received_msgs.append(msg)
-                    
-                    assert len(received_msgs) == 5
-                    
-                    messages_in_queue = await receiver1.peek_messages()
-        
-                    assert len(messages_in_queue) == 0
+        async with ServiceBusClient.from_connection_string(
+            servicebus_namespace_connection_string, uamqp_transport=uamqp_transport) as sb_client:
+            sender = sb_client.get_queue_sender(servicebus_queue.name)
+            receiver1 = sb_client.get_queue_receiver(servicebus_queue.name)
+            receiver2 = sb_client.get_queue_receiver(servicebus_queue.name)
+
+            async with sender, receiver1, receiver2:
+                await sender.send_messages([ServiceBusMessage('test') for _ in range(5)])
+                received_msgs = []
+                # the amount of messages returned by receive call is not stable, especially in live tests
+                # of different os platforms, this is why a while loop is used here to receive the specific
+                # amount of message we want to receive
+                while len(received_msgs) < 5:
+                    # start receives on the first receiver and complete them on the other one.
+                    # the messages should settle over the management of the second receiver.
+                    for msg in await receiver1.receive_messages(max_message_count=10, max_wait_time=5):
+                        await receiver2.complete_message(msg)
+                        received_msgs.append(msg)
+
+                assert len(received_msgs) == 5
+
+                if uamqp_transport:
+                    # wait to make sure message completed, since uamqp is taking longer in the CI
+                    await asyncio.sleep(20)
+                messages_in_queue = await receiver1.peek_messages()
+
+                assert len(messages_in_queue) == 0

sdk/servicebus/azure-servicebus/tests/async_tests/test_sb_client_async.py

@@ -688,7 +688,12 @@ async def test_custom_endpoint_connection_verify_exception_async(self,
                 uamqp_transport=uamqp_transport
             )
             async with client:
-                with pytest.raises(ServiceBusConnectionError):
+                if not uamqp_transport:
+                    error = ServiceBusConnectionError
+                else:
+                    # if uamqp, catches an "Authorization timeout" error and raises ServiceBusError
+                    error = ServiceBusError
+                with pytest.raises(error):
                     async with client.get_queue_sender(servicebus_queue.name) as sender:
                         await sender.send_messages(ServiceBusMessage("foo"))
 

sdk/servicebus/azure-servicebus/tests/test_queues.py

@@ -3381,29 +3381,30 @@ def test_state_deferred(self, uamqp_transport, *, servicebus_namespace=None, ser
     @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
     @ArgPasser()
     def test_queue_complete_message_on_different_receiver(self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs):
-        # Skipping if uamqp: This bug will not be fixed in uamqp.
-        if not uamqp_transport:
-            with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string, uamqp_transport=uamqp_transport) as sb_client:
-                sender = sb_client.get_queue_sender(servicebus_queue.name)
-                receiver1 = sb_client.get_queue_receiver(servicebus_queue.name)
-                receiver2 = sb_client.get_queue_receiver(servicebus_queue.name)
-                
-                with sender, receiver1, receiver2:
-                    sender.send_messages([ServiceBusMessage('test') for _ in range(5)])
-                    received_msgs = []
-                    # the amount of messages returned by receive call is not stable, especially in live tests
-                    # of different os platforms, this is why a while loop is used here to receive the specific
-                    # amount of message we want to receive
-                    while len(received_msgs) < 5:
-                        # start receives on the first receiver and complete them on the other one.
-                        # the messages should settle over the management of the second receiver.
-                        for msg in receiver1.receive_messages(max_message_count=10, max_wait_time=5):
-                            receiver2.complete_message(msg)
-                            received_msgs.append(msg)
-                    
-                    assert len(received_msgs) == 5
-                    
-                    messages_in_queue = receiver1.peek_messages()
-        
-                    assert len(messages_in_queue) == 0
+        with ServiceBusClient.from_connection_string(
+            servicebus_namespace_connection_string, uamqp_transport=uamqp_transport) as sb_client:
+            sender = sb_client.get_queue_sender(servicebus_queue.name)
+            receiver1 = sb_client.get_queue_receiver(servicebus_queue.name)
+            receiver2 = sb_client.get_queue_receiver(servicebus_queue.name)
+
+            with sender, receiver1, receiver2:
+                sender.send_messages([ServiceBusMessage('test') for _ in range(5)])
+                received_msgs = []
+                # the amount of messages returned by receive call is not stable, especially in live tests
+                # of different os platforms, this is why a while loop is used here to receive the specific
+                # amount of message we want to receive
+                while len(received_msgs) < 5:
+                    # start receives on the first receiver and complete them on the other one.
+                    # the messages should settle over the management of the second receiver.
+                    for msg in receiver1.receive_messages(max_message_count=10, max_wait_time=5):
+                        receiver2.complete_message(msg)
+                        received_msgs.append(msg)
+
+                assert len(received_msgs) == 5
+
+                if uamqp_transport:
+                    # wait to make sure message completed, since uamqp is taking longer in the CI
+                    time.sleep(10)
+                messages_in_queue = receiver1.peek_messages()
+
+                assert len(messages_in_queue) == 0

sdk/servicebus/azure-servicebus/tests/test_sb_client.py

@@ -716,7 +716,12 @@ def test_custom_endpoint_connection_verify_exception(self,
                 uamqp_transport=uamqp_transport
             )
             with client:
-                with pytest.raises(ServiceBusConnectionError):
+                if not uamqp_transport:
+                    error = ServiceBusConnectionError
+                else:
+                    # if uamqp, catches an "Authorization timeout" error and raises ServiceBusError
+                    error = ServiceBusError
+                with pytest.raises(error):
                     with client.get_queue_sender(servicebus_queue.name) as sender:
                         sender.send_messages(ServiceBusMessage("foo"))