[SB] settled as false (Azure#30607)

* deliveryid

* test

* assert message strung together correctly

* add settlement fix

* changing fix location

* fix tests

* update to just receive to clear out queue

* complete messages

Author: l0lawrence

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_receiver_async.py

@@ -124,4 +124,5 @@ async def send_disposition(
         if self._is_closed:
             raise ValueError("Link already closed.")
         await self._outgoing_disposition(first_delivery_id, last_delivery_id, settled, delivery_state, batchable)
-        await self._wait_for_response(wait)
+        if not settled:
+            await self._wait_for_response(wait)

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/receiver.py

@@ -121,4 +121,5 @@ def send_disposition(
         if self._is_closed:
             raise ValueError("Link already closed.")
         self._outgoing_disposition(first_delivery_id, last_delivery_id, settled, delivery_state, batchable)
-        self._wait_for_response(wait)
+        if not settled:
+            self._wait_for_response(wait)

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

@@ -981,6 +981,11 @@ async def test_async_queue_by_servicebus_client_session_fail(self, uamqp_transpo
 
             async with sb_client.get_queue_sender(servicebus_queue.name) as sender:
                 await sender.send_messages(ServiceBusMessage("test session sender", session_id="test"))
+
+            async with sb_client.get_queue_receiver(servicebus_queue.name) as receiver:
+                messages = await receiver.receive_messages()
+                for message in messages:
+                    await receiver.complete_message(message)
 
     @pytest.mark.asyncio
     @pytest.mark.liveTest
@@ -2083,57 +2088,27 @@ async def _hack_amqp_sender_run_async(self, **kwargs):
     @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
     @ArgPasserAsync()
     async def test_async_queue_send_large_message_receive(self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs):
-        async def _hack_amqp_sender_run_async(self, **kwargs):
-            time.sleep(6)  # sleep until timeout
-            if uamqp_transport:
-                await self.message_handler.work_async()
-                self._waiting_messages = 0
-                self._pending_messages = self._filter_pending()
-                if self._backoff and not self._waiting_messages:
-                    _logger.info("Client told to backoff - sleeping for %r seconds", self._backoff)
-                    await self._connection.sleep_async(self._backoff)
-                    self._backoff = 0
-                await self._connection.work_async()
-            else:
-                try:
-                    await self._link.update_pending_deliveries()
-                    await self._connection.listen(wait=self._socket_timeout, **kwargs)
-                except ValueError:
-                    self._shutdown = True
-                    return False
-            return True
-
         async with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string, logging_enable=False, uamqp_transport=uamqp_transport) as sb_client:
-            async with sb_client.get_queue_sender(servicebus_queue.name) as sender:
-                # this one doesn't need to reset the method, as it's hacking the method on the instance
-                sender._handler._client_run_async = types.MethodType(_hack_amqp_sender_run_async, sender._handler)
-                with pytest.raises(OperationTimeoutError):
-                    await sender.send_messages(ServiceBusMessage("body"), timeout=5)
-
-        if not uamqp_transport:
-            # Amqp
-            async with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string,
-                uamqp_transport=uamqp_transport
-            ) as sb_client:
-                async with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=1.0) as sender:
-                    payload = "A" * 250 * 1024
-                    await sender.send_messages(ServiceBusMessage(payload))
+            servicebus_namespace_connection_string,
+            uamqp_transport=uamqp_transport
+        ) as sb_client:
+            async with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=1.0) as sender:
+                payload = "A" * 250 * 1024
+                await sender.send_messages(ServiceBusMessage(payload))
 
-            if uamqp:
-                transport_type = uamqp.constants.TransportType.AmqpOverWebsocket
-            else:
-                transport_type = TransportType.AmqpOverWebsocket
-            # AmqpOverWebsocket
-            async with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string,
-                transport_type=transport_type,
-                uamqp_transport=uamqp_transport
-            ) as sb_client:
-                async with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=1.2) as sender:
-                    payload = "A" * 250 * 1024
-                    await sender.send_messages(ServiceBusMessage(payload))
+        if uamqp:
+            transport_type = uamqp.constants.TransportType.AmqpOverWebsocket
+        else:
+            transport_type = TransportType.AmqpOverWebsocket
+        # AmqpOverWebsocket
+        async with ServiceBusClient.from_connection_string(
+            servicebus_namespace_connection_string,
+            transport_type=transport_type,
+            uamqp_transport=uamqp_transport
+        ) as sb_client:
+            async with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=1.2) as sender:
+                payload = "A" * 250 * 1024
+                await sender.send_messages(ServiceBusMessage(payload))
 
         # ReceiveMessages
         async with ServiceBusClient.from_connection_string(
@@ -2144,7 +2119,7 @@ async def _hack_amqp_sender_run_async(self, **kwargs):
                     if not uamqp_transport:
                         assert message._delivery_id is not None
                         assert message._message.data[0].decode("utf-8")  == "A" * 250 * 1024
-                    receiver.complete_message(message)  # complete messages 
+                    await receiver.complete_message(message)  # complete messages 
 
     @pytest.mark.asyncio
     @pytest.mark.liveTest
@@ -2723,7 +2698,7 @@ async def test_queue_async_send_amqp_annotated_message(self, uamqp_transport, *,
                             assert raw_amqp_message.annotations[b'ann_key'] == b'ann_value'
                             assert raw_amqp_message.application_properties[b'body_type'] == b'value'
                             recv_value_msg += 1
-                        receiver.complete_message(message)
+                        await receiver.complete_message(message)
 
                     assert recv_data_msg == 3
                     assert recv_sequence_msg == 3

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

@@ -1070,6 +1070,11 @@ def test_queue_by_servicebus_client_session_fail(self, uamqp_transport, *, servi
 
             with sb_client.get_queue_sender(servicebus_queue.name) as sender:
                 sender.send_messages(ServiceBusMessage("test session sender", session_id="test"))
+            
+            with sb_client.get_queue_receiver(servicebus_queue.name) as receiver:
+                messages = receiver.receive_messages()
+                for message in messages:
+                    receiver.complete_message(message)
 
 
     @pytest.mark.liveTest
@@ -2498,57 +2503,28 @@ def _hack_amqp_sender_run(self, **kwargs):
     @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_connection_string=None, servicebus_queue=None, **kwargs):
-        def _hack_amqp_sender_run(self, **kwargs):
-            time.sleep(6)  # sleep until timeout
-            if uamqp_transport:
-                self.message_handler.work()
-                self._waiting_messages = 0
-                self._pending_messages = self._filter_pending()
-                if self._backoff and not self._waiting_messages:
-                    _logger.info("Client told to backoff - sleeping for %r seconds", self._backoff)
-                    self._connection.sleep(self._backoff)
-                    self._backoff = 0
-                self._connection.work()
-            else:
-                try:
-                    self._link.update_pending_deliveries()
-                    self._connection.listen(wait=self._socket_timeout, **kwargs)
-                except ValueError:
-                    self._shutdown = True
-                    return False
-            return True
-
+        # Amqp
         with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string, logging_enable=False, uamqp_transport=uamqp_transport) as sb_client:
-            with sb_client.get_queue_sender(servicebus_queue.name) as sender:
-                # this one doesn't need to reset the method, as it's hacking the method on the instance
-                sender._handler._client_run = types.MethodType(_hack_amqp_sender_run, sender._handler)
-                with pytest.raises(OperationTimeoutError):
-                    sender.send_messages(ServiceBusMessage("body"), timeout=5)
-
-        if not uamqp_transport:
-            # Amqp
-            with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string,
-                uamqp_transport=uamqp_transport
-            ) as sb_client:
-                with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=0.6) as sender:
-                    payload = "A" * 250 * 1024
-                    sender.send_messages(ServiceBusMessage(payload))
+            servicebus_namespace_connection_string,
+            uamqp_transport=uamqp_transport
+        ) as sb_client:
+            with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=0.6) as sender:
+                payload = "A" * 250 * 1024
+                sender.send_messages(ServiceBusMessage(payload))
 
-            if uamqp:
-                transport_type = uamqp.constants.TransportType.AmqpOverWebsocket
-            else:
-                transport_type = TransportType.AmqpOverWebsocket
-            # AmqpOverWebsocket
-            with ServiceBusClient.from_connection_string(
-                servicebus_namespace_connection_string,
-                transport_type=transport_type,
-                uamqp_transport=uamqp_transport
-            ) as sb_client:
-                with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=0.8) as sender:
-                    payload = "A" * 250 * 1024
-                    sender.send_messages(ServiceBusMessage(payload))
+        if uamqp:
+            transport_type = uamqp.constants.TransportType.AmqpOverWebsocket
+        else:
+            transport_type = TransportType.AmqpOverWebsocket
+        # AmqpOverWebsocket
+        with ServiceBusClient.from_connection_string(
+            servicebus_namespace_connection_string,
+            transport_type=transport_type,
+            uamqp_transport=uamqp_transport
+        ) as sb_client:
+            with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=0.8) as sender:
+                payload = "A" * 250 * 1024
+                sender.send_messages(ServiceBusMessage(payload))
 
         # ReceiveMessages
         with ServiceBusClient.from_connection_string(