test(servicebus): cover the response half and the uamqp value type

Address review feedback.

Nothing tested the response half: every assertion covered what the client
sends, while the changelog promises a retryable OperationTimeoutError that
rests on the service returning com.microsoft:timeout in errorCondition.
Push a fake non-200 response with that condition through
mgmt_handlers.default and assert the type, for both transports, with no
live namespace.

uamqp is the one place the value type changes, since pyamqp uses a plain
dict where uamqp uses an AMQPuInt that has not previously gone into
application_properties. One test stubs the transport type to prove the
property is built from the transport's own AMQP_UINT_VALUE, and one
encodes a real message when uamqp is installed.

Also drop the redundant outer min in get_server_timeout_ms, which never
binds because the buffer is subtracted after the cap, and move the
changelog entry to Other Changes since it changes existing operations
rather than adding a feature.

api.metadata.yml carries the parserVersion bump only. That field is gated
by the consistency check and the package still had 0.3.30 while the check
generates 0.3.31. apiMdSha256 and pythonVersion are unchanged.

Refs: AB#38822503

Author: Pranjal Patel

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -6,7 +6,6 @@
 
 - Added `ServiceBusReceivedMessage.from_bytes()` classmethod to construct a `ServiceBusReceivedMessage` from raw AMQP payload bytes without requiring the deprecated `uamqp` library. ([#43979](https://github.com/Azure/azure-sdk-for-python/issues/43979))
 - Added `ServiceBusClient.list_queue_sessions()` and `ServiceBusClient.list_subscription_sessions()` (sync and async) to list session IDs for entities with active messages, with optional filtering by session-state update timestamp. The methods return an `ItemPaged[str]` (`AsyncItemPaged[str]` on the async client) so callers can iterate every session transparently or page with `by_page()`. Implements the `com.microsoft:get-message-sessions` management operation. ([#46575](https://github.com/Azure/azure-sdk-for-python/pull/46575))
-- Management operations (peek, deferred receive, message settlement over the management link, lock renewal, session state, session listing, schedule/cancel) now send `com.microsoft:server-timeout`: the caller's remaining time less one second, or 60 seconds when none was given. Previously no bound was sent, so a stalled service held the call until the AMQP link failed; it now raises a retryable `OperationTimeoutError`, so a persistently stalled service surfaces after roughly four minutes at default retry settings. Matches the .NET, Java and Go SDKs.
 
 ### Bugs Fixed
 
@@ -23,6 +22,7 @@
 ### Other Changes
 
 - When using the async `AmqpOverWebsocket` transport on Python 3.10 or later, `aiohttp>=3.14.0` is now recommended. Earlier `aiohttp` versions have a WebSocket heartbeat bug ([aio-libs/aiohttp#12030](https://github.com/aio-libs/aiohttp/pull/12030)) that can cause the connection to be dropped during long message processing, surfacing as a `SocketError` ("Cannot write to closing transport"). Python 3.9 users must upgrade Python to install an `aiohttp` release containing this fix. ([#44028](https://github.com/Azure/azure-sdk-for-python/issues/44028))
+- Management operations (peek, deferred receive, message settlement over the management link, lock renewal, session state, session listing, schedule/cancel) now send `com.microsoft:server-timeout`: the caller's remaining time less one second, or 60 seconds when none was given. Previously no bound was sent, so a stalled service held the call until the AMQP link failed; it now raises a retryable `OperationTimeoutError`, so a persistently stalled service surfaces after roughly four minutes at default retry settings. Matches the .NET, Java and Go SDKs.
 
 ## 7.14.3 (2025-11-11)
 

sdk/servicebus/azure-servicebus/api.metadata.yml

@@ -1,3 +1,3 @@
 apiMdSha256: 68c237729c78165bea132f330a2e89519b2e6ba19373d40c48c55f09e8a994de
-parserVersion: 0.3.30
+parserVersion: 0.3.31
 pythonVersion: 3.12.10

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

@@ -106,8 +106,7 @@ def get_server_timeout_ms(timeout: Optional[float]) -> int:
     if timeout is None:
         return DEFAULT_SERVER_TIMEOUT_MS
     capped = min(timeout, MAX_SERVER_TIMEOUT_MS / 1000)
-    remaining_ms = int(capped * 1000) - SERVER_TIMEOUT_BUFFER_MS
-    return min(max(remaining_ms, 0), MAX_SERVER_TIMEOUT_MS)
+    return max(int(capped * 1000) - SERVER_TIMEOUT_BUFFER_MS, 0)
 
 
 def build_uri(address, entity):

sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py

@@ -17,9 +17,16 @@
 
 import pytest
 
-from azure.servicebus._common.constants import MAX_SERVER_TIMEOUT_MS, REQUEST_RESPONSE_TIMEOUT
+from azure.servicebus._common import mgmt_handlers
+from azure.servicebus._common.constants import (
+    ERROR_CODE_TIMEOUT,
+    MAX_SERVER_TIMEOUT_MS,
+    MGMT_RESPONSE_MESSAGE_ERROR_CONDITION,
+    REQUEST_RESPONSE_TIMEOUT,
+)
 from azure.servicebus._common.utils import get_server_timeout_ms
 from azure.servicebus._transport._pyamqp_transport import PyamqpTransport
+from azure.servicebus.exceptions import OperationTimeoutError
 
 
 class TestServerTimeoutMillis:
@@ -87,8 +94,7 @@ def fake_create_mgmt_msg(message, application_properties, config, reply_to, **kw
         return handler, captured
 
     def test_default_sent_when_caller_gave_no_timeout(self):
-        # The gap this closes: without a caller timeout the service previously received no bound at
-        # all.
+        # The gap this closes: previously no bound was sent at all.
         handler, captured = self._make_handler()
         handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=None)
         assert captured[REQUEST_RESPONSE_TIMEOUT] == {"TYPE": "UINT", "VALUE": 60000}
@@ -110,8 +116,7 @@ def test_associated_link_name_preserved(self):
         assert REQUEST_RESPONSE_TIMEOUT in captured
 
     def test_sent_on_calls_without_an_associated_link(self):
-        # Operations such as list_sessions pass `keep_alive_associated_link=False` and start from an
-        # empty property map; they must still be bounded.
+        # list_sessions passes keep_alive_associated_link=False, starting from an empty map.
         handler, captured = self._make_handler()
         handler._mgmt_request_response(
             b"op",
@@ -123,6 +128,86 @@ def test_sent_on_calls_without_an_associated_link(self):
         assert captured == {REQUEST_RESPONSE_TIMEOUT: {"TYPE": "UINT", "VALUE": 60000}}
 
 
+class TestServiceTimeoutResponse:
+    """The response half: the service's answer must surface as a retryable
+    `OperationTimeoutError`, which rests on `com.microsoft:timeout` in `errorCondition`."""
+
+    @staticmethod
+    def _response(condition):
+        message = MagicMock()
+        message.application_properties = {MGMT_RESPONSE_MESSAGE_ERROR_CONDITION: condition}
+        return message
+
+    def test_timeout_condition_raises_retryable_operation_timeout_error(self):
+        with pytest.raises(OperationTimeoutError) as exc_info:
+            mgmt_handlers.default(408, self._response(ERROR_CODE_TIMEOUT), "The operation timed out.", PyamqpTransport)
+
+        assert exc_info.value._retryable is True
+
+    def test_success_returns_the_value_untouched(self):
+        message = self._response(None)
+        message.value = {"ok": True}
+        assert mgmt_handlers.default(200, message, None, PyamqpTransport) == {"ok": True}
+
+    def test_uamqp_maps_the_same_condition(self):
+        uamqp_transport = pytest.importorskip(
+            "azure.servicebus._transport._uamqp_transport", reason="uamqp not installed"
+        )
+        with pytest.raises(OperationTimeoutError):
+            mgmt_handlers.default(
+                408,
+                self._response(ERROR_CODE_TIMEOUT),
+                "The operation timed out.",
+                uamqp_transport.UamqpTransport,
+            )
+
+
+class TestUamqpValueType:
+    """uamqp is the one place the value type changes: pyamqp uses a plain dict, uamqp an `AMQPuInt`."""
+
+    def test_transport_supplied_type_is_what_reaches_the_message(self):
+
+        from azure.servicebus._base_handler import BaseHandler
+
+        sentinel = object()
+        captured = {}
+
+        def fake_create_mgmt_msg(message, application_properties, config, reply_to, **kwargs):
+            captured.update(application_properties)
+            return MagicMock()
+
+        handler = BaseHandler.__new__(BaseHandler)
+        handler._amqp_transport = MagicMock()
+        handler._amqp_transport.create_mgmt_msg = fake_create_mgmt_msg
+        handler._amqp_transport.AMQP_UINT_VALUE = lambda ms: sentinel
+        handler._amqp_transport.get_handler_link_name = lambda h: "link-1"
+        handler._amqp_transport.mgmt_client_request = lambda *args, **kwargs: "response"
+        handler._amqp_transport.TIMEOUT_ERROR = TimeoutError
+        handler._open = lambda: None
+        handler._handler = MagicMock()
+        handler._config = MagicMock(encoding="UTF-8")
+        handler._mgmt_target = "queue/$management"
+
+        handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=None)
+        assert captured[REQUEST_RESPONSE_TIMEOUT] is sentinel
+
+    def test_real_uamqp_type_encodes_in_application_properties(self):
+        uamqp = pytest.importorskip("uamqp", reason="uamqp not installed")
+        from azure.servicebus._transport._uamqp_transport import UamqpTransport
+
+        value = UamqpTransport.AMQP_UINT_VALUE(get_server_timeout_ms(None))
+        assert isinstance(value, uamqp.types.AMQPuInt)
+
+        message = UamqpTransport.create_mgmt_msg(
+            message={"operation": "peek"},
+            application_properties={REQUEST_RESPONSE_TIMEOUT: value},
+            config=MagicMock(encoding="UTF-8"),
+            reply_to="queue/$management",
+        )
+
+        assert message.encode_message()
+
+
 class TestAsyncParity:
     """The async management path is a separate implementation and can drift from the sync one."""