[ServiceBus] allow optionally testing pyamqp (#37010)

* optional test pyamqp

* fix failing flaky tests

Author: swathipil

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

@@ -1151,7 +1151,7 @@ def test_queue_by_servicebus_client_receive_batch_with_retrieve_deadletter(self,
     @pytest.mark.live_test_only
     @CachedServiceBusResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
-    @CachedServiceBusQueuePreparer(name_prefix='servicebustest', dead_lettering_on_message_expiration=True)
+    @ServiceBusQueuePreparer(name_prefix='servicebustest', dead_lettering_on_message_expiration=True)
     @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
     @ArgPasser()
     def test_queue_by_servicebus_client_session_fail(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs):
@@ -2651,7 +2651,7 @@ def _hack_amqp_sender_run(self, **kwargs):
     @pytest.mark.live_test_only
     @CachedServiceBusResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
-    @CachedServiceBusQueuePreparer(name_prefix='servicebustest', dead_lettering_on_message_expiration=True)
+    @ServiceBusQueuePreparer(name_prefix='servicebustest', dead_lettering_on_message_expiration=True)
     @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
     @ArgPasser()
     def test_queue_send_large_message_receive(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs):

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

@@ -17,7 +17,7 @@
 
 # temporary - disable uamqp if China b/c of 8+ hr runtime
 uamqp_available = uamqp_available and os.environ.get('SERVICEBUS_ENDPOINT_SUFFIX') != '.servicebus.chinacloudapi.cn'
-
+test_pyamqp = os.environ.get('TEST_PYAMQP', 'true') == 'true'
 
 def _get_default_handler():
     handler = logging.StreamHandler(stream=sys.stdout)
@@ -60,7 +60,7 @@ def sleep_until_expired(entity):
     time.sleep(max(0,(entity.locked_until_utc - utc_now()).total_seconds()+1))
 
 
-def uamqp_transport(use_uamqp=uamqp_available, use_pyamqp=True):
+def uamqp_transport(use_uamqp=uamqp_available, use_pyamqp=test_pyamqp):
     uamqp_transport_params = []
     uamqp_transport_ids = []
     if use_uamqp: