import os
from decimal import Decimal
import pytest

try:
    import uamqp
    from azure.servicebus._transport._uamqp_transport import UamqpTransport
except (ModuleNotFoundError, ImportError):
    uamqp = None
from datetime import datetime, timedelta, timezone
from azure.servicebus import (
    ServiceBusClient,
    ServiceBusMessage,
    ServiceBusReceivedMessage,
    ServiceBusMessageState,
    ServiceBusReceiveMode,
    ServiceBusMessageBatch,
)
from azure.servicebus._common.constants import (
    _X_OPT_PARTITION_KEY,
    _X_OPT_VIA_PARTITION_KEY,
    _X_OPT_SCHEDULED_ENQUEUE_TIME,
    _X_OPT_ENQUEUED_TIME,
)
from azure.servicebus._common.utils import (
    CE_ZERO_SECONDS,
)
from azure.servicebus.amqp import AmqpAnnotatedMessage, AmqpMessageBodyType, AmqpMessageProperties, AmqpMessageHeader
from azure.servicebus._pyamqp.message import Message
from azure.servicebus._pyamqp._message_backcompat import LegacyBatchMessage
from azure.servicebus._transport._pyamqp_transport import PyamqpTransport

from devtools_testutils import AzureMgmtRecordedTestCase
from servicebus_preparer import (
    CachedServiceBusNamespacePreparer,
    ServiceBusQueuePreparer,
    CachedServiceBusResourceGroupPreparer,
)

from utilities import uamqp_transport as get_uamqp_transport, ArgPasser

uamqp_transport_params, uamqp_transport_ids = get_uamqp_transport()


def test_servicebus_message_repr():
    message = ServiceBusMessage("hello")
    assert "application_properties=None, session_id=None," in message.__repr__()
    assert (
        "content_type=None, correlation_id=None, to=None, reply_to=None, reply_to_session_id=None, subject=None, time_to_live=None, partition_key=None, scheduled_enqueue_time_utc"
        in message.__repr__()
    )


def test_servicebus_message_repr_with_props():
    message = ServiceBusMessage(
        body="hello",
        application_properties={"prop": "test"},
        session_id="id_session",
        message_id="id_message",
        scheduled_enqueue_time_utc=datetime.now(),
        time_to_live=timedelta(seconds=30),
        content_type="content type",
        correlation_id="correlation",
        subject="github",
        partition_key="id_session",
        to="forward to",
        reply_to="reply to",
        reply_to_session_id="reply to session",
    )
    assert "application_properties={'prop': 'test'}, session_id=id_session," in message.__repr__()
    assert (
        "content_type=content type, correlation_id=correlation, to=forward to, reply_to=reply to, reply_to_session_id=reply to session, subject=github, time_to_live=0:00:30, partition_key=id_session, scheduled_enqueue_time_utc"
        in message.__repr__()
    )

def test_servicebus_message_min_timestamp():
    received_message = Message(
        data=[b"data"],
        message_annotations={
            _X_OPT_ENQUEUED_TIME: CE_ZERO_SECONDS*1000,
        },
    )
    received_message = ServiceBusReceivedMessage(received_message, receiver=None)
    assert received_message.enqueued_time_utc == datetime.min.replace(tzinfo=timezone.utc)

@pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
def test_servicebus_received_message_repr(uamqp_transport):
    my_frame = [0, 0, 0]
    if uamqp_transport:
        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: 123424566,
                _X_OPT_ENQUEUED_TIME: CE_ZERO_SECONDS * 1000,
            },
            properties={},
        )
    else:
        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: 123424566,
            },
            properties={},
        )
    received_message = ServiceBusReceivedMessage(received_message, receiver=None, frame=my_frame)
    repr_str = received_message.__repr__()
    assert "application_properties=None, session_id=None" in repr_str
    assert (
        "content_type=None, correlation_id=None, to=None, reply_to=None, reply_to_session_id=None, subject=None,"
        in repr_str
    )
    assert "partition_key=r_key, scheduled_enqueue_time_utc" in repr_str


@pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
def test_servicebus_received_state(uamqp_transport):
    my_frame = [0, 0, 0]
    if uamqp_transport:
        amqp_received_message = uamqp.message.Message(
            body=[b"data"],
            annotations={b"x-opt-message-state": 3},
        )
    else:
        amqp_received_message = Message(
            data=[b"data"],
            message_annotations={b"x-opt-message-state": 3},
        )
    received_message = ServiceBusReceivedMessage(amqp_received_message, receiver=None, frame=my_frame)
    assert received_message.state == 3

    if uamqp_transport:
        amqp_received_message = uamqp.message.Message(
            body=[b"data"], annotations={b"x-opt-message-state": 1}, properties={}
        )
    else:
        amqp_received_message = Message(data=[b"data"], message_annotations={b"x-opt-message-state": 1}, properties={})
    received_message = ServiceBusReceivedMessage(amqp_received_message, receiver=None)
    assert received_message.state == ServiceBusMessageState.DEFERRED

    if uamqp_transport:
        amqp_received_message = uamqp.message.Message(body=[b"data"], annotations={}, properties={})
    else:
        amqp_received_message = Message(data=[b"data"], message_annotations={}, properties={})
    received_message = ServiceBusReceivedMessage(amqp_received_message, receiver=None)
    assert received_message.state == ServiceBusMessageState.ACTIVE

    if uamqp_transport:
        amqp_received_message = uamqp.message.Message(body=[b"data"], properties={})
    else:
        amqp_received_message = Message(data=[b"data"], properties={})
    received_message = ServiceBusReceivedMessage(amqp_received_message, receiver=None)
    assert received_message.state == ServiceBusMessageState.ACTIVE

    if uamqp_transport:
        amqp_received_message = uamqp.message.Message(
            body=[b"data"], annotations={b"x-opt-message-state": 0}, properties={}
        )
    else:
        amqp_received_message = Message(data=[b"data"], message_annotations={b"x-opt-message-state": 0}, properties={})
    received_message = ServiceBusReceivedMessage(amqp_received_message, receiver=None)
    assert received_message.state == ServiceBusMessageState.ACTIVE


@pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
def test_servicebus_received_message_repr_with_props(uamqp_transport):
    my_frame = [0, 0, 0]
    properties = AmqpMessageProperties(
        message_id="id_message",
        absolute_expiry_time=100,
        content_type="content type",
        correlation_id="correlation",
        subject="github",
        group_id="id_session",
        reply_to="reply to",
        reply_to_group_id="reply to group",
    )
    message_annotations = {
        _X_OPT_PARTITION_KEY: b"r_key",
        _X_OPT_VIA_PARTITION_KEY: b"r_via_key",
        _X_OPT_SCHEDULED_ENQUEUE_TIME: 123424566,
    }
    data = [b"data"]
    if uamqp_transport:
        amqp_received_message = uamqp.message.Message(body=data, annotations=message_annotations, properties=properties)
    else:
        amqp_received_message = Message(data=data, message_annotations=message_annotations, properties=properties)
    received_message = ServiceBusReceivedMessage(message=amqp_received_message, receiver=None, frame=my_frame)
    assert "application_properties=None, session_id=id_session" in received_message.__repr__()
    assert (
        "content_type=content type, correlation_id=correlation, to=None, reply_to=reply to, reply_to_session_id=reply to group, subject=github"
        in received_message.__repr__()
    )
    assert "partition_key=r_key, scheduled_enqueue_time_utc" in received_message.__repr__()


@pytest.mark.parametrize("uamqp_transport", uamqp_transport_params, ids=uamqp_transport_ids)
def test_servicebus_message_batch(uamqp_transport):
    if uamqp_transport:
        amqp_transport = UamqpTransport
    else:
        amqp_transport = PyamqpTransport

    batch = ServiceBusMessageBatch(max_size_in_bytes=240, partition_key="par", amqp_transport=amqp_transport)
    batch.add_message(
        ServiceBusMessage(
            "A",
            application_properties={b"val1": b"a", "val2": "b"},
            session_id="session_id",
            message_id="message_id",
            scheduled_enqueue_time_utc=datetime.now(),
            time_to_live=timedelta(seconds=60),
            content_type="content_type",
            correlation_id="cid",
            subject="sub",
            partition_key="session_id",
            to="to",
            reply_to="reply_to",
            reply_to_session_id="reply_to_session_id",
        )
    )
    assert str(batch) == "ServiceBusMessageBatch(max_size_in_bytes=240, message_count=1)"
    assert repr(batch) == "ServiceBusMessageBatch(max_size_in_bytes=240, message_count=1)"

    assert batch.size_in_bytes == 238 and len(batch) == 1

    with pytest.raises(ValueError):
        batch.add_message(ServiceBusMessage("A"))

    assert batch.message


def test_amqp_message():
    sb_message = ServiceBusMessage(body=None)
    assert sb_message.body_type == AmqpMessageBodyType.VALUE
    assert not sb_message.body

    amqp_annotated_message = AmqpAnnotatedMessage(data_body=b"data")
    assert amqp_annotated_message.body_type == AmqpMessageBodyType.DATA
    body = [data for data in amqp_annotated_message.body]
    assert len(body) == 1
    assert body[0] == b"data"

    amqp_annotated_message = AmqpAnnotatedMessage(value_body={b"key": b"value"})
    assert amqp_annotated_message.body_type == AmqpMessageBodyType.VALUE
    assert amqp_annotated_message.body == {b"key": b"value"}

    amqp_annotated_message = AmqpAnnotatedMessage(sequence_body=[1, 2, 3])
    body = [sequence for sequence in amqp_annotated_message.body]
    assert amqp_annotated_message.body_type == AmqpMessageBodyType.SEQUENCE
    assert len(body) == 1
    assert body[0] == [1, 2, 3]

    amqp_annotated_message = AmqpAnnotatedMessage(
        value_body=None,
        header=AmqpMessageHeader(priority=1, delivery_count=1, time_to_live=1, first_acquirer=True, durable=True),
        properties=AmqpMessageProperties(
            message_id="id",
            user_id="id",
            to="to",
            subject="sub",
            correlation_id="cid",
            content_type="ctype",
            content_encoding="cencoding",
            creation_time=1,
            absolute_expiry_time=1,
            group_id="id",
            group_sequence=1,
            reply_to_group_id="id",
        ),
        footer={"key": "value"},
        delivery_annotations={"key": "value"},
        annotations={"key": "value"},
        application_properties={"key": "value"},
    )

    assert amqp_annotated_message.body_type == AmqpMessageBodyType.VALUE
    assert amqp_annotated_message.header.priority == 1
    assert amqp_annotated_message.header.delivery_count == 1
    assert amqp_annotated_message.header.time_to_live == 1
    assert amqp_annotated_message.header.first_acquirer
    assert amqp_annotated_message.header.durable

    assert amqp_annotated_message.footer == {"key": "value"}
    assert amqp_annotated_message.delivery_annotations == {"key": "value"}
    assert amqp_annotated_message.annotations == {"key": "value"}
    assert amqp_annotated_message.application_properties == {"key": "value"}

    assert amqp_annotated_message.properties.message_id == "id"
    assert amqp_annotated_message.properties.user_id == "id"
    assert amqp_annotated_message.properties.to == "to"
    assert amqp_annotated_message.properties.subject == "sub"
    assert amqp_annotated_message.properties.correlation_id == "cid"
    assert amqp_annotated_message.properties.content_type == "ctype"
    assert amqp_annotated_message.properties.content_encoding == "cencoding"
    assert amqp_annotated_message.properties.creation_time == 1
    assert amqp_annotated_message.properties.absolute_expiry_time == 1
    assert amqp_annotated_message.properties.group_id == "id"
    assert amqp_annotated_message.properties.group_sequence == 1
    assert amqp_annotated_message.properties.reply_to_group_id == "id"

    amqp_annotated_message = AmqpAnnotatedMessage(
        value_body=None,
        header={"priority": 1, "delivery_count": 1, "time_to_live": 1, "first_acquirer": True, "durable": True},
        properties={
            "message_id": "id",
            "user_id": "id",
            "to": "to",
            "subject": "sub",
            "correlation_id": "cid",
            "content_type": "ctype",
            "content_encoding": "cencoding",
            "creation_time": 1,
            "absolute_expiry_time": 1,
            "group_id": "id",
            "group_sequence": 1,
            "reply_to_group_id": "id",
        },
        footer={"key": "value"},
        delivery_annotations={"key": "value"},
        annotations={"key": "value"},
        application_properties={"key": "value"},
    )

    assert amqp_annotated_message.body_type == AmqpMessageBodyType.VALUE
    assert amqp_annotated_message.header.priority == 1
    assert amqp_annotated_message.header.delivery_count == 1
    assert amqp_annotated_message.header.time_to_live == 1
    assert amqp_annotated_message.header.first_acquirer
    assert amqp_annotated_message.header.durable

    assert amqp_annotated_message.footer == {"key": "value"}
    assert amqp_annotated_message.delivery_annotations == {"key": "value"}
    assert amqp_annotated_message.annotations == {"key": "value"}
    assert amqp_annotated_message.application_properties == {"key": "value"}

    assert amqp_annotated_message.properties.message_id == "id"
    assert amqp_annotated_message.properties.user_id == "id"
    assert amqp_annotated_message.properties.to == "to"
    assert amqp_annotated_message.properties.subject == "sub"
    assert amqp_annotated_message.properties.correlation_id == "cid"
    assert amqp_annotated_message.properties.content_type == "ctype"
    assert amqp_annotated_message.properties.content_encoding == "cencoding"
    assert amqp_annotated_message.properties.creation_time == 1
    assert amqp_annotated_message.properties.absolute_expiry_time == 1
    assert amqp_annotated_message.properties.group_id == "id"
    assert amqp_annotated_message.properties.group_sequence == 1
    assert amqp_annotated_message.properties.reply_to_group_id == "id"


def test_servicebus_message_time_to_live():
    message = ServiceBusMessage(body="hello")
    message.time_to_live = timedelta(seconds=30)
    assert message.time_to_live == timedelta(seconds=30)
    message.time_to_live = timedelta(days=1)
    assert message.time_to_live == timedelta(days=1)


class TestServiceBusMessageBackcompat(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_message_backcompat_receive_and_delete_databody(
        self, uamqp_transport, *, servicebus_namespace_connection_string, servicebus_queue, **kwargs
    ):
        queue_name = servicebus_queue.name
        outgoing_message = ServiceBusMessage(
            body="hello",
            application_properties={"prop": "test"},
            session_id="id_session",
            message_id="id_message",
            time_to_live=timedelta(seconds=30),
            content_type="content type",
            correlation_id="correlation",
            subject="github",
            partition_key="id_session",
            to="forward to",
            reply_to="reply to",
            reply_to_session_id="reply to session",
        )

        sb_client = ServiceBusClient.from_connection_string(
            servicebus_namespace_connection_string, logging_enable=True, uamqp_transport=uamqp_transport
        )
        with sb_client.get_queue_sender(queue_name) as sender:
            sender.send_messages(outgoing_message)

        # outgoing_message.message will be LegacyMessage for both uamqp and pyamqp same as in EH.
        # Previously, "empty"/useless uamqp.Message was returned, # b/c outgoing message is constructed
        # in send. So, returning LegacyMessage now should not cause issues.
        assert outgoing_message.message
        with pytest.raises(TypeError):
            outgoing_message.message.accept()
        with pytest.raises(TypeError):
            outgoing_message.message.release()
        with pytest.raises(TypeError):
            outgoing_message.message.reject()
        with pytest.raises(TypeError):
            outgoing_message.message.modify(True, True)
        try:
            assert outgoing_message.message.state == uamqp.constants.MessageState.SendComplete
        except AttributeError:  # uamqp not installed
            pass
        assert outgoing_message.message.settled
        assert outgoing_message.message.delivery_annotations is None
        assert outgoing_message.message.delivery_no is None
        assert outgoing_message.message.delivery_tag is None
        assert outgoing_message.message.on_send_complete is None
        assert outgoing_message.message.footer is None
        assert outgoing_message.message.retries >= 0
        assert outgoing_message.message.idle_time >= 0
        with pytest.raises(Exception):
            outgoing_message.message.gather()
        assert isinstance(outgoing_message.message.encode_message(), bytes)
        assert outgoing_message.message.get_message_encoded_size() == 208
        assert list(outgoing_message.message.get_data()) == [b"hello"]
        assert outgoing_message.message.application_properties == {"prop": "test"}
        assert outgoing_message.message.get_message()  # C instance.
        assert len(outgoing_message.message.annotations) == 1
        assert list(outgoing_message.message.annotations.values())[0] == "id_session"
        assert str(outgoing_message.message.header) == str(
            {"delivery_count": 0, "time_to_live": 30000, "first_acquirer": None, "durable": None, "priority": None}
        )
        assert outgoing_message.message.header.get_header_obj().delivery_count == 0
        assert outgoing_message.message.properties.message_id == b"id_message"
        assert outgoing_message.message.properties.user_id is None
        assert outgoing_message.message.properties.to == b"forward to"
        assert outgoing_message.message.properties.subject == b"github"
        assert outgoing_message.message.properties.reply_to == b"reply to"
        assert outgoing_message.message.properties.correlation_id == b"correlation"
        assert outgoing_message.message.properties.content_type == b"content type"
        assert outgoing_message.message.properties.content_encoding is None
        assert outgoing_message.message.properties.absolute_expiry_time
        assert outgoing_message.message.properties.creation_time
        assert outgoing_message.message.properties.group_id == b"id_session"
        assert outgoing_message.message.properties.group_sequence is None
        assert outgoing_message.message.properties.reply_to_group_id == b"reply to session"
        assert outgoing_message.message.properties.get_properties_obj().message_id

        with sb_client.get_queue_receiver(
            queue_name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=10
        ) as receiver:
            batch = receiver.receive_messages()
            incoming_message = batch[0]
            assert incoming_message.message
            try:
                assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedSettled
            except AttributeError:  # uamqp not installed
                pass
            assert incoming_message.message.settled
            assert incoming_message.message.delivery_annotations == {}
            assert incoming_message.message.delivery_no >= 1
            assert incoming_message.message.delivery_tag is None
            assert incoming_message.message.on_send_complete is None
            assert incoming_message.message.footer is None
            assert incoming_message.message.retries >= 0
            assert incoming_message.message.idle_time == 0
            with pytest.raises(Exception):
                incoming_message.message.gather()
            assert isinstance(incoming_message.message.encode_message(), bytes)
            # TODO: uamqp size = pyamqp size + 4?
            # uamqp bug accounts for 3 bytes:
            # - durable/first_acquirer/priority set by default in uamqp, None in pyamqp
            # - setting pyamqp values for durable/first_acquirer increases pyamqp size = 269
            if uamqp_transport:
                encoded_size = 267
            else:
                encoded_size = 263
            assert incoming_message.message.get_message_encoded_size() == encoded_size
            assert list(incoming_message.message.get_data()) == [b"hello"]
            assert incoming_message.message.application_properties == {b"prop": b"test"}
            assert incoming_message.message.get_message()  # C instance.
            assert len(incoming_message.message.annotations) == 3
            assert incoming_message.message.annotations[b"x-opt-enqueued-time"] > 0
            assert incoming_message.message.annotations[b"x-opt-sequence-number"] > 0
            assert incoming_message.message.annotations[b"x-opt-partition-key"] == b"id_session"
            if uamqp_transport:
                # uamqp bugs:
                # 1) in uamqp.get_header_obj():
                #   delivery_count should be 0, but b/c header obj is not mutable, value is not replaced.
                # 2) MessageHeader.durable/first_acquirer are being set to True always on received message
                #   These properties should not be modified by uamqp.
                assert ", 'time_to_live': 30000" in str(incoming_message.message.header)
            else:
                assert incoming_message.message.header.get_header_obj().delivery_count == 0
                assert ", 'time_to_live': 30000, 'first_acquirer': None, 'durable': None, 'priority': None}" in str(
                    incoming_message.message.header
                )
            assert incoming_message.message.properties.message_id == b"id_message"
            assert incoming_message.message.properties.user_id is None
            assert incoming_message.message.properties.to == b"forward to"
            assert incoming_message.message.properties.subject == b"github"
            assert incoming_message.message.properties.reply_to == b"reply to"
            assert incoming_message.message.properties.correlation_id == b"correlation"
            assert incoming_message.message.properties.content_type == b"content type"
            assert incoming_message.message.properties.content_encoding is None
            assert incoming_message.message.properties.absolute_expiry_time
            assert incoming_message.message.properties.creation_time
            assert incoming_message.message.properties.group_id == b"id_session"
            assert incoming_message.message.properties.group_sequence is None
            assert incoming_message.message.properties.reply_to_group_id == b"reply to session"
            assert incoming_message.message.properties.get_properties_obj().message_id
            assert not incoming_message.message.accept()
            assert not incoming_message.message.release()
            assert not incoming_message.message.reject()
            assert not incoming_message.message.modify(True, True)

    @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_message_backcompat_peek_lock_databody(
        self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs
    ):
        queue_name = servicebus_queue.name
        outgoing_message = ServiceBusMessage(
            body="hello",
            application_properties={"prop": "test"},
            session_id="id_session",
            message_id="id_message",
            time_to_live=timedelta(seconds=30),
            content_type="content type",
            correlation_id="correlation",
            subject="github",
            partition_key="id_session",
            to="forward to",
            reply_to="reply to",
            reply_to_session_id="reply to session",
        )

        sb_client = ServiceBusClient.from_connection_string(
            servicebus_namespace_connection_string, logging_enable=True, uamqp_transport=uamqp_transport
        )
        with sb_client.get_queue_sender(queue_name) as sender:
            sender.send_messages(outgoing_message)

        assert outgoing_message.message
        with pytest.raises(TypeError):
            outgoing_message.message.accept()
        with pytest.raises(TypeError):
            outgoing_message.message.release()
        with pytest.raises(TypeError):
            outgoing_message.message.reject()
        with pytest.raises(TypeError):
            outgoing_message.message.modify(True, True)
        try:
            assert outgoing_message.message.state == uamqp.constants.MessageState.SendComplete
        except AttributeError:  # uamqp not installed
            pass
        assert outgoing_message.message.settled
        assert outgoing_message.message.delivery_annotations is None
        assert outgoing_message.message.delivery_no is None
        assert outgoing_message.message.delivery_tag is None
        assert outgoing_message.message.on_send_complete is None
        assert outgoing_message.message.footer is None
        assert outgoing_message.message.retries >= 0
        assert outgoing_message.message.idle_time >= 0
        with pytest.raises(Exception):
            outgoing_message.message.gather()
        assert isinstance(outgoing_message.message.encode_message(), bytes)
        assert outgoing_message.message.get_message_encoded_size() == 208
        assert list(outgoing_message.message.get_data()) == [b"hello"]
        assert outgoing_message.message.application_properties == {"prop": "test"}
        assert outgoing_message.message.get_message()  # C instance.
        assert len(outgoing_message.message.annotations) == 1
        assert list(outgoing_message.message.annotations.values())[0] == "id_session"
        assert str(outgoing_message.message.header) == str(
            {"delivery_count": 0, "time_to_live": 30000, "first_acquirer": None, "durable": None, "priority": None}
        )
        assert outgoing_message.message.header.get_header_obj().delivery_count == 0
        assert outgoing_message.message.properties.message_id == b"id_message"
        assert outgoing_message.message.properties.user_id is None
        assert outgoing_message.message.properties.to == b"forward to"
        assert outgoing_message.message.properties.subject == b"github"
        assert outgoing_message.message.properties.reply_to == b"reply to"
        assert outgoing_message.message.properties.correlation_id == b"correlation"
        assert outgoing_message.message.properties.content_type == b"content type"
        assert outgoing_message.message.properties.content_encoding is None
        assert outgoing_message.message.properties.absolute_expiry_time
        assert outgoing_message.message.properties.creation_time
        assert outgoing_message.message.properties.group_id == b"id_session"
        assert outgoing_message.message.properties.group_sequence is None
        assert outgoing_message.message.properties.reply_to_group_id == b"reply to session"
        assert outgoing_message.message.properties.get_properties_obj().message_id

        with sb_client.get_queue_receiver(
            queue_name, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, max_wait_time=10
        ) as receiver:
            batch = receiver.receive_messages()
            incoming_message = batch[0]
            assert incoming_message.message
            try:
                assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedUnsettled
            except AttributeError:  # uamqp not installed
                pass
            assert not incoming_message.message.settled
            assert incoming_message.message.delivery_annotations[b"x-opt-lock-token"]
            assert incoming_message.message.delivery_no >= 1
            assert incoming_message.message.delivery_tag
            assert incoming_message.message.on_send_complete is None
            assert incoming_message.message.footer is None
            assert incoming_message.message.retries >= 0
            assert incoming_message.message.idle_time == 0
            with pytest.raises(Exception):
                incoming_message.message.gather()
            assert isinstance(incoming_message.message.encode_message(), bytes)

            if uamqp_transport:
                encoded_size = 334
            else:
                # uamqp bug: sets durable/first_acquirer/priority by default on incoming message
                # pyamqp = 339 if durable/first_acquirer set
                encoded_size = 333
            assert incoming_message.message.get_message_encoded_size() == encoded_size
            assert list(incoming_message.message.get_data()) == [b"hello"]
            assert incoming_message.message.application_properties == {b"prop": b"test"}
            assert incoming_message.message.get_message()  # C instance.
            assert len(incoming_message.message.annotations) == 4
            assert incoming_message.message.annotations[b"x-opt-enqueued-time"] > 0
            assert incoming_message.message.annotations[b"x-opt-sequence-number"] > 0
            assert incoming_message.message.annotations[b"x-opt-partition-key"] == b"id_session"
            assert incoming_message.message.annotations[b"x-opt-locked-until"]
            if uamqp_transport:
                # uamqp bugs:
                # 1) in uamqp.get_header_obj():
                #   delivery_count should be 0, but b/c header obj is not mutable, value is not replaced.
                # 2) MessageHeader.durable/first_acquirer are always being set to True on received message
                #   These properties should not be modified by uamqp. By default, should be None when not set.
                assert ", 'time_to_live': 30000" in str(incoming_message.message.header)
            else:
                assert incoming_message.message.header.get_header_obj().delivery_count == 0
                assert str(incoming_message.message.header) == str(
                    {
                        "delivery_count": 0,
                        "time_to_live": 30000,
                        "first_acquirer": None,
                        "durable": None,
                        "priority": None,
                    }
                )
            assert incoming_message.message.properties.message_id == b"id_message"
            assert incoming_message.message.properties.user_id is None
            assert incoming_message.message.properties.to == b"forward to"
            assert incoming_message.message.properties.subject == b"github"
            assert incoming_message.message.properties.reply_to == b"reply to"
            assert incoming_message.message.properties.correlation_id == b"correlation"
            assert incoming_message.message.properties.content_type == b"content type"
            assert incoming_message.message.properties.content_encoding is None
            assert incoming_message.message.properties.absolute_expiry_time
            assert incoming_message.message.properties.creation_time
            assert incoming_message.message.properties.group_id == b"id_session"
            assert incoming_message.message.properties.group_sequence is None
            assert incoming_message.message.properties.reply_to_group_id == b"reply to session"
            assert incoming_message.message.properties.get_properties_obj().message_id
            assert incoming_message.message.accept()
            # TODO: State isn't updated if settled correctly via the receiver.
            try:
                assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedSettled
            except AttributeError:  # uamqp not installed
                pass
            assert incoming_message.message.settled
            assert not incoming_message.message.release()
            assert not incoming_message.message.reject()
            assert not incoming_message.message.modify(True, True)

    @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_message_backcompat_receive_and_delete_valuebody(
        self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs
    ):
        queue_name = servicebus_queue.name
        outgoing_message = AmqpAnnotatedMessage(value_body={b"key": b"value"})

        with pytest.raises(AttributeError):
            outgoing_message.message

        sb_client = ServiceBusClient.from_connection_string(
            servicebus_namespace_connection_string, logging_enable=False, uamqp_transport=uamqp_transport
        )
        with sb_client.get_queue_sender(queue_name) as sender:
            sender.send_messages(outgoing_message)

        with pytest.raises(AttributeError):
            outgoing_message.message

        with sb_client.get_queue_receiver(
            queue_name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=10
        ) as receiver:
            batch = receiver.receive_messages()
            incoming_message = batch[0]
            assert incoming_message.message
            try:
                assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedSettled
            except AttributeError:  # uamqp not installed
                pass
            assert incoming_message.message.settled
            with pytest.raises(Exception):
                incoming_message.message.gather()
            assert incoming_message.message.get_data() == {b"key": b"value"}
            assert not incoming_message.message.accept()
            assert not incoming_message.message.release()
            assert not incoming_message.message.reject()
            assert not incoming_message.message.modify(True, True)

    @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_message_backcompat_peek_lock_valuebody(
        self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs
    ):
        queue_name = servicebus_queue.name
        outgoing_message = AmqpAnnotatedMessage(value_body={b"key": b"value"})

        with pytest.raises(AttributeError):
            outgoing_message.message

        sb_client = ServiceBusClient.from_connection_string(
            servicebus_namespace_connection_string, logging_enable=False, uamqp_transport=uamqp_transport
        )
        with sb_client.get_queue_sender(queue_name) as sender:
            sender.send_messages(outgoing_message)

        with pytest.raises(AttributeError):
            outgoing_message.message

        with sb_client.get_queue_receiver(
            queue_name, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, max_wait_time=10
        ) as receiver:
            batch = receiver.receive_messages()
            incoming_message = batch[0]
            assert incoming_message.message
            # assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedUnsettled
            assert not incoming_message.message.settled
            assert incoming_message.message.delivery_annotations[b"x-opt-lock-token"]
            assert incoming_message.message.delivery_no >= 1
            assert incoming_message.message.delivery_tag
            with pytest.raises(Exception):
                incoming_message.message.gather()
            assert incoming_message.message.get_data() == {b"key": b"value"}
            assert incoming_message.message.accept()
            # assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedSettled
            assert incoming_message.message.settled
            assert not incoming_message.message.release()
            assert not incoming_message.message.reject()
            assert not incoming_message.message.modify(True, True)

    @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_message_backcompat_receive_and_delete_sequencebody(
        self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs
    ):
        queue_name = servicebus_queue.name
        outgoing_message = AmqpAnnotatedMessage(sequence_body=[1, 2, 3])

        with pytest.raises(AttributeError):
            outgoing_message.message

        sb_client = ServiceBusClient.from_connection_string(
            servicebus_namespace_connection_string, logging_enable=False, uamqp_transport=uamqp_transport
        )
        with sb_client.get_queue_sender(queue_name) as sender:
            sender.send_messages(outgoing_message)

        with pytest.raises(AttributeError):
            outgoing_message.message

        with sb_client.get_queue_receiver(
            queue_name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=10
        ) as receiver:
            batch = receiver.receive_messages()
            incoming_message = batch[0]
            assert incoming_message.message
            try:
                assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedSettled
            except AttributeError:  # uamqp not installed
                pass
            assert incoming_message.message.settled
            with pytest.raises(Exception):
                incoming_message.message.gather()
            assert list(incoming_message.message.get_data()) == [[1, 2, 3]]
            assert not incoming_message.message.accept()
            assert not incoming_message.message.release()
            assert not incoming_message.message.reject()
            assert not incoming_message.message.modify(True, True)

    @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_message_batch_backcompat(
        self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs
    ):
        queue_name = servicebus_queue.name
        outgoing_message = AmqpAnnotatedMessage(sequence_body=[1, 2, 3])

        with pytest.raises(AttributeError):
            outgoing_message.message

        sb_client = ServiceBusClient.from_connection_string(
            servicebus_namespace_connection_string, logging_enable=False, uamqp_transport=uamqp_transport
        )
        with sb_client.get_queue_sender(queue_name) as sender:
            sender.send_messages(outgoing_message)

        with pytest.raises(AttributeError):
            outgoing_message.message

        with sb_client.get_queue_receiver(
            queue_name, receive_mode=ServiceBusReceiveMode.PEEK_LOCK, max_wait_time=10
        ) as receiver:
            batch = receiver.receive_messages()
            incoming_message = batch[0]
            assert incoming_message.message
            try:
                assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedUnsettled
            except AttributeError:  # uamqp not installed
                pass
            assert not incoming_message.message.settled
            assert incoming_message.message.delivery_annotations[b"x-opt-lock-token"]
            assert incoming_message.message.delivery_no >= 1
            assert incoming_message.message.delivery_tag
            with pytest.raises(Exception):
                incoming_message.message.gather()
            assert list(incoming_message.message.get_data()) == [[1, 2, 3]]
            assert incoming_message.message.accept()
            try:
                assert incoming_message.message.state == uamqp.constants.MessageState.ReceivedSettled
            except AttributeError:  # uamqp not installed
                pass
            assert incoming_message.message.settled
            assert not incoming_message.message.release()
            assert not incoming_message.message.reject()
            assert not incoming_message.message.modify(True, True)
    
    @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_message_property(
        self, uamqp_transport, *, servicebus_namespace_connection_string=None, servicebus_queue=None, **kwargs
    ):
        queue_name = servicebus_queue.name
        sb_message = ServiceBusMessage(
            body="hello",
            application_properties={
                "prop": "test",
                "intprop": 1,
                "byteprop": b"byte",
                "decimalprop": Decimal("1.2345"),
                },
        )
        sb_client = ServiceBusClient.from_connection_string(
            servicebus_namespace_connection_string, logging_enable=False, uamqp_transport=uamqp_transport
        )
        with sb_client.get_queue_sender(queue_name) as sender:
            sender.send_messages(sb_message)

        with sb_client.get_queue_receiver(
            queue_name, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE, max_wait_time=10
        ) as receiver:
            batch = receiver.receive_messages()
            incoming_message = batch[0]
            assert incoming_message.application_properties
            assert incoming_message.application_properties[b"prop"] == b"test"
            assert incoming_message.application_properties[b"intprop"] == 1
            assert incoming_message.application_properties[b"byteprop"] == b"byte"
            assert incoming_message.application_properties[b"decimalprop"] == Decimal("1.2345")