[EventHub] handle lowest/"unset" timestamp value from service (Azure#40363)

* [EventHub/ServiceBus] update datetime from timestamp to account for min AMQP timestamp sentinel val

* add sb tests

* eh test

* copilot comments

* pylint

* lint

Author: swathipil

sdk/eventhub/azure-eventhub/CHANGELOG.md

@@ -31,6 +31,8 @@
 ### Bugs Fixed
 
 - Fixed a bug where service errors were incorrectly required and expected to have info/description fields.
+- Fixed a bug where max number of messages was not being requested when receiving from the service due to an incorrect link credit calculation.
+- Fixed a bug where the lowest possible timestamp returned by the service to represent an unset time was not being parsed correctly.
 
 ## 5.14.0 (2025-02-13)
 

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/utils.py

@@ -14,9 +14,23 @@
 from ._encode import encode_payload
 
 TZ_UTC: timezone = timezone.utc
+# Number of seconds between the Unix epoch (1/1/1970) and year 1 CE.
+# This is the lowest value that can be represented by an AMQP timestamp.
+CE_ZERO_SECONDS: int = -62_135_596_800
 
 
-def utc_from_timestamp(timestamp):
+def utc_from_timestamp(timestamp: float) -> datetime.datetime:
+    """
+    :param float timestamp: Timestamp in seconds to be converted to datetime.
+    :rtype: datetime.datetime
+    :returns: A datetime object representing the timestamp in UTC.
+    """
+    # The AMQP timestamp is the number of seconds since the Unix epoch.
+    # AMQP brokers represent the lowest value as -62_135_596_800 (the
+    # number of seconds between the Unix epoch (1/1/1970) and year 1 CE) as
+    # a sentinel for a time which is not set.
+    if timestamp == CE_ZERO_SECONDS:
+        return datetime.datetime.min.replace(tzinfo=TZ_UTC)
     return datetime.datetime.fromtimestamp(timestamp, tz=TZ_UTC)
 
 

sdk/eventhub/azure-eventhub/azure/eventhub/_utils.py

@@ -22,7 +22,6 @@
     PROP_LAST_ENQUEUED_OFFSET,
 )
 
-
 if TYPE_CHECKING:
     from ._transport._base import AmqpTransport
     from ._pyamqp.message import Message as pyamqp_Message
@@ -44,11 +43,24 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-
 TZ_UTC: timezone = timezone.utc
+# Number of seconds between the Unix epoch (1/1/1970) and year 1 CE.
+# This is the lowest value that can be represented by an AMQP timestamp.
+CE_ZERO_SECONDS: int = -62_135_596_800
 
 
-def utc_from_timestamp(timestamp):
+def utc_from_timestamp(timestamp: float) -> datetime.datetime:
+    """
+    :param float timestamp: Timestamp in seconds to be converted to datetime.
+    :rtype: datetime.datetime
+    :returns: A datetime object representing the timestamp in UTC.
+    """
+    # The AMQP timestamp is the number of seconds since the Unix epoch.
+    # AMQP brokers represent the lowest value as -62_135_596_800 (the
+    # number of seconds between the Unix epoch (1/1/1970) and year 1 CE) as
+    # a sentinel for a time which is not set.
+    if timestamp == CE_ZERO_SECONDS:
+        return datetime.datetime.min.replace(tzinfo=TZ_UTC)
     return datetime.datetime.fromtimestamp(timestamp, tz=TZ_UTC)
 
 

sdk/eventhub/azure-eventhub/tests/livetest/asynctests/test_properties_async.py

@@ -100,10 +100,6 @@ async def test_get_partition_ids(auth_credentials_async, uamqp_transport):
         assert partition_ids == ["0", "1"]
 
 
-@pytest.mark.skipif(
-    sys.platform.startswith("win"),
-    reason="Large negative timestamp to datetime conversion fails on Windows with: https://bugs.python.org/issue36439",
-)
 @pytest.mark.liveTest
 @pytest.mark.asyncio
 async def test_get_partition_properties(auth_credentials_async, uamqp_transport):

sdk/eventhub/azure-eventhub/tests/livetest/synctests/test_properties.py

@@ -97,10 +97,6 @@ def test_get_partition_ids(auth_credentials, uamqp_transport):
         assert partition_ids == ["0", "1"]
 
 
-@pytest.mark.skipif(
-    sys.platform.startswith("win"),
-    reason="Large negative timestamp to datetime conversion fails on Windows with: https://bugs.python.org/issue36439",
-)
 @pytest.mark.liveTest
 def test_get_partition_properties(auth_credentials, uamqp_transport):
     fully_qualified_namespace, eventhub_name, credential = auth_credentials

sdk/eventhub/azure-eventhub/tests/unittest/test_event_data.py

@@ -18,6 +18,8 @@
     UamqpTransport = None
 from azure.eventhub._transport._pyamqp_transport import PyamqpTransport
 from azure.eventhub._pyamqp.message import Message, Properties, Header
+from azure.eventhub._utils import CE_ZERO_SECONDS
+from azure.eventhub._constants import PROP_TIMESTAMP
 from azure.eventhub.amqp import AmqpAnnotatedMessage, AmqpMessageHeader, AmqpMessageProperties
 
 from azure.eventhub import _common
@@ -101,7 +103,7 @@ def test_sys_properties(uamqp_transport):
         properties.group_sequence = 1
         properties.reply_to_group_id = "reply_to_group_id"
         message = uamqp.message.Message(properties=properties)
-        message.annotations = {_common.PROP_OFFSET: "@latest"}
+        message.annotations = {_common.PROP_OFFSET: "@latest", PROP_TIMESTAMP: CE_ZERO_SECONDS * 1000}
     else:
         properties = Properties(
             message_id="message_id",
@@ -118,7 +120,7 @@ def test_sys_properties(uamqp_transport):
             group_sequence=1,
             reply_to_group_id="reply_to_group_id",
         )
-        message_annotations = {_common.PROP_OFFSET: "@latest"}
+        message_annotations = {_common.PROP_OFFSET: "@latest", PROP_TIMESTAMP: CE_ZERO_SECONDS * 1000}
         message = Message(properties=properties, message_annotations=message_annotations)
     ed = EventData._from_message(message)  # type: EventData
 
@@ -136,6 +138,7 @@ def test_sys_properties(uamqp_transport):
     assert ed.system_properties[_common.PROP_GROUP_ID] == properties.group_id
     assert ed.system_properties[_common.PROP_GROUP_SEQUENCE] == properties.group_sequence
     assert ed.system_properties[_common.PROP_REPLY_TO_GROUP_ID] == properties.reply_to_group_id
+    assert ed.enqueued_time == datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
 
 
 def test_event_data_batch(uamqp_transport):

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Fixed a bug where service errors were incorrectly expected to have info/description fields set in all cases.
 - Fixed a bug where the type in azure.servicebus.management.AuthorizationRule was not being correctly passed to the request.
+- Fixed a bug where max number of messages was not being requested when receiving from the service due to an incorrect link credit calculation. ([#40156](https://github.com/Azure/azure-sdk-for-python/issues/40156))
 
 ## 7.14.0 (2025-02-13)
 

sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py

@@ -65,9 +65,24 @@
 
 _log = logging.getLogger(__name__)
 
+TZ_UTC: timezone = timezone.utc
+# Number of seconds between the Unix epoch (1/1/1970) and year 1 CE.
+# This is the lowest value that can be represented by an AMQP timestamp.
+CE_ZERO_SECONDS: int = -62_135_596_800
 
-def utc_from_timestamp(timestamp):
-    return datetime.datetime.fromtimestamp(timestamp, tz=timezone.utc)
+def utc_from_timestamp(timestamp: float) -> datetime.datetime:
+    """
+    :param float timestamp: Timestamp in seconds to be converted to datetime.
+    :rtype: datetime.datetime
+    :returns: A datetime object representing the timestamp in UTC.
+    """
+    # The AMQP timestamp is the number of seconds since the Unix epoch.
+    # AMQP brokers represent the lowest value as -62_135_596_800 (the
+    # number of seconds between the Unix epoch (1/1/1970) and year 1 CE) as
+    # a sentinel for a time which is not set.
+    if timestamp == CE_ZERO_SECONDS:
+        return datetime.datetime.min.replace(tzinfo=TZ_UTC)
+    return datetime.datetime.fromtimestamp(timestamp, tz=TZ_UTC)
 
 
 def utc_now():

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

@@ -14,9 +14,22 @@
 from ._encode import encode_payload
 
 TZ_UTC: timezone = timezone.utc
+# Number of seconds between the Unix epoch (1/1/1970) and year 1 CE.
+# This is the lowest value that can be represented by an AMQP timestamp.
+CE_ZERO_SECONDS: int = -62_135_596_800
 
-
-def utc_from_timestamp(timestamp):
+def utc_from_timestamp(timestamp: float) -> datetime.datetime:
+    """
+    :param float timestamp: Timestamp in seconds to be converted to datetime.
+    :rtype: datetime.datetime
+    :returns: A datetime object representing the timestamp in UTC.
+    """
+    # The AMQP timestamp is the number of seconds since the Unix epoch.
+    # AMQP brokers represent the lowest value as -62_135_596_800 (the
+    # number of seconds between the Unix epoch (1/1/1970) and year 1 CE) as
+    # a sentinel for a time which is not set.
+    if timestamp == CE_ZERO_SECONDS:
+        return datetime.datetime.min.replace(tzinfo=TZ_UTC)
     return datetime.datetime.fromtimestamp(timestamp, tz=TZ_UTC)
 
 

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

@@ -7,7 +7,7 @@
     from azure.servicebus._transport._uamqp_transport import UamqpTransport
 except (ModuleNotFoundError, ImportError):
     uamqp = None
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from azure.servicebus import (
     ServiceBusClient,
     ServiceBusMessage,
@@ -20,6 +20,10 @@
     _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
@@ -69,6 +73,15 @@ def test_servicebus_message_repr_with_props():
         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):
@@ -80,6 +93,7 @@ def test_servicebus_received_message_repr(uamqp_transport):
                 _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={},
         )