Add Unit test for AMQP Client Link with max message size = 0 (#43582)

* Added unit test for max message size in aqmp attach frame

* triggering PR tests, no code changes

* addressing copilot comments

* nit changes

* nit change

---------

Co-authored-by: Himanshu Arora <[email protected]>

Author: PIRO-V

sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_link.py

@@ -3,6 +3,9 @@
 from azure.eventhub._pyamqp.link import Link
 from azure.eventhub._pyamqp.receiver import ReceiverLink
 from azure.eventhub._pyamqp.constants import LinkState
+from azure.eventhub._pyamqp.link import Source, Target
+from unittest.mock import Mock, patch
+from azure.eventhub._pyamqp.constants import LINK_MAX_MESSAGE_SIZE
 import pytest
 
 
@@ -84,6 +87,80 @@ def test_receive_transfer_frame_multiple():
     link._incoming_transfer(transfer_frame_two)
     assert link.current_link_credit == 1
 
+def test_max_message_size_negotiation_with_client_unlimited():
+    """
+    Test AMQP attach frame negotiation where client sends max_message_size=0 ( unlimited )
+    and server responds with its limit (20MB), resulting in final size of 20MB.
+    """
+    # Test constants to improve maintainability and reduce duplication
+    TEST_LINK_NAME = "test_link"
+    TEST_SOURCE_ADDRESS = "test_source"
+    TEST_TARGET_ADDRESS = "test_target"
+    TEST_HANDLE = 3
+    TEST_SND_SETTLE_MODE = 0
+    TEST_RCV_SETTLE_MODE = 1
+    SERVER_MAX_MESSAGE_SIZE = 20 * 1024 * 1024
+
+    mock_session = Mock()
+    mock_connection = Mock()
+    mock_session._connection = mock_connection
+
+    link = Link(
+        mock_session,
+        TEST_HANDLE,
+        name=TEST_LINK_NAME,
+        role=False,  # Sender role
+        source_address=TEST_SOURCE_ADDRESS, 
+        target_address=TEST_TARGET_ADDRESS,
+        network_trace=False,
+        network_trace_params={},
+        max_message_size=LINK_MAX_MESSAGE_SIZE
+    )
+
+    # Verifying that client sends 0 (unlimited) in attach frame
+    assert link.max_message_size == 0, f"Expected client max_message_size=0, got {link.max_message_size}"
+    
+    # Simulating server's attach response with 20MB limit, Mock incoming attach frame from server
+    mock_attach_frame = [
+        TEST_LINK_NAME,                           # 0: name
+        TEST_HANDLE,                              # 1: handle
+        False,                                    # 2: role
+        TEST_SND_SETTLE_MODE,                     # 3: snd-settle-mode
+        TEST_RCV_SETTLE_MODE,                     # 4: rcv-settle-mode
+        Source(address=TEST_SOURCE_ADDRESS),      # 5: source
+        Target(address=TEST_TARGET_ADDRESS),      # 6: target
+        None,                                     # 7: unsettled
+        False,                                    # 8: incomplete-unsettled
+        None,                                     # 9: initial-delivery-count
+        SERVER_MAX_MESSAGE_SIZE,                  # 10: max-message-size
+        None,                                     # 11: offered_capabilities
+        None,                                     # 12: desired_capabilities
+        None,                                     # 13: remote_properties
+    ]
+
+    # Testing _outgoing_attach()
+    with patch.object(link, '_outgoing_attach') as mock_outgoing_attach:
+        # Trigger outgoing attach
+        link.attach()
+
+        # Verifying _outgoing_attach was called
+        assert mock_outgoing_attach.called, "_outgoing_attach should have been called during attach()"
+
+        # Get the attach frame that would be sent to server
+        call_args = mock_outgoing_attach.call_args
+        if call_args and call_args[0]:
+            attach_frame = call_args[0][0]
+            assert attach_frame.max_message_size == 0, f"Expected client to send max_message_size=0, got {attach_frame.max_message_size}"
+
+    # Testing _incoming_attach()
+    with patch.object(link, '_outgoing_attach') as mock_outgoing_response:
+
+        # Calling _incoming_attach to process server's response
+        link._incoming_attach(mock_attach_frame)
+
+        expected_final_size = SERVER_MAX_MESSAGE_SIZE
+        # Verifying remote_max_message_size is set correctly
+        assert link.remote_max_message_size == expected_final_size, f"Expected remote_max_message_size={expected_final_size}, got {link.remote_max_message_size}"
 
 def test_receive_transfer_continuation_frame():
     session = None