fix(servicebus): Forward **kwargs to _build_pipeline in ServiceBusAdministrationClient

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

@@ -124,18 +124,18 @@ def __init__(
                 self._raw_amqp_message = AmqpAnnotatedMessage(message=kwargs["message"])
         else:
             self._build_annotated_message(body)
-            self.application_properties = application_properties
-            self.session_id = session_id
-            self.message_id = message_id
-            self.content_type = content_type
-            self.correlation_id = correlation_id
-            self.to = to
-            self.reply_to = reply_to
-            self.reply_to_session_id = reply_to_session_id
-            self.subject = subject
-            self.scheduled_enqueue_time_utc = scheduled_enqueue_time_utc
-            self.time_to_live = time_to_live
-            self.partition_key = partition_key
+            self.application_properties = application_properties  # type: ignore[assignment]
+            self.session_id = session_id  # type: ignore[assignment]
+            self.message_id = message_id  # type: ignore[assignment]
+            self.content_type = content_type  # type: ignore[assignment]
+            self.correlation_id = correlation_id  # type: ignore[assignment]
+            self.to = to  # type: ignore[assignment]
+            self.reply_to = reply_to  # type: ignore[assignment]
+            self.reply_to_session_id = reply_to_session_id  # type: ignore[assignment]
+            self.subject = subject  # type: ignore[assignment]
+            self.scheduled_enqueue_time_utc = scheduled_enqueue_time_utc  # type: ignore[assignment]
+            self.time_to_live = time_to_live  # type: ignore[assignment]
+            self.partition_key = partition_key  # type: ignore[assignment]
 
     def __str__(self) -> str:
         return str(self.raw_amqp_message)

sdk/servicebus/azure-servicebus/azure/servicebus/aio/management/_management_client_async.py

@@ -130,7 +130,7 @@ def __init__(
         self._config = ServiceBusManagementClientConfiguration(
             self._endpoint, credential=self._credential, api_version=api_version, **kwargs
         )
-        self._pipeline = self._build_pipeline()
+        self._pipeline = self._build_pipeline(**kwargs)
         self._impl = ServiceBusManagementClientImpl(
             endpoint=fully_qualified_namespace,
             credential=self._credential,

sdk/servicebus/azure-servicebus/azure/servicebus/management/_management_client.py

@@ -130,7 +130,7 @@ def __init__(
         self._config = ServiceBusManagementClientConfiguration(
             self._endpoint, credential=self._credential, api_version=api_version, **kwargs
         )
-        self._pipeline = self._build_pipeline()
+        self._pipeline = self._build_pipeline(**kwargs)
         self._impl = ServiceBusManagementClientImpl(
             endpoint=fully_qualified_namespace,
             credential=self._credential,

sdk/servicebus/azure-servicebus/mypy.ini

@@ -1,9 +1,15 @@
 [mypy]
-python_version = 3.7
+python_version = 3.10
 warn_unused_configs = True
 ignore_missing_imports = True
 
 # Per-module options:
 
 [mypy-azure.servicebus.management._generated.*]
 ignore_errors = True
+
+[mypy-azure.servicebus._transport._uamqp_transport]
+ignore_errors = True
+
+[mypy-azure.servicebus._transport._pyamqp_transport]
+ignore_errors = True

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

@@ -0,0 +1,103 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+"""Unit tests verifying that ServiceBusAdministrationClient forwards
+**kwargs from __init__ to _build_pipeline (both sync and async).
+
+Regression test for https://github.com/Azure/azure-sdk-for-python/issues/44999
+"""
+
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from azure.servicebus.management import ServiceBusAdministrationClient
+
+
+class TestServiceBusAdministrationClientKwargs:
+    """Verify kwargs like connection_verify and transport reach _build_pipeline."""
+
+    def test_sync_build_pipeline_receives_kwargs(self):
+        """Sync client should forward **kwargs to _build_pipeline."""
+        credential = MagicMock()
+        with patch.object(
+            ServiceBusAdministrationClient,
+            "_build_pipeline",
+            return_value=MagicMock(),
+        ) as mock_build:
+            ServiceBusAdministrationClient(
+                "fake.servicebus.windows.net",
+                credential,
+                connection_verify="/path/to/ca-bundle.crt",
+            )
+            mock_build.assert_called_once()
+            call_kwargs = mock_build.call_args.kwargs
+            assert call_kwargs.get("connection_verify") == "/path/to/ca-bundle.crt"
+
+    def test_sync_build_pipeline_receives_custom_transport(self):
+        """Sync client should forward a custom transport kwarg to _build_pipeline."""
+        credential = MagicMock()
+        custom_transport = MagicMock()
+        with patch.object(
+            ServiceBusAdministrationClient,
+            "_build_pipeline",
+            return_value=MagicMock(),
+        ) as mock_build:
+            ServiceBusAdministrationClient(
+                "fake.servicebus.windows.net",
+                credential,
+                transport=custom_transport,
+            )
+            mock_build.assert_called_once()
+            call_kwargs = mock_build.call_args.kwargs
+            assert call_kwargs.get("transport") is custom_transport
+
+
+@pytest.mark.asyncio
+class TestServiceBusAdministrationClientKwargsAsync:
+    """Verify kwargs reach _build_pipeline on the async client."""
+
+    async def test_async_build_pipeline_receives_kwargs(self):
+        """Async client should forward **kwargs to _build_pipeline."""
+        from azure.servicebus.aio.management import (
+            ServiceBusAdministrationClient as AsyncClient,
+        )
+
+        credential = MagicMock()
+        with patch.object(
+            AsyncClient,
+            "_build_pipeline",
+            return_value=MagicMock(),
+        ) as mock_build:
+            AsyncClient(
+                "fake.servicebus.windows.net",
+                credential,
+                connection_verify="/path/to/ca-bundle.crt",
+            )
+            mock_build.assert_called_once()
+            call_kwargs = mock_build.call_args.kwargs
+            assert call_kwargs.get("connection_verify") == "/path/to/ca-bundle.crt"
+
+    async def test_async_build_pipeline_receives_custom_transport(self):
+        """Async client should forward a custom transport kwarg to _build_pipeline."""
+        from azure.servicebus.aio.management import (
+            ServiceBusAdministrationClient as AsyncClient,
+        )
+
+        credential = MagicMock()
+        custom_transport = MagicMock()
+        with patch.object(
+            AsyncClient,
+            "_build_pipeline",
+            return_value=MagicMock(),
+        ) as mock_build:
+            AsyncClient(
+                "fake.servicebus.windows.net",
+                credential,
+                transport=custom_transport,
+            )
+            mock_build.assert_called_once()
+            call_kwargs = mock_build.call_args.kwargs
+            assert call_kwargs.get("transport") is custom_transport