[ServiceBus] fix transient timeout errors (Azure#23136)

* try fixing test send queues twice

* fix typo

* increase timeout, ignore service errors

* ignore log error for now

* remove print

* mark expected fail

* more ignore cannot log

* update time for send twice queues

* send twice complete messages

* await complete message

Author: swathipil

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

@@ -1624,18 +1624,20 @@ async def test_async_queue_send_twice(self, servicebus_namespace_connection_stri
                 await sender.send_messages(batch_message)
                 await sender.send_messages(batch_message)
                 messages = []
-                async with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver:
+                async with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=20) as receiver:
                     async for message in receiver:
                         messages.append(message)
-                assert len(messages) == 4
+                        await receiver.complete_message(message)
+                    assert len(messages) == 4
                 # then normal message resending
                 await sender.send_messages(message)
                 await sender.send_messages(message)
                 messages = []
-                async with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver:
+                async with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=20) as receiver:
                     async for message in receiver:
                         messages.append(message)
-                assert len(messages) == 2
+                        await receiver.complete_message(message)
+                    assert len(messages) == 2
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only

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

@@ -178,14 +178,15 @@ async def test_async_session_by_session_client_conn_str_receive_handler_with_sto
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
+    @pytest.mark.xfail(reason="'Cannot open log' error, potential service bug", raises=ServiceBusError)
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
     @ServiceBusQueuePreparer(name_prefix='servicebustest', requires_session=True)
     async def test_async_session_by_session_client_conn_str_receive_handler_with_no_session(self, servicebus_namespace_connection_string, servicebus_queue, **kwargs):
         async with ServiceBusClient.from_connection_string(
             servicebus_namespace_connection_string, logging_enable=False) as sb_client:
 
-            receiver = sb_client.get_queue_receiver(servicebus_queue.name, session_id=NEXT_AVAILABLE_SESSION, max_wait_time=5)
+            receiver = sb_client.get_queue_receiver(servicebus_queue.name, session_id=NEXT_AVAILABLE_SESSION, max_wait_time=10)
             with pytest.raises(OperationTimeoutError):
                 await receiver._open_with_retry()
 
@@ -815,8 +816,8 @@ async def test_async_session_cancel_scheduled_messages(self, servicebus_namespac
             messages = []
             async with sb_client.get_queue_receiver(servicebus_queue.name, session_id=session_id) as receiver:
                 renewer.register(receiver, receiver.session, max_lock_renewal_duration=140)
-                messages.extend(await receiver.receive_messages(max_wait_time=120))
-                messages.extend(await receiver.receive_messages(max_wait_time=5))
+                messages.extend(await receiver.receive_messages(max_wait_time=115))
+                messages.extend(await receiver.receive_messages(max_wait_time=10))
                 try:
                     assert len(messages) == 0
                 except AssertionError:
@@ -934,6 +935,7 @@ async def test_async_session_by_servicebus_client_list_sessions_with_client(self
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
+    @pytest.mark.xfail(reason="'Cannot open log' error, potential service bug", raises=ServiceBusError)
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
     @ServiceBusQueuePreparer(name_prefix='servicebustest', requires_session=True)
@@ -999,6 +1001,7 @@ async def test_async_session_basic_topic_subscription_send_and_receive(self, ser
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
+    @pytest.mark.xfail(reason="'Cannot open log' error, potential service bug", raises=ServiceBusError)
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
     @ServiceBusQueuePreparer(name_prefix='servicebustest', requires_session=True)

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

@@ -1018,7 +1018,7 @@ def test_queue_by_queue_client_conn_str_receive_handler_with_autolockrenew(self,
                                                  prefetch_count=10) as receiver:
                 received_msgs = receiver.receive_messages(max_message_count=10, max_wait_time=5)
                 for msg in received_msgs:
-                    renewer.register(receiver, msg, max_lock_renewal_duration=10)
+                    renewer.register(receiver, msg, max_lock_renewal_duration=30)
                 time.sleep(10)
 
                 for msg in received_msgs:
@@ -1039,7 +1039,7 @@ def test_queue_by_queue_client_conn_str_receive_handler_with_autolockrenew(self,
                                                  prefetch_count=3) as receiver:
                 received_msgs = receiver.receive_messages(max_message_count=3, max_wait_time=5)
                 for msg in received_msgs:
-                    renewer.register(receiver, msg, max_lock_renewal_duration=10)
+                    renewer.register(receiver, msg, max_lock_renewal_duration=30)
                 time.sleep(10)
 
                 for msg in received_msgs:
@@ -1061,7 +1061,7 @@ def test_queue_by_queue_client_conn_str_receive_handler_with_autolockrenew(self,
                                                  prefetch_count=3) as receiver:
                 received_msgs = receiver.receive_messages(max_message_count=3, max_wait_time=5)
                 for msg in received_msgs:
-                    renewer.register(receiver, msg, max_lock_renewal_duration=10)
+                    renewer.register(receiver, msg, max_lock_renewal_duration=30)
                 time.sleep(10)
 
                 for msg in received_msgs:
@@ -1993,18 +1993,20 @@ def test_queue_send_twice(self, servicebus_namespace_connection_string, serviceb
                 sender.send_messages(batch_message)
                 sender.send_messages(batch_message)
                 messages = []
-                with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver:
+                with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=20) as receiver:
                     for message in receiver:
                         messages.append(message)
-                assert len(messages) == 4
+                        receiver.complete_message(message)
+                    assert len(messages) == 4
                 # then normal message resending
                 sender.send_messages(message)
                 sender.send_messages(message)
                 messages = []
-                with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver:
+                with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=20) as receiver:
                     for message in receiver:
                         messages.append(message)
-                assert len(messages) == 2
+                        receiver.complete_message(message)
+                    assert len(messages) == 2
 
 
     @pytest.mark.liveTest

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

@@ -249,6 +249,7 @@ def test_session_by_session_client_conn_str_receive_handler_with_stop(self, serv
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
+    @pytest.mark.xfail(reason="'Cannot open log' error, potential service bug", raises=ServiceBusError)
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
     @ServiceBusQueuePreparer(name_prefix='servicebustest', requires_session=True)
@@ -258,11 +259,12 @@ def test_session_by_session_client_conn_str_receive_handler_with_no_session(self
             with pytest.raises(OperationTimeoutError):
                 with sb_client.get_queue_receiver(servicebus_queue.name, 
                                                   session_id=NEXT_AVAILABLE_SESSION, 
-                                                  max_wait_time=5,) as session:
+                                                  max_wait_time=10,) as session:
                     pass
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
+    @pytest.mark.xfail(reason="'Cannot open log' error, potential service bug", raises=ServiceBusError)
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
     @ServiceBusQueuePreparer(name_prefix='servicebustest', requires_session=True)
@@ -998,7 +1000,7 @@ def test_session_cancel_scheduled_messages(self, servicebus_namespace_connection
                 messages = []
                 count = 0
                 while not messages and count < 13:
-                    messages = receiver.receive_messages(max_wait_time=10)
+                    messages = receiver.receive_messages(max_wait_time=20)
                     receiver.session.renew_lock()
                     count += 1
                 assert len(messages) == 0
@@ -1032,7 +1034,7 @@ def test_session_get_set_state_with_receiver(self, servicebus_namespace_connecti
             assert count == 3
 
 
-    @pytest.mark.skip(reasion="Needs list sessions")
+    @pytest.mark.skip(reason="Needs list sessions")
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
@@ -1095,6 +1097,7 @@ def test_session_by_servicebus_client_list_sessions_with_client(self, servicebus
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
+    @pytest.mark.xfail(reason="'Cannot open log' error, potential service bug", raises=ServiceBusError)
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
     @ServiceBusQueuePreparer(name_prefix='servicebustest', requires_session=True)
@@ -1106,7 +1109,7 @@ def test_session_by_servicebus_client_session_pool(self, servicebus_namespace_co
         def message_processing(sb_client):
             while True:
                 try:
-                    with sb_client.get_queue_receiver(servicebus_queue.name, session_id=NEXT_AVAILABLE_SESSION, max_wait_time=5) as receiver:
+                    with sb_client.get_queue_receiver(servicebus_queue.name, session_id=NEXT_AVAILABLE_SESSION, max_wait_time=10) as receiver:
                         for message in receiver:
                             print("ServiceBusReceivedMessage: {}".format(message))
                             messages.append(message)