# ------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for # license information. # -------------------------------------------------------------------------- import logging import sys import os import json from concurrent.futures import ThreadPoolExecutor import types import pytest import time import uuid from datetime import datetime, timedelta import calendar import unittest import pickle try: import uamqp from azure.servicebus._transport._uamqp_transport import UamqpTransport except ImportError: uamqp = None from azure.servicebus._transport._pyamqp_transport import PyamqpTransport from azure.servicebus._pyamqp.message import Message from azure.servicebus._pyamqp import error, client, management_operation from azure.servicebus._pyamqp._decode import decode_payload from azure.servicebus import ( ServiceBusClient, AutoLockRenewer, TransportType, ServiceBusMessage, ServiceBusMessageBatch, ServiceBusReceivedMessage, ServiceBusReceiveMode, ServiceBusSubQueue, ServiceBusMessageState, ) from azure.servicebus.amqp import ( AmqpMessageHeader, AmqpMessageBodyType, AmqpAnnotatedMessage, AmqpMessageProperties, ) from azure.servicebus._common.constants import ( _X_OPT_LOCK_TOKEN, _X_OPT_PARTITION_KEY, _X_OPT_VIA_PARTITION_KEY, _X_OPT_SCHEDULED_ENQUEUE_TIME, ServiceBusMessageState, ) from azure.servicebus._common.utils import utc_now from azure.servicebus.management._models import DictMixin from azure.servicebus.exceptions import ( ServiceBusConnectionError, ServiceBusError, MessageLockLostError, MessageAlreadySettled, AutoLockRenewTimeout, MessageSizeExceededError, OperationTimeoutError, ) from devtools_testutils import AzureMgmtRecordedTestCase, get_credential from servicebus_preparer import ( SERVICEBUS_ENDPOINT_SUFFIX, CachedServiceBusNamespacePreparer, ServiceBusQueuePreparer, CachedServiceBusQueuePreparer, CachedServiceBusResourceGroupPreparer, ) from utilities import get_logger, print_message, sleep_until_expired from mocks import MockReceivedMessage, MockReceiver from utilities import uamqp_transport as get_uamqp_transport, ArgPasser uamqp_transport_params, uamqp_transport_ids = get_uamqp_transport() _logger = get_logger(logging.DEBUG) # A note regarding live_test_only. # Old servicebus tests were not written to work on both stubs and live entities. # This disables those tests for non-live scenarios, and should be removed as tests # are ported to offline-compatible code. class TestServiceBusQueue(AzureMgmtRecordedTestCase): @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_receive_and_delete_reconnect_interaction( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): # Note: This test was to guard against github issue 7079 fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() sb_client = ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(5): sender.send_messages(ServiceBusMessage("ServiceBusMessage {}".format(i))) with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=10 ) as receiver: batch = receiver.receive_messages() count = len(batch) for message in receiver: _logger.debug(message) count += 1 assert count == 5 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer() @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT5S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_github_issue_6178(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(3): sender.send_messages(ServiceBusMessage("ServiceBusMessage {}".format(i))) with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=60) as receiver: for message in receiver: _logger.debug(message) _logger.debug(message.sequence_number) _logger.debug(message.enqueued_time_utc) _logger.debug(message._lock_expired) receiver.complete_message(message) time.sleep(10) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_by_queue_client_conn_str_receive_handler_peeklock( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) for i in range(10): message = ServiceBusMessage("Handler message no. {}".format(i)) message.application_properties = {"key": "value"} message.subject = "label" message.content_type = "application/text" message.correlation_id = "cid" message.message_id = str(i) message.partition_key = "pk" message.to = "to" message.reply_to = "reply_to" sender.send_messages(message) # Test that noop empty send works properly. sender.send_messages([]) sender.send_messages(ServiceBusMessageBatch()) assert len(sender.schedule_messages([], utc_now())) == 0 sender.cancel_scheduled_messages([]) sender.close() # Then test expected failure modes. with pytest.raises(ValueError): with sender: raise AssertionError("Should raise ValueError") with pytest.raises(ValueError): sender.send_messages(ServiceBusMessage("msg")) with pytest.raises(ValueError): sender.schedule_messages(ServiceBusMessage("msg"), utc_now()) with pytest.raises(ValueError): sender.cancel_scheduled_messages([1, 2, 3]) with pytest.raises(ValueError): sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=0) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) assert len(receiver.receive_deferred_messages([])) == 0 with pytest.raises(ValueError): receiver.receive_messages(max_wait_time=0) count = 0 for message in receiver: print_message(_logger, message) assert message.delivery_count == 0 assert message.application_properties assert message.application_properties[b"key"] == b"value" assert message.subject == "label" assert message.content_type == "application/text" assert message.correlation_id == "cid" assert message.message_id == str(count) assert message.to == "to" assert message.reply_to == "reply_to" assert message.sequence_number assert message.enqueued_time_utc assert message.message.delivery_tag is not None assert message.lock_token == message.message.delivery_annotations.get(_X_OPT_LOCK_TOKEN) assert message.lock_token == uuid.UUID(bytes_le=message.message.delivery_tag) assert not message.scheduled_enqueue_time_utc assert not message.time_to_live assert not message.session_id assert not message.reply_to_session_id count += 1 receiver.complete_message(message) receiver.close() with pytest.raises(ValueError): receiver.receive_messages() with pytest.raises(ValueError): with receiver: raise AssertionError("Should raise ValueError") with pytest.raises(ValueError): receiver.receive_deferred_messages([1, 2, 3]) with pytest.raises(ValueError): receiver.peek_messages() assert count == 10 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_by_queue_client_conn_str_receive_handler_release_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: def sub_test_releasing_messages(): # test releasing messages when prefetch is 1 and link credits are issue dynamically receiver = sb_client.get_queue_receiver(servicebus_queue.name) sender = sb_client.get_queue_sender(servicebus_queue.name) with sender, receiver: # send 10 msgs to queue first 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: # issue link credits more than 5, client should consume 5 msgs from the service in total, # leaving the extra credits on the wire for msg in receiver.receive_messages(max_message_count=10, max_wait_time=5): receiver.complete_message(msg) received_msgs.append(msg) assert len(received_msgs) == 5 # send 5 more messages, those messages would arrive at the client while the program is sleeping sender.send_messages([ServiceBusMessage("test") for _ in range(5)]) time.sleep(15) # sleep > message expiration time target_msgs_count = 5 received_msgs = [] while len(received_msgs) < target_msgs_count: # issue 10 link credits, client should consume 5 msgs from the service, leaving no link credits for msg in receiver.receive_messages( max_message_count=target_msgs_count - len(received_msgs), max_wait_time=5 ): assert msg.delivery_count == 0 # release would not increase delivery count receiver.complete_message(msg) received_msgs.append(msg) assert len(received_msgs) == 5 def sub_test_releasing_messages_iterator(): # test nested iterator scenario receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) sender = sb_client.get_queue_sender(servicebus_queue.name) with sender, receiver: # send 5 msgs to queue first sender.send_messages([ServiceBusMessage("test") for _ in range(5)]) first_time = True iterator_recv_cnt = 0 # case: iterator + receive batch for msg in receiver: assert msg.delivery_count == 0 # release would not increase delivery count receiver.complete_message(msg) iterator_recv_cnt += 1 if first_time: # for the first time, we call nested receive message call received_msgs = [] while len(received_msgs) < 4: # there supposed to be 5 msgs in the queue # we issue 10 link credits, leaving more credits on the wire for sub_msg in receiver.receive_messages(max_message_count=10, max_wait_time=5): assert sub_msg.delivery_count == 0 receiver.complete_message(sub_msg) received_msgs.append(sub_msg) assert len(received_msgs) == 4 sender.send_messages([ServiceBusMessage("test") for _ in range(10)]) time.sleep(15) # sleep > message expiration time received_msgs = [] target_msgs_count = 5 # we want to receive 5 with the receive message call while len(received_msgs) < target_msgs_count: for sub_msg in receiver.receive_messages( max_message_count=target_msgs_count - len(received_msgs), max_wait_time=5 ): assert sub_msg.delivery_count == 0 # release would not increase delivery count receiver.complete_message(sub_msg) received_msgs.append(sub_msg) assert len(received_msgs) == target_msgs_count first_time = False assert ( iterator_recv_cnt == 6 ) # 1 before nested receive message call + 5 after nested receive message call # case: iterator + iterator case sender.send_messages([ServiceBusMessage("test") for _ in range(10)]) outter_recv_cnt = 0 inner_recv_cnt = 0 for msg in receiver: assert msg.delivery_count == 0 outter_recv_cnt += 1 receiver.complete_message(msg) for sub_msg in receiver: assert sub_msg.delivery_count == 0 inner_recv_cnt += 1 receiver.complete_message(sub_msg) if inner_recv_cnt == 5: time.sleep( 15 ) # innner finish receiving first 5 messages then sleep until lock expiration break assert outter_recv_cnt == 1 outter_recv_cnt = 0 for msg in receiver: assert msg.delivery_count == 0 outter_recv_cnt += 1 receiver.complete_message(msg) assert outter_recv_cnt == 4 def sub_test_non_releasing_messages(): # test not releasing messages when prefetch is not 1 receiver = sb_client.get_queue_receiver(servicebus_queue.name) sender = sb_client.get_queue_sender(servicebus_queue.name) if uamqp_transport: def _hack_disable_receive_context_message_received(self, message): # pylint: disable=protected-access self._handler._was_message_received = True self._handler._received_messages.put(message) else: def _hack_disable_receive_context_message_received(self, frame, message): # pylint: disable=protected-access self._handler._last_activity_timestamp = time.time() self._handler._received_messages.put((frame, message)) with sender, receiver: # send 5 msgs to queue first sender.send_messages([ServiceBusMessage("test") for _ in range(5)]) if uamqp_transport: receiver._handler.message_handler.on_message_received = types.MethodType( _hack_disable_receive_context_message_received, receiver ) else: receiver._handler._link._on_transfer = types.MethodType( _hack_disable_receive_context_message_received, receiver ) received_msgs = [] while len(received_msgs) < 5: # issue 10 link credits, client should consume 5 msgs from the service # leaving 5 credits on the wire for msg in receiver.receive_messages(max_message_count=10, max_wait_time=5): receiver.complete_message(msg) received_msgs.append(msg) assert len(received_msgs) == 5 # send 5 more messages, those messages would arrive at the client while the program is sleeping sender.send_messages([ServiceBusMessage("test") for _ in range(5)]) time.sleep(15) # sleep > message expiration time # issue 5 link credits, client should consume 5 msgs from the internal buffer which is already lock expired target_msgs_count = 5 received_msgs = [] while len(received_msgs) < target_msgs_count: received_msgs.extend(receiver.receive_messages(max_message_count=5, max_wait_time=5)) assert len(received_msgs) == 5 for msg in received_msgs: # queue ordering I think assert msg.delivery_count == 0 with pytest.raises(ServiceBusError): receiver.complete_message(msg) # re-received message with delivery count increased target_msgs_count = 5 received_msgs = [] while len(received_msgs) < target_msgs_count: received_msgs.extend(receiver.receive_messages(max_message_count=5, max_wait_time=5)) assert len(received_msgs) == 5 for msg in received_msgs: assert msg.delivery_count > 0 receiver.complete_message(msg) sub_test_releasing_messages() sub_test_releasing_messages_iterator() sub_test_non_releasing_messages() @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_queue_client_send_multiple_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) with sender: messages = [] for i in range(10): message = ServiceBusMessage("Handler message no. {}".format(i)) message.partition_key = "pkey" message.time_to_live = timedelta(seconds=60) message.scheduled_enqueue_time_utc = utc_now() + timedelta(seconds=60) message.partition_key = None message.time_to_live = None message.scheduled_enqueue_time_utc = None message.session_id = None messages.append(message) sender.send_messages(messages) with pytest.raises(ValueError): with sender: raise AssertionError("Should raise ValueError") with pytest.raises(ValueError): sender.send_messages(ServiceBusMessage("msg")) with pytest.raises(ValueError): sender.schedule_messages(ServiceBusMessage("msg"), utc_now()) with pytest.raises(ValueError): sender.cancel_scheduled_messages([1, 2, 3]) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) with receiver: count = 0 for message in receiver: print_message(_logger, message) assert message.delivery_count == 0 assert not message.application_properties assert not message.subject assert not message.content_type assert not message.correlation_id assert not message.to assert not message.reply_to assert not message.scheduled_enqueue_time_utc assert not message.time_to_live assert not message.session_id assert not message.reply_to_session_id count += 1 receiver.complete_message(message) assert count == 10 with pytest.raises(ValueError): receiver.receive_messages() with pytest.raises(ValueError): with receiver: raise AssertionError("Should raise ValueError") with pytest.raises(ValueError): receiver.receive_deferred_messages([1, 2, 3]) with pytest.raises(ValueError): receiver.peek_messages() sender = sb_client.get_queue_sender(servicebus_queue.name) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) with sender, receiver: # send previously unpicklable message msg = { "body": "W1tdLCB7ImlucHV0X2lkIjogNH0sIHsiY2FsbGJhY2tzIjogbnVsbCwgImVycmJhY2tzIjogbnVsbCwgImNoYWluIjogbnVsbCwgImNob3JkIjogbnVsbH1d", "content-encoding": "utf-8", "content-type": "application/json", "headers": { "lang": "py", "task": "tasks.example_task", "id": "7c66557d-e4bc-437f-b021-b66dcc39dfdf", "shadow": None, "eta": "2021-10-07T02:30:23.764066+00:00", "expires": None, "group": None, "group_index": None, "retries": 1, "timelimit": [None, None], "root_id": "7c66557d-e4bc-437f-b021-b66dcc39dfdf", "parent_id": "7c66557d-e4bc-437f-b021-b66dcc39dfdf", "argsrepr": "()", "kwargsrepr": "{'input_id': 4}", "origin": "gen36@94713e01a9c0", "ignore_result": 1, "x_correlator": "44a1978d-c869-4173-afe4-da741f0edfb9", }, "properties": { "correlation_id": "7c66557d-e4bc-437f-b021-b66dcc39dfdf", "reply_to": "7b9a3672-2fed-3e9b-8bfd-23ae2397d9ad", "origin": "gen68@c33d4eef123a", "delivery_mode": 2, "delivery_info": {"exchange": "", "routing_key": "celery_task_queue"}, "priority": 0, "body_encoding": "base64", "delivery_tag": "dc83ddb6-8cdc-4413-b88a-06c56cbde90d", }, } sender.send_messages(ServiceBusMessage(json.dumps(msg))) messages = receiver.receive_messages(max_wait_time=10, max_message_count=1) # complete first then pickle receiver.complete_message(messages[0]) if not uamqp_transport: pickled = pickle.loads(pickle.dumps(messages[0])) assert json.loads(str(pickled)) == json.loads(str(messages[0])) else: with pytest.raises(TypeError): pickled = pickle.loads(pickle.dumps(messages[0])) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_by_queue_client_conn_str_receive_handler_receiveanddelete( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Handler message no. {}".format(i)) sender.send_messages(message) messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=8 ) as receiver: for message in receiver: assert not message.application_properties assert not message.subject assert not message.content_type assert not message.correlation_id assert not message.partition_key assert not message.to assert not message.reply_to assert not message.scheduled_enqueue_time_utc assert not message.time_to_live assert not message.session_id assert not message.reply_to_session_id messages.append(message) with pytest.raises(ValueError): receiver.complete_message(message) with pytest.raises(ValueError): # RECEIVE_AND_DELETE messages cannot be lock renewed. renewer = AutoLockRenewer() renewer.register(receiver, message) assert len(messages) == 10 assert not receiver._running time.sleep(10) messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=5 ) as receiver: for message in receiver: messages.append(message) assert len(messages) == 0 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_by_queue_client_conn_str_receive_handler_receiveanddelete_prefetch( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: # send 10 messages with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Handler message no. {}".format(i)) sender.send_messages(message) # check peek_messages returns correctly, with default prefetch_count = 0 messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=10 ) as receiver: # peek messages checks current state of queue, which should return 10 # since none were prefetched, added to internal queue, and deleted peeked_msgs = receiver.peek_messages(max_message_count=10, timeout=10) assert len(peeked_msgs) == 10 # iterator receives and deletes each message from SB queue for msg in receiver: messages.append(msg) assert len(messages) == 10 # queue should be empty now peeked_msgs = receiver.peek_messages(max_message_count=10, timeout=10) assert len(peeked_msgs) == 0 # send 10 messages with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Handler message no. {}".format(i)) sender.send_messages(message) # check peek_messages returns correctly, with default prefetch_count > 0 messages = [] # prefetch 2 messages from SB queue when receive is called and not on open with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, prefetch_count=2, max_wait_time=30, ) as receiver: # peek messages checks current state of SB queue, and returns 10 peeked_msgs = receiver.peek_messages(max_message_count=10, timeout=10) assert len(peeked_msgs) == 10 # receive 3 messages recvd_msgs = receiver.receive_messages(max_message_count=3, max_wait_time=10) assert len(recvd_msgs) == 3 # receive rest of messages in queue for msg in receiver: messages.append(msg) assert len(messages) == 7 # queue should be empty now peeked_msgs = receiver.peek_messages(max_message_count=10, timeout=10) assert len(peeked_msgs) == 0 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_receive_handler_peek_check_uniq_corr_id( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: # send 10 messages with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Handler message no. {}".format(i)) sender.send_messages(message) def _hack_parse_received_message(message, message_type, **kwargs): parsed = [] # Get correlation ID from mgmt msg wrapper. Individual msgs in "value" do not have correlation IDs. correlation_ids.add(message.properties.correlation_id) if message.value: for m in message.value[b"messages"]: wrapped = decode_payload(memoryview(m[b"message"])) parsed.append(message_type(wrapped, **kwargs)) return parsed def peek_message(receiver, seq_num): receiver.peek_messages(max_message_count=1, sequence_number=seq_num) # Peek 10 unique messages concurrently with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: received_msgs = receiver.receive_messages(max_message_count=10, max_wait_time=5) while len(received_msgs) < 10: received_msgs.extend(receiver.receive_messages(max_message_count=1, max_wait_time=5)) assert len(received_msgs) == 10 # Grab the sequence numbers of the received messages seq_nums = set() for msg in received_msgs: seq_nums.add(int(msg.sequence_number)) # Use the sequence numbers to peek messages concurrently # Add the correlation IDs of the peeked msgs to a set # If the correlation ID is not unique, the set will be smaller than 10 # Otherwise, the set will be 10 correlation_ids = set() receiver._amqp_transport.parse_received_message = _hack_parse_received_message with ThreadPoolExecutor(max_workers=10) as executor: for seq_num in seq_nums: executor.submit(peek_message, receiver, seq_num) assert len(correlation_ids) == 10 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_queue_client_conn_str_receive_handler_with_stop( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Stop message no. {}".format(i)) sender.send_messages(message) messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) as receiver: for message in receiver: messages.append(message) receiver.complete_message(message) if len(messages) >= 5: break assert receiver._running assert len(messages) == 5 for message in receiver: messages.append(message) receiver.complete_message(message) if len(messages) >= 5: break assert not receiver._running assert len(messages) == 6 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_iter_messages_simple( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=10, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Iter message no. {}".format(i)) sender.send_messages(message) count = 0 for message in receiver: print_message(_logger, message) receiver.complete_message(message) with pytest.raises(MessageAlreadySettled): receiver.complete_message(message) with pytest.raises(MessageAlreadySettled): receiver.renew_message_lock(message) count += 1 with pytest.raises(StopIteration): next(receiver) assert count == 10 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_conn_str_client_iter_messages_with_abandon( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Abandoned message no. {}".format(i)) sender.send_messages(message) count = 0 for message in receiver: print_message(_logger, message) if not message.delivery_count: count += 1 receiver.abandon_message(message) else: assert message.delivery_count == 1 receiver.complete_message(message) assert count == 10 with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=20, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: count = 0 for message in receiver: print_message(_logger, message) receiver.complete_message(message) count += 1 assert count == 0 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_iter_messages_with_defer( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: deferred_messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Deferred message no. {}".format(i)) sender.send_messages(message) count = 0 for message in receiver: deferred_messages.append(message.sequence_number) print_message(_logger, message) count += 1 receiver.defer_message(message) assert count == 10 with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: count = 0 for message in receiver: print_message(_logger, message) receiver.complete_message(message) count += 1 assert count == 0 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_iter_messages_with_retrieve_deferred_client( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: deferred_messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Deferred message no. {}".format(i)) sender.send_messages(message) count = 0 for message in receiver: deferred_messages.append(message.sequence_number) print_message(_logger, message) count += 1 receiver.defer_message(message) assert count == 10 deferred = receiver.receive_deferred_messages(deferred_messages) assert len(deferred) == 10 for message in deferred: assert isinstance(message, ServiceBusReceivedMessage) receiver.complete_message(message) with pytest.raises(ServiceBusError): receiver.receive_deferred_messages(deferred_messages) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_iter_messages_with_retrieve_deferred_receiver_complete( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: deferred_messages = [] for i in range(10): message = ServiceBusMessage("Deferred message no. {}".format(i), session_id="test_session") sender.send_messages(message) with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: count = 0 for message in receiver: deferred_messages.append(message.sequence_number) print_message(_logger, message) count += 1 receiver.defer_message(message) assert count == 10 with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: deferred = receiver.receive_deferred_messages(deferred_messages) assert len(deferred) == 10 for message in deferred: assert isinstance(message, ServiceBusReceivedMessage) assert message.lock_token assert message.locked_until_utc assert message._receiver receiver.renew_message_lock(message) receiver.complete_message(message) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_iter_messages_with_retrieve_deferred_receiver_deadletter( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: deferred_messages = [] for i in range(10): message = ServiceBusMessage("Deferred message no. {}".format(i)) sender.send_messages(message) with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=10, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: count = 0 for message in receiver: deferred_messages.append(message.sequence_number) print_message(_logger, message) count += 1 receiver.defer_message(message) assert count == 10 with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) as receiver: deferred = receiver.receive_deferred_messages(deferred_messages) assert len(deferred) == 10 for message in deferred: assert isinstance(message, ServiceBusReceivedMessage) receiver.dead_letter_message( message, reason="Testing reason", error_description="Testing description" ) count = 0 with sb_client.get_queue_receiver( servicebus_queue.name, sub_queue=ServiceBusSubQueue.DEAD_LETTER, max_wait_time=10 ) as receiver: for message in receiver: count += 1 print_message(_logger, message) assert message.dead_letter_reason == "Testing reason" assert message.dead_letter_error_description == "Testing description" assert message.application_properties[b"DeadLetterReason"] == b"Testing reason" assert message.application_properties[b"DeadLetterErrorDescription"] == b"Testing description" receiver.complete_message(message) assert count == 10 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_iter_messages_with_retrieve_deferred_receiver_deletemode( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): sender.send_messages(ServiceBusMessage("Deferred message no. {}".format(i))) deferred_messages = [] count = 0 with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: for message in receiver: deferred_messages.append(message.sequence_number) print_message(_logger, message) count += 1 receiver.defer_message(message) assert count == 10 with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE.value, max_wait_time=5 ) as receiver: deferred = receiver.receive_deferred_messages(deferred_messages) assert len(deferred) == 10 for message in deferred: assert isinstance(message, ServiceBusReceivedMessage) with pytest.raises(ValueError): receiver.complete_message(message) with pytest.raises(ServiceBusError): deferred = receiver.receive_deferred_messages(deferred_messages) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_iter_messages_with_retrieve_deferred_not_found( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: deferred_messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(3): message = ServiceBusMessage("Deferred message no. {}".format(i)) sender.send_messages(message) count = 0 for message in receiver: deferred_messages.append(message.sequence_number) print_message(_logger, message) count += 1 receiver.defer_message(message) assert count == 3 with pytest.raises(ServiceBusError): deferred = receiver.receive_deferred_messages([3, 4]) with pytest.raises(ServiceBusError): deferred = receiver.receive_deferred_messages([5, 6, 7]) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_receive_batch_with_deadletter( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=10 ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Dead lettered message no. {}".format(i)) sender.send_messages(message) count = 0 messages = receiver.receive_messages() while messages: for message in messages: print_message(_logger, message) count += 1 receiver.dead_letter_message( message, reason="Testing reason", error_description="Testing description" ) messages = receiver.receive_messages() assert count == 10 with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) as receiver: count = 0 for message in receiver: print_message(_logger, message) receiver.complete_message(message) count += 1 assert count == 0 with sb_client.get_queue_receiver( servicebus_queue.name, sub_queue=ServiceBusSubQueue.DEAD_LETTER.value, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, ) as dl_receiver: count = 0 for message in dl_receiver: dl_receiver.complete_message(message) count += 1 assert message.dead_letter_reason == "Testing reason" assert message.dead_letter_error_description == "Testing description" assert message.application_properties[b"DeadLetterReason"] == b"Testing reason" assert message.application_properties[b"DeadLetterErrorDescription"] == b"Testing description" assert count == 10 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_receive_batch_with_retrieve_deadletter( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=10 ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("Dead lettered message no. {}".format(i)) sender.send_messages(message) count = 0 messages = receiver.receive_messages() while messages: for message in messages: print_message(_logger, message) receiver.dead_letter_message( message, reason="Testing reason", error_description="Testing description" ) count += 1 messages = receiver.receive_messages() receiver.receive_messages(1, 5) assert count == 10 with sb_client.get_queue_receiver( servicebus_queue.name, sub_queue=ServiceBusSubQueue.DEAD_LETTER, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, ) as dl_receiver: count = 0 for message in dl_receiver: print_message(_logger, message) assert message.dead_letter_reason == "Testing reason" assert message.dead_letter_error_description == "Testing description" assert message.application_properties[b"DeadLetterReason"] == b"Testing reason" assert message.application_properties[b"DeadLetterErrorDescription"] == b"Testing description" dl_receiver.complete_message(message) count += 1 assert count == 10 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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 ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with pytest.raises(ServiceBusError): sb_client.get_queue_receiver(servicebus_queue.name, session_id="test")._open_with_retry() 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: assert str(message) == "test session sender" receiver.complete_message(message) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_browse_messages_client( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(5): message = ServiceBusMessage("Test message no. {}".format(i)) sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.peek_messages(5) assert len(messages) == 5 assert all(isinstance(m, ServiceBusReceivedMessage) for m in messages) for message in messages: print_message(_logger, message) with pytest.raises(ValueError): receiver.complete_message(message) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_browse_messages_with_receiver( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: receiver = sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK ) sender = sb_client.get_queue_sender(servicebus_queue.name) with receiver, sender: for i in range(5): message = ServiceBusMessage( body="Test message", application_properties={"key": "value"}, subject="label", content_type="application/text", correlation_id="cid", message_id="mid", to="to", reply_to="reply_to", time_to_live=timedelta(seconds=60), ) sender.send_messages(message) messages = receiver.peek_messages(5) assert len(messages) > 0 assert all(isinstance(m, ServiceBusReceivedMessage) for m in messages) for message in messages: print_message(_logger, message) assert b"".join(message.body) == b"Test message" assert message.application_properties[b"key"] == b"value" assert message.subject == "label" assert message.content_type == "application/text" assert message.correlation_id == "cid" assert message.message_id == "mid" assert message.to == "to" assert message.reply_to == "reply_to" assert message.time_to_live == timedelta(seconds=60) with pytest.raises(ValueError): receiver.complete_message(message) sender.send_messages(message) cnt = 0 for message in receiver: assert b"".join(message.body) == b"Test message" assert message.application_properties[b"key"] == b"value" assert message.subject == "label" assert message.content_type == "application/text" assert message.correlation_id == "cid" assert message.message_id == "mid" assert message.to == "to" assert message.reply_to == "reply_to" assert message.time_to_live == timedelta(seconds=60) receiver.complete_message(message) cnt += 1 assert cnt == 10 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_browse_empty_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=10 ) as receiver: messages = receiver.peek_messages(10) assert len(messages) == 0 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer(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_fail_send_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: too_large = "A" * 256 * 1024 with sb_client.get_queue_sender(servicebus_queue.name, socket_timeout=0.5) as sender: with pytest.raises(MessageSizeExceededError): sender.send_messages(ServiceBusMessage(too_large)) half_too_large = "A" * int((1024 * 256) / 2) with pytest.raises(MessageSizeExceededError): sender.send_messages([ServiceBusMessage(half_too_large), ServiceBusMessage(half_too_large)]) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_renew_message_locks( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: messages = [] locks = 3 with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=10 ) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(locks): message = ServiceBusMessage("Test message no. {}".format(i)) sender.send_messages(message) messages.extend(receiver.receive_messages()) recv = True while recv: recv = receiver.receive_messages() messages.extend(recv) try: for message in messages: assert not message._lock_expired time.sleep(5) initial_expiry = message.locked_until_utc receiver.renew_message_lock(message) assert (message.locked_until_utc - initial_expiry) >= timedelta(seconds=5) finally: receiver.complete_message(messages[0]) receiver.complete_message(messages[1]) assert (messages[2].locked_until_utc - utc_now()) <= timedelta(seconds=60) sleep_until_expired(messages[2]) with pytest.raises(MessageLockLostError): receiver.complete_message(messages[2]) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_by_queue_client_conn_str_receive_handler_with_autolockrenew( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("{}".format(i)) sender.send_messages(message) renewer = AutoLockRenewer() messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=10, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=10 ) as receiver: for message in receiver: if not messages: messages.append(message) assert not message._lock_expired renewer.register(receiver, message, max_lock_renewal_duration=10) print("Registered lock renew thread", message.locked_until_utc, utc_now()) time.sleep(10) print("Finished first sleep", message.locked_until_utc) assert not message._lock_expired time.sleep(15) # generate autolockrenewtimeout error by going one iteration past. sleep_until_expired(message) print("Finished second sleep", message.locked_until_utc, utc_now()) assert message._lock_expired try: receiver.complete_message(message) raise AssertionError("Didn't raise MessageLockLostError") except MessageLockLostError as e: assert isinstance(e.inner_exception, AutoLockRenewTimeout) else: if message._lock_expired: print("Remaining messages", message.locked_until_utc, utc_now()) assert message._lock_expired with pytest.raises(MessageLockLostError): receiver.complete_message(message) else: assert message.delivery_count >= 1 print("Remaining messages", message.locked_until_utc, utc_now()) messages.append(message) receiver.complete_message(message) renewer.close() assert len(messages) == 11 renewer = AutoLockRenewer(max_workers=8) with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(10): message = ServiceBusMessage("{}".format(i)) sender.send_messages(message) with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=10, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, 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=30) time.sleep(10) for msg in received_msgs: receiver.complete_message(msg) assert len(received_msgs) == 10 renewer.close() executor = ThreadPoolExecutor(max_workers=1) renewer = AutoLockRenewer(executor=executor) with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(2): message = ServiceBusMessage("{}".format(i)) sender.send_messages(message) with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=10, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=3 ) as receiver: received_msgs = receiver.receive_messages(max_message_count=3, max_wait_time=10) for msg in received_msgs: renewer.register(receiver, msg, max_lock_renewal_duration=30) time.sleep(10) for msg in received_msgs: receiver.complete_message(msg) assert len(received_msgs) == 2 assert not renewer._is_max_workers_greater_than_one renewer.close() executor = ThreadPoolExecutor(max_workers=2) renewer = AutoLockRenewer(executor=executor) with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(3): message = ServiceBusMessage("{}".format(i)) sender.send_messages(message) with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=10, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=3 ) as receiver: received_msgs = receiver.receive_messages(max_message_count=3, max_wait_time=10) for msg in received_msgs: renewer.register(receiver, msg, max_lock_renewal_duration=30) time.sleep(10) for msg in received_msgs: receiver.complete_message(msg) assert len(received_msgs) == 3 assert renewer._is_max_workers_greater_than_one renewer.close() @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_by_queue_client_conn_str_receive_handler_with_auto_autolockrenew( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: # The 10 iterations is "important" because it gives time for the timed out message to be received again. for i in range(10): message = ServiceBusMessage("{}".format(i)) sender.send_messages(message) renewer = AutoLockRenewer(max_lock_renewal_duration=10) messages = [] with sb_client.get_queue_receiver( servicebus_queue.name, max_wait_time=10, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, prefetch_count=10, auto_lock_renewer=renewer, ) as receiver: for message in receiver: if not messages: messages.append(message) assert not message._lock_expired print("Registered lock renew thread", message.locked_until_utc, utc_now()) time.sleep(10) print("Finished first sleep", message.locked_until_utc) assert not message._lock_expired time.sleep(15) # generate autolockrenewtimeout error by going one iteration past. sleep_until_expired(message) print("Finished second sleep", message.locked_until_utc, utc_now()) assert message._lock_expired try: receiver.complete_message(message) raise AssertionError("Didn't raise MessageLockLostError") except MessageLockLostError as e: assert isinstance(e.inner_exception, AutoLockRenewTimeout) else: if message._lock_expired: print("Remaining messages", message.locked_until_utc, utc_now()) assert message._lock_expired with pytest.raises(MessageLockLostError): receiver.complete_message(message) else: assert message.delivery_count >= 1 print("Remaining messages", message.locked_until_utc, utc_now()) messages.append(message) receiver.complete_message(message) renewer.close() assert len(messages) == 11 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT55S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_receive_handler_with_autolockrenew_many_msgs( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): # number of messages to receive and renew locks for max_msg_count = 300 # only tested on sync since async can concurrently send renew-lock requests without waiting for the transfer frame from the service if sys.platform.startswith("darwin"): pytest.skip("Skipping since ALR renewal is slower, so message locks are expiring. Potentially due to lower # of available threads/workers.") fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=True, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: msgs_to_send = [ServiceBusMessage("message: {}".format(i)) for i in range(400)] sender.send_messages(msgs_to_send) # Can also be called via "with AutoLockRenewer() as renewer" to automate shutdown. renewer = AutoLockRenewer(max_lock_renewal_duration=60*5) with sb_client.get_queue_receiver(queue_name=servicebus_queue.name, auto_lock_renewer=renewer) as receiver: received_msgs = receiver.receive_messages(max_message_count=max_msg_count) # At least 300 messages should be renewed # TODO: This should be bumped up once sync alr perf is improved while len(received_msgs) < max_msg_count: count = max_msg_count - len(received_msgs) received_msgs.extend(receiver.receive_messages(max_message_count=count)) assert len(received_msgs) == max_msg_count, "Did not receive all messages" time.sleep(100) for msg in received_msgs: receiver.complete_message(msg) # Settling the message deregisters it from the AutoLockRenewer renewer.close() @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_time_to_live( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: content = str(uuid.uuid4()) message_id = uuid.uuid4() message = ServiceBusMessage(content) message.time_to_live = timedelta(seconds=15) sender.send_messages(message) time.sleep(15) with sb_client.get_queue_receiver(servicebus_queue.name, prefetch_count=5) as receiver: messages = receiver.receive_messages(5, max_wait_time=10) assert not messages with sb_client.get_queue_receiver( servicebus_queue.name, sub_queue=ServiceBusSubQueue.DEAD_LETTER, max_wait_time=5, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, ) as dl_receiver: count = 0 for message in dl_receiver: print_message(_logger, message) dl_receiver.complete_message(message) count += 1 assert count == 1 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", requires_duplicate_detection=True, dead_lettering_on_message_expiration=True ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_message_duplicate_detection( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: message_id = uuid.uuid4() with sb_client.get_queue_sender(servicebus_queue.name) as sender: for i in range(5): message = ServiceBusMessage(str(i)) message.message_id = message_id sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: count = 0 for message in receiver: print_message(_logger, message) assert message.message_id == message_id receiver.complete_message(message) count += 1 assert count == 1 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_connection_closed( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: content = str(uuid.uuid4()) message = ServiceBusMessage(content) sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=10) assert len(messages) == 1 with pytest.raises(ValueError): receiver.complete_message(messages[0]) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_expiry(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: content = str(uuid.uuid4()) message = ServiceBusMessage(content) sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=10) assert len(messages) == 1 time.sleep((messages[0].locked_until_utc - utc_now()).total_seconds() + 1) assert messages[0]._lock_expired with pytest.raises(MessageLockLostError): receiver.complete_message(messages[0]) with pytest.raises(MessageLockLostError): receiver.renew_message_lock(messages[0]) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=30) assert len(messages) == 1 print_message(_logger, messages[0]) assert messages[0].delivery_count > 0 receiver.complete_message(messages[0]) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_lock_renew( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: content = str(uuid.uuid4()) message = ServiceBusMessage(content) sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=10) assert len(messages) == 1 time.sleep(15) receiver.renew_message_lock(messages[0]) time.sleep(15) receiver.renew_message_lock(messages[0]) time.sleep(15) assert not messages[0]._lock_expired receiver.complete_message(messages[0]) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=10) assert len(messages) == 0 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_message_receive_and_delete( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): if uamqp: transport_type = uamqp.constants.TransportType.Amqp else: transport_type = TransportType.Amqp fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, transport_time=transport_type, uamqp_transport=uamqp_transport, ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message = ServiceBusMessage("Receive and delete test") sender.send_messages(message) with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE ) as receiver: messages = receiver.receive_messages(max_wait_time=10) assert len(messages) == 1 message = messages[0] print_message(_logger, message) with pytest.raises(ValueError): receiver.complete_message(message) with pytest.raises(ValueError): receiver.abandon_message(message) with pytest.raises(ValueError): receiver.defer_message(message) with pytest.raises(ValueError): receiver.dead_letter_message(message) with pytest.raises(ValueError): receiver.renew_message_lock(message) time.sleep(10) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=10) for m in messages: print_message(_logger, m) assert len(messages) == 0 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_batch(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: def message_content(): for i in range(5): message = ServiceBusMessage("ServiceBusMessage no. {}".format(i)) message.application_properties = {"key": "value"} message.subject = "label" message.content_type = "application/text" message.correlation_id = "cid" message.message_id = str(i) message.to = "to" message.reply_to = "reply_to" message.time_to_live = timedelta(seconds=60) assert ( message.raw_amqp_message.properties.absolute_expiry_time == message.raw_amqp_message.properties.creation_time + message.raw_amqp_message.header.time_to_live ) yield message with sb_client.get_queue_sender(servicebus_queue.name) as sender: # sending manually created message batch (with default pyamqp) should work for both uamqp/pyamqp message = ServiceBusMessageBatch() for each in message_content(): message.add_message(each) sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=10) recv = True while recv: recv = receiver.receive_messages(max_wait_time=10) messages.extend(recv) assert len(messages) == 5 count = 0 for message in messages: assert message.delivery_count == 0 assert message.application_properties assert message.application_properties[b"key"] == b"value" assert message.subject == "label" assert message.content_type == "application/text" assert message.correlation_id == "cid" assert message.message_id == str(count) assert message.to == "to" assert message.reply_to == "reply_to" assert message.sequence_number assert message.enqueued_time_utc assert message.expires_at_utc == (message.enqueued_time_utc + timedelta(seconds=60)) print_message(_logger, message) receiver.complete_message(message) count += 1 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_batch_large( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: with pytest.raises(ValueError): batch_message = [ServiceBusMessage("test message " * 256 * 1024)] sender.send_messages(batch_message) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_schedule_message( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: scheduled_enqueue_time = (utc_now() + timedelta(minutes=2)).replace(microsecond=0) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: content = str(uuid.uuid4()) message_id = uuid.uuid4() message = ServiceBusMessage(content) message.message_id = message_id message.scheduled_enqueue_time_utc = scheduled_enqueue_time sender.send_messages(message) messages = receiver.receive_messages(max_wait_time=120) if messages: try: data = str(messages[0]) assert data == content assert messages[0].message_id == message_id assert messages[0].scheduled_enqueue_time_utc == scheduled_enqueue_time assert messages[0].scheduled_enqueue_time_utc <= messages[0].enqueued_time_utc.replace( microsecond=0 ) assert len(messages) == 1 finally: for message in messages: receiver.complete_message(message) else: raise Exception("Failed to receive schdeduled message.") @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_schedule_multiple_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: scheduled_enqueue_time = (utc_now() + timedelta(minutes=2)).replace(microsecond=0) sender = sb_client.get_queue_sender(servicebus_queue.name) receiver = sb_client.get_queue_receiver(servicebus_queue.name, prefetch_count=20, max_wait_time=5) with sender, receiver: content = str(uuid.uuid4()) message_id_a = uuid.uuid4() message_a = ServiceBusMessage(content) message_a.message_id = message_id_a message_id_b = uuid.uuid4() message_b = ServiceBusMessage(content) message_b.message_id = message_id_b message_arry = [message_a, message_b] for message in message_arry: message.application_properties = {"key": "value"} message.subject = "label" message.content_type = "application/text" message.correlation_id = "cid" message.to = "to" message.reply_to = "reply_to" sender.send_messages(message_arry) received_messages = [] for message in receiver: received_messages.append(message) receiver.complete_message(message) tokens = sender.schedule_messages(received_messages, scheduled_enqueue_time) assert len(tokens) == 2 messages = receiver.receive_messages(max_wait_time=120) messages.extend(receiver.receive_messages(max_wait_time=5)) if messages: try: data = str(messages[0]) assert data == content assert messages[0].message_id in (message_id_a, message_id_b) assert messages[0].scheduled_enqueue_time_utc == scheduled_enqueue_time assert messages[0].scheduled_enqueue_time_utc <= messages[0].enqueued_time_utc.replace( microsecond=0 ) assert messages[0].delivery_count == 0 assert messages[0].application_properties assert messages[0].application_properties[b"key"] == b"value" assert messages[0].subject == "label" assert messages[0].content_type == "application/text" assert messages[0].correlation_id == "cid" assert messages[0].to == "to" assert messages[0].reply_to == "reply_to" assert messages[0].sequence_number assert messages[0].enqueued_time_utc assert messages[0].message.delivery_tag is not None assert len(messages) == 2 if not uamqp_transport: pickled = pickle.loads(pickle.dumps(messages[0])) assert pickled.message_id == messages[0].message_id assert pickled.scheduled_enqueue_time_utc == messages[0].scheduled_enqueue_time_utc assert pickled.scheduled_enqueue_time_utc <= pickled.enqueued_time_utc.replace( microsecond=0 ) assert pickled.delivery_count == messages[0].delivery_count assert pickled.application_properties == messages[0].application_properties assert pickled.application_properties[b"key"] == messages[0].application_properties[b"key"] assert pickled.subject == messages[0].subject assert pickled.content_type == messages[0].content_type assert pickled.correlation_id == messages[0].correlation_id assert pickled.to == messages[0].to assert pickled.reply_to == messages[0].reply_to assert pickled.sequence_number assert pickled.enqueued_time_utc finally: for message in messages: receiver.complete_message(message) else: raise Exception("Failed to receive schdeduled message.") @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_cancel_scheduled_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: enqueue_time = (utc_now() + timedelta(minutes=2)).replace(microsecond=0) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message_a = ServiceBusMessage("Test scheduled message") message_b = ServiceBusMessage("Test scheduled message") tokens = sender.schedule_messages([message_a, message_b], enqueue_time) assert len(tokens) == 2 sender.cancel_scheduled_messages(tokens) messages = receiver.receive_messages(max_wait_time=120) try: assert len(messages) == 0 except AssertionError: for message in messages: print(str(message)) receiver.complete_message(message) raise @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_amqp_over_websocket( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, transport_type=TransportType.AmqpOverWebsocket, logging_enable=False, uamqp_transport=uamqp_transport, ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: assert sender._config.transport_type == TransportType.AmqpOverWebsocket message = ServiceBusMessage("Test") sender.send_messages(message) with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE ) as receiver: assert receiver._config.transport_type == TransportType.AmqpOverWebsocket messages = receiver.receive_messages(max_wait_time=5) assert len(messages) == 1 @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) def test_queue_message_http_proxy_setting(self, uamqp_transport): mock_fully_qualified_namespace = "mock_namespace" credential = get_credential() http_proxy = {"proxy_hostname": "127.0.0.1", "proxy_port": 8899, "username": "admin", "password": "123456"} sb_client = ServiceBusClient( mock_fully_qualified_namespace, credential, http_proxy=http_proxy, uamqp_transport=uamqp_transport ) assert sb_client._config.http_proxy == http_proxy assert sb_client._config.transport_type == TransportType.AmqpOverWebsocket sender = sb_client.get_queue_sender(queue_name="mock") assert sender._config.http_proxy == http_proxy assert sender._config.transport_type == TransportType.AmqpOverWebsocket receiver = sb_client.get_queue_receiver(queue_name="mock") assert receiver._config.http_proxy == http_proxy assert receiver._config.transport_type == TransportType.AmqpOverWebsocket @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_message_settle_through_mgmt_link_due_to_broken_receiver_link( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message = ServiceBusMessage("Test") sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: messages = receiver.receive_messages(max_wait_time=5) # destroy the underlying receiver link if uamqp_transport: receiver._handler.message_handler.destroy() else: receiver._handler._link.detach() assert len(messages) == 1 receiver.complete_message(messages[0]) @unittest.skip("hard to test") def test_queue_mock_auto_lock_renew_callback(self): # A warning to future devs: If the renew period override heuristic in registration # ever changes, it may break this (since it adjusts renew period if it is not short enough) results = [] errors = [] def callback_mock(renewable, error): results.append(renewable) if error: errors.append(error) receiver = MockReceiver() auto_lock_renew = AutoLockRenewer() with pytest.raises(TypeError): auto_lock_renew.register(Exception()) # an arbitrary invalid type. auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 # So we can run the test fast. with auto_lock_renew: # Check that it is called when the object expires for any reason (silent renew failure) message = MockReceivedMessage(prevent_renew_lock=True) auto_lock_renew.register(receiver, renewable=message, on_lock_renew_failure=callback_mock) time.sleep(3) assert len(results) == 1 and results[-1]._lock_expired is True assert not errors del results[:] del errors[:] auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 with auto_lock_renew: # Check that in normal operation it does not get called auto_lock_renew.register( receiver, renewable=MockReceivedMessage(lock_duration=5), on_lock_renew_failure=callback_mock ) time.sleep(6) assert not results assert not errors del results[:] del errors[:] auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 with auto_lock_renew: # Check that when a message is settled, it will not get called even after expiry message = MockReceivedMessage(prevent_renew_lock=True) auto_lock_renew.register(receiver, renewable=message, on_lock_renew_failure=callback_mock) message._settled = True time.sleep(3) assert not results assert not errors del results[:] del errors[:] auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 with auto_lock_renew: # Check that it is called when there is an overt renew failure message = MockReceivedMessage(exception_on_renew_lock=True) auto_lock_renew.register(receiver, renewable=message, on_lock_renew_failure=callback_mock) time.sleep(3) assert len(results) == 1 and results[-1]._lock_expired == True assert errors[-1] del results[:] del errors[:] auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 with auto_lock_renew: # Check that it is not called when the renewer is shutdown message = MockReceivedMessage(prevent_renew_lock=True) auto_lock_renew.register(receiver, renewable=message, on_lock_renew_failure=callback_mock) auto_lock_renew.close() time.sleep(3) assert not results assert not errors del results[:] del errors[:] auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 with auto_lock_renew: # Check that it is not called when the receiver is shutdown message = MockReceivedMessage(prevent_renew_lock=True) auto_lock_renew.register(receiver, renewable=message, on_lock_renew_failure=callback_mock) message._receiver._running = False time.sleep(3) assert not results assert not errors def test_queue_mock_no_reusing_auto_lock_renew(self): receiver = MockReceiver() auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 # So we can run the test fast. with auto_lock_renew: auto_lock_renew.register(receiver, renewable=MockReceivedMessage()) time.sleep(3) with pytest.raises(ServiceBusError): with auto_lock_renew: pass with pytest.raises(ServiceBusError): auto_lock_renew.register(receiver, renewable=MockReceivedMessage()) auto_lock_renew = AutoLockRenewer() auto_lock_renew._renew_period = 1 with auto_lock_renew: auto_lock_renew.register(receiver, renewable=MockReceivedMessage()) time.sleep(3) auto_lock_renew.close() with pytest.raises(ServiceBusError): with auto_lock_renew: pass with pytest.raises(ServiceBusError): auto_lock_renew.register(receiver, renewable=MockReceivedMessage()) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) def test_queue_message_properties(self, uamqp_transport): scheduled_enqueue_time = (utc_now() + timedelta(seconds=20)).replace(microsecond=0) message = ServiceBusMessage( body="data", application_properties={"key": "value"}, session_id="sid", subject="label", content_type="application/text", correlation_id="cid", message_id="mid", to="to", reply_to="reply_to", reply_to_session_id="reply_to_sid", scheduled_enqueue_time_utc=scheduled_enqueue_time, ) assert message.application_properties assert message.application_properties["key"] == "value" assert message.subject == "label" assert message.content_type == "application/text" assert message.correlation_id == "cid" assert message.message_id == "mid" assert message.to == "to" assert message.reply_to == "reply_to" assert message.session_id == "sid" assert message.reply_to_session_id == "reply_to_sid" assert message.scheduled_enqueue_time_utc == scheduled_enqueue_time message.partition_key = "sid" new_scheduled_time = (utc_now() + timedelta(hours=5)).replace(microsecond=0) message.scheduled_enqueue_time_utc = new_scheduled_time assert message.scheduled_enqueue_time_utc == new_scheduled_time message.via_partition_key = None message.scheduled_enqueue_time_utc = None assert message.scheduled_enqueue_time_utc is None try: timestamp = new_scheduled_time.timestamp() * 1000 except AttributeError: timestamp = calendar.timegm(new_scheduled_time.timetuple()) * 1000 my_frame = [0, 0, 0] if uamqp_transport: amqp_transport = UamqpTransport amqp_received_message = uamqp.message.Message( body=[b"data"], annotations={ _X_OPT_PARTITION_KEY: b"r_key", _X_OPT_VIA_PARTITION_KEY: b"r_via_key", _X_OPT_SCHEDULED_ENQUEUE_TIME: timestamp, }, properties={}, ) else: amqp_transport = PyamqpTransport amqp_received_message = Message( data=[b"data"], message_annotations={ _X_OPT_PARTITION_KEY: b"r_key", _X_OPT_VIA_PARTITION_KEY: b"r_via_key", _X_OPT_SCHEDULED_ENQUEUE_TIME: timestamp, }, properties={}, ) received_message = ServiceBusReceivedMessage( amqp_received_message, receiver=None, frame=my_frame, amqp_transport=amqp_transport ) assert received_message.scheduled_enqueue_time_utc == new_scheduled_time new_scheduled_time = utc_now() + timedelta(hours=1, minutes=49, seconds=32) received_message.scheduled_enqueue_time_utc = new_scheduled_time assert received_message.scheduled_enqueue_time_utc == new_scheduled_time received_message.scheduled_enqueue_time_utc = None assert message.scheduled_enqueue_time_utc is None @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_receive_batch_without_setting_prefetch( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: def message_content(): for i in range(20): yield ServiceBusMessage( body="Test message", # application_properties={'key': 'value'}, subject="1st", # content_type='application/text', # correlation_id='cid', # message_id='mid', # to='to', # reply_to='reply_to', # time_to_live=timedelta(seconds=60) ) sender = sb_client.get_queue_sender(servicebus_queue.name) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) with sender, receiver: message = ServiceBusMessageBatch() for each in message_content(): message.add_message(each) sender.send_messages(message) receive_counter = 0 message_1st_received_cnt = 0 message_2nd_received_cnt = 0 while message_1st_received_cnt < 20 or message_2nd_received_cnt < 20: messages = [] for message in receiver: messages.append(message) if not messages: break receive_counter += 1 for message in messages: print_message(_logger, message) # assert b''.join(message.body) == b'Test message' # assert message.application_properties[b'key'] == b'value' # assert message.content_type == 'application/text' # assert message.correlation_id == 'cid' # assert message.message_id == 'mid' # assert message.to == 'to' # assert message.reply_to == 'reply_to' # assert message.time_to_live == timedelta(seconds=60) if message.subject == "1st": message_1st_received_cnt += 1 receiver.complete_message(message) message.subject = "2nd" sender.send_messages(message) # resending received message elif message.subject == "2nd": message_2nd_received_cnt += 1 receiver.complete_message(message) assert message_1st_received_cnt == 20 and message_2nd_received_cnt == 20 # Network/server might be unstable making flow control ineffective in the leading rounds of connection iteration assert receive_counter < 10 # Dynamic link credit issuing come info effect @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_receiver_alive_after_timeout( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message = ServiceBusMessage("0") message_1 = ServiceBusMessage("1") sender.send_messages([message, message_1]) messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: for message in receiver: messages.append(message) break for message in receiver: messages.append(message) for message in messages: receiver.complete_message(message) assert len(messages) == 2 assert str(messages[0]) == "0" assert str(messages[1]) == "1" message_2 = ServiceBusMessage("2") message_3 = ServiceBusMessage("3") sender.send_messages([message_2, message_3]) for message in receiver: messages.append(message) for message in receiver: messages.append(message) assert len(messages) == 4 assert str(messages[2]) == "2" assert str(messages[3]) == "3" for message in messages[2:]: receiver.complete_message(message) messages = receiver.receive_messages() assert not messages @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT5M" ) @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_receive_keep_conn_alive( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) with sender, receiver: sender.send_messages([ServiceBusMessage("message1"), ServiceBusMessage("message2")]) messages = [] for message in receiver: messages.append(message) receiver_handler = receiver._handler assert len(messages) == 2 time.sleep(4 * 60 + 5) # 240s is the service defined connection idle timeout receiver.renew_message_lock(messages[0]) # check mgmt link operation receiver.complete_message(messages[0]) receiver.complete_message(messages[1]) # check receiver link operation time.sleep(60) # sleep another one minute to ensure we pass the lock_duration time messages = [] for message in receiver: messages.append(message) assert len(messages) == 0 # make sure messages are removed from the queue assert receiver_handler == receiver._handler # make sure no reconnection happened @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_receiver_sender_resume_after_link_timeout( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message = ServiceBusMessage("0") sender.send_messages(message) time.sleep(60 * 5) message_1 = ServiceBusMessage("1") sender.send_messages(message_1) messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: for message in receiver: messages.append(message) assert len(messages) == 2 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_send_twice(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs): if uamqp: transport_type = uamqp.constants.TransportType.AmqpOverWebsocket else: transport_type = TransportType.AmqpOverWebsocket fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, transport_type=transport_type, uamqp_transport=uamqp_transport, ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message = ServiceBusMessage("ServiceBusMessage") message2 = ServiceBusMessage("Message2") # first test batch message resending. batch_message = sender.create_message_batch() batch_message._from_list([message, message2]) # pylint: disable=protected-access sender.send_messages(batch_message) sender.send_messages(batch_message) messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=20) as receiver: for message in receiver: messages.append(message) receiver.complete_message(message) assert len(messages) == 4 # then normal message resending sender.send_messages(message) sender.send_messages(message) expected_count = 2 if not uamqp_transport: pickled_recvd = pickle.loads(pickle.dumps(messages[0])) sender.send_messages(pickled_recvd) expected_count = 3 messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=20) as receiver: for message in receiver: messages.append(message) receiver.complete_message(message) assert len(messages) == expected_count @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_receiver_invalid_mode( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with pytest.raises(ValueError): with sb_client.get_queue_receiver(servicebus_queue.name, receive_mode=2) as receiver: raise Exception("Should not get here, should fail fast.") @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_receiver_invalid_autolockrenew_mode( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with pytest.raises(ValueError): with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, auto_lock_renewer=AutoLockRenewer(), ) as receiver: raise Exception( "Should not get here, should fail fast because RECEIVE_AND_DELETE messages cannot be autorenewed." ) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_message_inner_amqp_properties( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): message = ServiceBusMessage("body") message.raw_amqp_message.properties.subject = "subject" message.raw_amqp_message.application_properties = {b"application_properties": 1} message.raw_amqp_message.annotations = {b"annotations": 2} message.raw_amqp_message.delivery_annotations = {b"delivery_annotations": 3} message.raw_amqp_message.header.priority = 5 message.raw_amqp_message.footer = {b"footer": 6} fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: sender.send_messages(message) with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: message = receiver.receive_messages()[0] assert ( message.raw_amqp_message.application_properties != None and message.raw_amqp_message.annotations != None and message.raw_amqp_message.delivery_annotations != None and message.raw_amqp_message.footer != None and message.raw_amqp_message.properties != None and message.raw_amqp_message.header != None ) assert message.raw_amqp_message.properties.subject == b"subject" assert message.raw_amqp_message.application_properties[b"application_properties"] == 1 assert message.raw_amqp_message.annotations[b"annotations"] == 2 assert message.raw_amqp_message.delivery_annotations[b"delivery_annotations"] == 3 assert message.raw_amqp_message.header.priority == 5 assert message.raw_amqp_message.footer[b"footer"] == 6 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer(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_timeout(self, uamqp_transport, *, servicebus_namespace=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 fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, 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(fully_qualified_namespace, credential, 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( fully_qualified_namespace, credential, 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)) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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 ): # Amqp fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient(fully_qualified_namespace, credential, 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 fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, 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( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: messages = receiver.receive_messages(max_message_count=5, max_wait_time=5) for message in messages: 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 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer(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_mgmt_operation_timeout( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): if uamqp_transport: def hack_mgmt_execute(self, operation, op_type, message, timeout=0): start_time = self._counter.get_current_ms() operation_id = str(uuid.uuid4()) self._responses[operation_id] = None time.sleep(6) # sleep until timeout while not self._responses[operation_id] and not self.mgmt_error: if timeout > 0: now = self._counter.get_current_ms() if (now - start_time) >= timeout: raise uamqp.compat.TimeoutException( "Failed to receive mgmt response in {}ms".format(timeout) ) self.connection.work() if self.mgmt_error: raise self.mgmt_error response = self._responses.pop(operation_id) return response original_execute_method = uamqp.mgmt_operation.MgmtOperation.execute # hack the mgmt method on the class, not on an instance, so it needs reset else: def hack_mgmt_execute(self, message, operation=None, operation_type=None, timeout=0): start_time = time.time() operation_id = str(uuid.uuid4()) self._responses[operation_id] = None time.sleep(6) # sleep until timeout while not self._responses[operation_id] and not self._mgmt_error: if timeout and timeout > 0: now = time.time() if (now - start_time) >= timeout: raise TimeoutError("Failed to receive mgmt response in {}ms".format(timeout)) self._connection.listen() if self._mgmt_error: self._responses.pop(operation_id) raise self._mgmt_error response = self._responses.pop(operation_id) return response original_execute_method = management_operation.ManagementOperation.execute # hack the mgmt method on the class, not on an instance, so it needs reset try: if uamqp_transport: uamqp.mgmt_operation.MgmtOperation.execute = hack_mgmt_execute else: management_operation.ManagementOperation.execute = hack_mgmt_execute fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: with pytest.raises(OperationTimeoutError): scheduled_time_utc = utc_now() + timedelta(seconds=30) sender.schedule_messages( ServiceBusMessage("ServiceBusMessage to be scheduled"), scheduled_time_utc, timeout=5 ) finally: # must reset the mgmt execute method, otherwise other test cases would use the hacked execute method, leading to timeout error if uamqp_transport: uamqp.mgmt_operation.MgmtOperation.execute = original_execute_method else: management_operation.ManagementOperation.execute = original_execute_method @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer(name_prefix="servicebustest", lock_duration="PT5S") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_operation_negative( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): if uamqp_transport: def _hack_amqp_message_complete(cls): raise RuntimeError() def _hack_amqp_mgmt_request(cls, message, operation, op_type=None, node=None, callback=None, **kwargs): raise uamqp.errors.AMQPConnectionError() def _hack_sb_receiver_settle_message( self, message, settle_operation, dead_letter_reason=None, dead_letter_error_description=None ): raise uamqp.errors.AMQPError() else: def _hack_amqp_message_complete(cls, _, _unused, settlement): if settlement == "completed": raise RuntimeError() def _hack_amqp_mgmt_request(cls, message, operation, op_type=None, node=None, callback=None, **kwargs): raise error.AMQPConnectionError(error.ErrorCondition.ConnectionCloseForced) def _hack_sb_receiver_settle_message( self, message, settle_operation, dead_letter_reason=None, dead_letter_error_description=None ): raise error.AMQPException(error.ErrorCondition.ClientError) fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) if not uamqp_transport: original_settlement = client.ReceiveClient.settle_messages try: with sender, receiver: # negative settlement via receiver link sender.send_messages(ServiceBusMessage("body"), timeout=10) message = receiver.receive_messages()[0] if uamqp_transport: message._message.accept = types.MethodType(_hack_amqp_message_complete, message._message) else: client.ReceiveClient.settle_messages = types.MethodType( _hack_amqp_message_complete, receiver._handler ) receiver.complete_message(message) # settle via mgmt link if uamqp_transport: origin_amqp_client_mgmt_request_method = uamqp.AMQPClient.mgmt_request try: uamqp.AMQPClient.mgmt_request = _hack_amqp_mgmt_request with pytest.raises(ServiceBusConnectionError): receiver.peek_messages() finally: uamqp.AMQPClient.mgmt_request = origin_amqp_client_mgmt_request_method else: origin_amqp_client_mgmt_request_method = client.AMQPClient.mgmt_request try: client.AMQPClient.mgmt_request = _hack_amqp_mgmt_request with pytest.raises(ServiceBusConnectionError): receiver.peek_messages() finally: client.AMQPClient.mgmt_request = origin_amqp_client_mgmt_request_method sender.send_messages(ServiceBusMessage("body"), timeout=10) message = receiver.receive_messages()[0] origin_sb_receiver_settle_message_method = receiver._settle_message receiver._settle_message = types.MethodType(_hack_sb_receiver_settle_message, receiver) with pytest.raises(ServiceBusError): receiver.complete_message(message) receiver._settle_message = origin_sb_receiver_settle_message_method message = receiver.receive_messages(max_wait_time=6)[0] receiver.complete_message(message) finally: if not uamqp_transport: client.ReceiveClient.settle_messages = original_settlement @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_send_message_no_body(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() sb_client = ServiceBusClient(fully_qualified_namespace, credential, uamqp_transport=uamqp_transport) with sb_client.get_queue_sender(servicebus_queue.name) as sender: sender.send_messages(ServiceBusMessage(body=None)) with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) as receiver: message = receiver.next() assert message.body is None receiver.complete_message(message) def test_send_message_alternate_body_types(self, **kwargs): with pytest.raises(TypeError): message = ServiceBusMessage(body=["1", "2"]) with pytest.raises(TypeError): message = ServiceBusMessage(body=[None, None]) with pytest.raises(TypeError): message = ServiceBusMessage(body=["1", None]) with pytest.raises(TypeError): message = ServiceBusMessage(body=[Exception()]) with pytest.raises(TypeError): message = ServiceBusMessage(body={}) with pytest.raises(TypeError): message = ServiceBusMessage(body=Exception()) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @CachedServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_by_servicebus_client_enum_case_sensitivity( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): # Note: This test is currently intended to enforce case-sensitivity. If we eventually upgrade to the Fancy Enums being used with new autorest, # we may want to tweak this. fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE.value, max_wait_time=5 ) as receiver: pass with pytest.raises(ValueError): with sb_client.get_queue_receiver( servicebus_queue.name, receive_mode=str.upper(ServiceBusReceiveMode.RECEIVE_AND_DELETE.value), max_wait_time=5, ) as receiver: raise Exception("Should not get here, should be case sensitive.") with sb_client.get_queue_receiver( servicebus_queue.name, sub_queue=ServiceBusSubQueue.DEAD_LETTER.value, max_wait_time=5 ) as receiver: pass with pytest.raises(ValueError): with sb_client.get_queue_receiver( servicebus_queue.name, sub_queue=str.upper(ServiceBusSubQueue.DEAD_LETTER.value), max_wait_time=5 ) as receiver: raise Exception("Should not get here, should be case sensitive.") @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_send_dict_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message_dict = {"body": "Message"} message2_dict = {"body": "Message2"} list_message_dicts = [message_dict, message2_dict] # send single dict sender.send_messages(message_dict) # send list of dicts sender.send_messages(list_message_dicts) # create and send BatchMessage with dicts batch_message = sender.create_message_batch() batch_message._from_list(list_message_dicts) # pylint: disable=protected-access batch_message.add_message(message_dict) sender.send_messages(batch_message) received_messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: for message in receiver: received_messages.append(message) assert len(received_messages) == 6 batch_message = sender.create_message_batch(max_size_in_bytes=73) for _ in range(2): try: batch_message.add_message(message_dict) except ValueError: break sender.send_messages(batch_message) received_messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: for message in receiver: received_messages.append(message) assert len(received_messages) == 1 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_send_mapping_messages( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): class MappingMessage(DictMixin): def __init__(self, content): self.body = content self.message_id = "foo" class BadMappingMessage(DictMixin): def __init__(self): self.message_id = "foo" fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message_dict = MappingMessage("Message") message2_dict = MappingMessage("Message2") message3_dict = BadMappingMessage() list_message_dicts = [message_dict, message2_dict] # send single dict sender.send_messages(message_dict) # send list of dicts sender.send_messages(list_message_dicts) # send bad dict with pytest.raises(TypeError): sender.send_messages(message3_dict) # create and send BatchMessage with dicts batch_message = sender.create_message_batch() batch_message._from_list(list_message_dicts) # pylint: disable=protected-access batch_message.add_message(message_dict) sender.send_messages(batch_message) received_messages = [] with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver: for message in receiver: received_messages.append(message) assert len(received_messages) == 6 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer(name_prefix="servicebustest") @pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids) @ArgPasser() def test_queue_send_dict_messages_error_badly_formatted_dicts( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message_dict = {"bad_key": "Message"} message2_dict = {"bad_key": "Message2"} list_message_dicts = [message_dict, message2_dict] # send single dict with pytest.raises(TypeError): sender.send_messages(message_dict) # send list of dicts with pytest.raises(TypeError): sender.send_messages(list_message_dicts) # create and send BatchMessage with dicts batch_message = sender.create_message_batch() with pytest.raises(TypeError): batch_message._from_list(list_message_dicts) # pylint: disable=protected-access @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_dict_messages_scheduled( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: content = "Test scheduled message" message_id = uuid.uuid4() message_id2 = uuid.uuid4() scheduled_enqueue_time = (utc_now() + timedelta(minutes=0.05)).replace(microsecond=0) message_dict = {"message_id": message_id, "body": content} message2_dict = {"message_id": message_id2, "body": content} list_message_dicts = [message_dict, message2_dict] # send single dict with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: tokens = sender.schedule_messages(message_dict, scheduled_enqueue_time) assert len(tokens) == 1 messages = receiver.receive_messages(max_wait_time=20) if messages: try: data = str(messages[0]) assert data == content assert messages[0].message_id == message_id assert messages[0].scheduled_enqueue_time_utc == scheduled_enqueue_time assert messages[0].scheduled_enqueue_time_utc <= messages[0].enqueued_time_utc.replace( microsecond=0 ) assert len(messages) == 1 finally: for m in messages: receiver.complete_message(m) else: raise Exception("Failed to receive schdeduled message.") # send list of dicts with sb_client.get_queue_receiver(servicebus_queue.name, prefetch_count=20) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: tokens = sender.schedule_messages(list_message_dicts, scheduled_enqueue_time) assert len(tokens) == 2 messages = receiver.receive_messages(max_wait_time=20) messages.extend(receiver.receive_messages(max_wait_time=5)) if messages: try: data = str(messages[0]) print(messages) assert data == content assert messages[0].message_id == message_id assert messages[0].scheduled_enqueue_time_utc == scheduled_enqueue_time assert messages[0].scheduled_enqueue_time_utc <= messages[0].enqueued_time_utc.replace( microsecond=0 ) assert len(messages) == 2 finally: for m in messages: receiver.complete_message(m) else: raise Exception("Failed to receive schdeduled message.") @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_dict_messages_scheduled_error_badly_formatted_dicts( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: content = "Test scheduled message" message_id = uuid.uuid4() message_id2 = uuid.uuid4() scheduled_enqueue_time = (utc_now() + timedelta(minutes=0.1)).replace(microsecond=0) with sb_client.get_queue_receiver(servicebus_queue.name) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: message_dict = {"message_id": message_id, "bad_key": content} message2_dict = {"message_id": message_id2, "bad_key": content} list_message_dicts = [message_dict, message2_dict] with pytest.raises(TypeError): sender.schedule_messages(message_dict, scheduled_enqueue_time) with pytest.raises(TypeError): sender.schedule_messages(list_message_dicts, scheduled_enqueue_time) @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_receive_iterator_resume_after_link_detach( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): def hack_iter_next_mock_error(self, wait_time=None): try: self._receive_context.set() self._open() # when trying to receive the second message (execution_times is 1), raising LinkDetach error to mock 10 mins idle timeout if self.execution_times == 1: # TODO: update uamqp errors to pyamqp if uamqp_transport: from uamqp.errors import LinkDetach from uamqp.constants import ErrorCodes error = LinkDetach error_condition = ErrorCodes else: from azure.servicebus._pyamqp.error import ErrorCondition, AMQPLinkError error = AMQPLinkError error_condition = ErrorCondition self.execution_times += 1 self.error_raised = True raise error(error_condition.LinkDetachForced) else: self.execution_times += 1 if not self._message_iter: if uamqp_transport: self._message_iter = self._handler.receive_messages_iter() else: self._message_iter = self._handler.receive_messages_iter(timeout=wait_time) amqp_message = next(self._message_iter) message = self._build_received_message(amqp_message) return message finally: self._receive_context.clear() fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: with sb_client.get_queue_sender(servicebus_queue.name) as sender: sender.send_messages( [ServiceBusMessage("test1"), ServiceBusMessage("test2"), ServiceBusMessage("test3")] ) with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) as receiver: receiver.execution_times = 0 receiver.error_raised = False receiver._iter_next = types.MethodType(hack_iter_next_mock_error, receiver) res = [] for msg in receiver: receiver.complete_message(msg) res.append(msg) assert len(res) == 3 assert receiver.error_raised assert receiver.execution_times >= 4 # at least 1 failure and 3 successful receiving iterator @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_amqp_annotated_message( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sequence_body = [b"message", 123.456, True] footer = {"footer_key": "footer_value"} prop = {"subject": "sequence"} seq_app_prop = {"body_type": "sequence"} sequence_message = AmqpAnnotatedMessage( sequence_body=sequence_body, footer=footer, properties=prop, application_properties=seq_app_prop ) value_body = {b"key": [-123, b"data", False]} header = {"priority": 10} anno = {"ann_key": "ann_value"} value_app_prop = {"body_type": "value"} value_message = AmqpAnnotatedMessage( value_body=value_body, header=header, annotations=anno, application_properties=value_app_prop ) data_body = [b"aa", b"bb", b"cc"] data_app_prop = {"body_type": "data"} del_anno = {"delann_key": "delann_value"} data_message = AmqpAnnotatedMessage( data_body=data_body, delivery_annotations=del_anno, application_properties=data_app_prop ) with pytest.raises(ValueError): AmqpAnnotatedMessage(data_body=data_body, value_body=value_body) with pytest.raises(ValueError): AmqpAnnotatedMessage() content = "normalmessage" dict_message = {"body": content} sb_message = ServiceBusMessage(body=content) message_with_ttl = AmqpAnnotatedMessage(data_body=data_body, header=AmqpMessageHeader(time_to_live=60000)) if uamqp_transport: amqp_transport = UamqpTransport else: amqp_transport = PyamqpTransport amqp_with_ttl = amqp_transport.to_outgoing_amqp_message(message_with_ttl) assert ( amqp_with_ttl.properties.absolute_expiry_time == amqp_with_ttl.properties.creation_time + amqp_with_ttl.header.ttl ) recv_data_msg = recv_sequence_msg = recv_value_msg = normal_msg = 0 with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=10) as receiver: with sb_client.get_queue_sender(servicebus_queue.name) as sender: batch = sender.create_message_batch() batch.add_message(data_message) batch.add_message(value_message) batch.add_message(sequence_message) batch.add_message(dict_message) batch.add_message(sb_message) sender.send_messages(batch) sender.send_messages([data_message, value_message, sequence_message, dict_message, sb_message]) sender.send_messages(data_message) sender.send_messages(value_message) sender.send_messages(sequence_message) for message in receiver: raw_amqp_message = message.raw_amqp_message if raw_amqp_message.body_type == AmqpMessageBodyType.DATA: if ( raw_amqp_message.application_properties and raw_amqp_message.application_properties.get(b"body_type") == b"data" ): body = [data for data in raw_amqp_message.body] assert data_body == body assert raw_amqp_message.delivery_annotations[b"delann_key"] == b"delann_value" assert raw_amqp_message.application_properties[b"body_type"] == b"data" recv_data_msg += 1 else: assert str(message) == content normal_msg += 1 elif raw_amqp_message.body_type == AmqpMessageBodyType.SEQUENCE: body = [sequence for sequence in raw_amqp_message.body] assert [sequence_body] == body assert raw_amqp_message.footer[b"footer_key"] == b"footer_value" assert raw_amqp_message.properties.subject == b"sequence" assert raw_amqp_message.application_properties[b"body_type"] == b"sequence" recv_sequence_msg += 1 elif raw_amqp_message.body_type == AmqpMessageBodyType.VALUE: assert raw_amqp_message.body == value_body assert raw_amqp_message.header.priority == 10 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) assert recv_sequence_msg == 3 assert recv_data_msg == 3 assert recv_value_msg == 3 assert normal_msg == 4 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_state_scheduled(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient(fully_qualified_namespace, credential, uamqp_transport=uamqp_transport) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) for i in range(10): message = ServiceBusMessage("message no. {}".format(i)) scheduled_time_utc = datetime.utcnow() + timedelta(seconds=30) sequence_number = sender.schedule_messages(message, scheduled_time_utc) receiver = sb_client.get_queue_receiver(servicebus_queue.name) with receiver: for msg in receiver.peek_messages(): assert msg.state == ServiceBusMessageState.SCHEDULED @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_state_deferred(self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient(fully_qualified_namespace, credential, uamqp_transport=uamqp_transport) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) for i in range(10): message = ServiceBusMessage("message no. {}".format(i)) sender.send_messages(message) receiver = sb_client.get_queue_receiver(servicebus_queue.name) deferred_messages = [] with receiver: received_msgs = receiver.receive_messages() for message in received_msgs: assert message.state == ServiceBusMessageState.ACTIVE deferred_messages.append(message.sequence_number) receiver.defer_message(message) if deferred_messages: received_deferred_msg = receiver.receive_deferred_messages(sequence_numbers=deferred_messages) for message in received_deferred_msg: assert message.state == ServiceBusMessageState.DEFERRED @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @ServiceBusQueuePreparer( name_prefix="servicebustest", dead_lettering_on_message_expiration=True, lock_duration="PT10S" ) @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 ): 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 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_queue_client_peek_auto_increment( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) with sender: messages = [] for i in range(3): message = ServiceBusMessage("Handler message no. {}".format(i), application_properties={"index": i}) messages.append(message) sender.send_messages(messages) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) with receiver: peek_message = receiver.peek_messages() assert peek_message[0].application_properties[b"index"] == 0 assert peek_message[0].sequence_number == 1 peek_message = receiver.peek_messages() assert peek_message[0].application_properties[b"index"] == 1 assert peek_message[0].sequence_number == 2 peek_message = receiver.peek_messages() assert peek_message[0].application_properties[b"index"] == 2 assert peek_message[0].sequence_number == 3 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_queue_client_peek_auto_increment_multiple( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) with sender: messages = [] for i in range(4): message = ServiceBusMessage("Handler message no. {}".format(i), application_properties={"index": i}) messages.append(message) sender.send_messages(messages) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) with receiver: peek_message = receiver.peek_messages(max_message_count=2) assert len(peek_message) == 2 assert peek_message[0].application_properties[b"index"] == 0 assert peek_message[0].sequence_number == 1 assert peek_message[1].application_properties[b"index"] == 1 assert peek_message[1].sequence_number == 2 peek_message = receiver.peek_messages(max_message_count=2) assert len(peek_message) == 2 assert peek_message[0].application_properties[b"index"] == 2 assert peek_message[0].sequence_number == 3 assert peek_message[1].application_properties[b"index"] == 3 assert peek_message[1].sequence_number == 4 @pytest.mark.liveTest @pytest.mark.live_test_only @CachedServiceBusResourceGroupPreparer(name_prefix="servicebustest") @CachedServiceBusNamespacePreparer(name_prefix="servicebustest") @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_queue_client_peek_and_receive( self, uamqp_transport, *, servicebus_namespace=None, servicebus_queue=None, **kwargs ): fully_qualified_namespace = f"{servicebus_namespace.name}{SERVICEBUS_ENDPOINT_SUFFIX}" credential = get_credential() with ServiceBusClient( fully_qualified_namespace, credential, logging_enable=False, uamqp_transport=uamqp_transport ) as sb_client: sender = sb_client.get_queue_sender(servicebus_queue.name) with sender: messages = [] for i in range(4): message = ServiceBusMessage("Handler message no. {}".format(i), application_properties={"index": i}) messages.append(message) sender.send_messages(messages) receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) with receiver: receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) peeked_messages = receiver.peek_messages(max_message_count=2) messages = receiver.receive_messages(max_message_count=3) last_received_sequnece_number = messages[-1].sequence_number for message in messages: receiver.complete_message(message) peeked_messages = receiver.peek_messages(max_message_count=2) assert peeked_messages[0].sequence_number == last_received_sequnece_number + 1