[sb] update conn str (Azure#38655)

* emulator

* nit

* fixing

* changelog

* parse_conn_string tests

* mgmt client not supported yet -- don't want to have customers see a diff in behavior if they add support suddenly

* changelog

* pylint

Author: l0lawrence

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 7.13.1 (Unreleased)
 
 ### Features Added
+- Added emulator support, ServiceBusAdministrationClient is currently not supported by the emulator.
 
 ### Breaking Changes
 

sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py

@@ -50,15 +50,16 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-def _parse_conn_str(
+def _parse_conn_str(  # pylint:disable=too-many-statements
     conn_str: str, check_case: Optional[bool] = False
-) -> Tuple[str, Optional[str], Optional[str], str, Optional[str], Optional[int]]:
+) -> Tuple[str, Optional[str], Optional[str], str, Optional[str], Optional[int], bool]:
     endpoint = None
     shared_access_key_name = None
     shared_access_key = None
     entity_path: Optional[str] = None
     shared_access_signature: Optional[str] = None
     shared_access_signature_expiry: Optional[int] = None
+    use_emulator: Optional[str] = None
 
     # split connection string into properties
     conn_properties = [s.split("=", 1) for s in conn_str.strip().rstrip(";").split(";")]
@@ -72,6 +73,7 @@ def _parse_conn_str(
         shared_access_key_name = conn_settings.get("SharedAccessKeyName")
         endpoint = conn_settings.get("Endpoint")
         entity_path = conn_settings.get("EntityPath")
+        use_emulator = conn_settings.get("UseDevelopmentEmulator")
 
     # non case sensitive check when parsing connection string for internal use
     for key, value in conn_settings.items():
@@ -101,6 +103,8 @@ def _parse_conn_str(
                 shared_access_key = value
             elif key.lower() == "entitypath":
                 entity_path = value
+            elif key.lower() == "usedevelopmentemulator":
+                use_emulator = value
 
     entity = cast(str, entity_path)
 
@@ -112,6 +116,8 @@ def _parse_conn_str(
         raise ValueError("Invalid Endpoint on the Connection String.")
     host = cast(str, parsed.netloc.strip())
 
+    emulator = use_emulator == "true"
+
     if any([shared_access_key, shared_access_key_name]) and not all([shared_access_key, shared_access_key_name]):
         raise ValueError("Connection string must have both SharedAccessKeyName and SharedAccessKey.")
     if shared_access_signature and shared_access_key:
@@ -126,6 +132,7 @@ def _parse_conn_str(
         entity,
         str(shared_access_signature) if shared_access_signature else None,
         shared_access_signature_expiry,
+        emulator,
     )
 
 
@@ -274,7 +281,7 @@ def __init__(
 
     @classmethod
     def _convert_connection_string_to_kwargs(cls, conn_str: str, **kwargs: Any) -> Dict[str, Any]:
-        host, policy, key, entity_in_conn_str, token, token_expiry = _parse_conn_str(conn_str)
+        host, policy, key, entity_in_conn_str, token, token_expiry, emulator = _parse_conn_str(conn_str)
         queue_name = kwargs.get("queue_name")
         topic_name = kwargs.get("topic_name")
         if not (queue_name or topic_name or entity_in_conn_str):
@@ -295,6 +302,9 @@ def _convert_connection_string_to_kwargs(cls, conn_str: str, **kwargs: Any) -> D
 
         kwargs["fully_qualified_namespace"] = host
         kwargs["entity_name"] = entity_in_conn_str or entity_in_kwargs
+        # Check if emulator is in use, unset tls if it is
+        if emulator:
+            kwargs["use_tls"] = False
 
         # Set the type to sync credentials, unless async credentials are passed in.
         token_cred_type = kwargs.pop("token_cred_type", ServiceBusSASTokenCredential)

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

@@ -41,6 +41,7 @@ def __init__(self, **kwargs):
         self.auto_reconnect = kwargs.get("auto_reconnect", True)
         self.keep_alive = kwargs.get("keep_alive", 30)
         self.timeout: float = kwargs.get("timeout", 60)
+        self.use_tls = kwargs.get("use_tls", True)
         default_socket_timeout = 0.2
 
         if self.http_proxy or self.transport_type.value == TransportType.AmqpOverWebsocket.value:

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

@@ -86,7 +86,9 @@ def parse_connection_string(conn_str: str) -> "ServiceBusConnectionStringPropert
     :return: A properties model containing the parsed connection string.
     :rtype: ~azure.servicebus.ServiceBusConnectionStringProperties
     """
-    fully_qualified_namespace, policy, key, entity, signature = _parse_conn_str(conn_str, True)[:-1]
+    fully_qualified_namespace, policy, key, entity, signature, emulator = _parse_conn_str(conn_str, True)[ # pylint: disable=unused-variable
+        :-1
+    ]
     endpoint = "sb://" + fully_qualified_namespace + "/"
     return ServiceBusConnectionStringProperties(
         fully_qualified_namespace=fully_qualified_namespace,

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_client.py

@@ -259,7 +259,8 @@ def from_connection_string(
                 :caption: Create a new instance of the ServiceBusClient from connection string.
 
         """
-        host, policy, key, entity_in_conn_str, token, token_expiry = _parse_conn_str(conn_str)
+        host, policy, key, entity_in_conn_str, token, token_expiry, emulator = _parse_conn_str(conn_str)
+        kwargs["use_tls"] = not emulator
         if token and token_expiry:
             credential = ServiceBusSASTokenCredential(token, token_expiry)
         elif policy and key:
@@ -324,6 +325,7 @@ def get_queue_sender(self, queue_name: str, **kwargs: Any) -> ServiceBusSender:
             connection_verify=self._connection_verify,
             ssl_context=self._ssl_context,
             amqp_transport=self._amqp_transport,
+            use_tls=self._config.use_tls,
             **kwargs,
         )
         self._handlers.add(handler)
@@ -445,6 +447,7 @@ def get_queue_receiver(
             connection_verify=self._connection_verify,
             ssl_context=self._ssl_context,
             amqp_transport=self._amqp_transport,
+            use_tls=self._config.use_tls,
             **kwargs,
         )
         self._handlers.add(handler)
@@ -507,6 +510,7 @@ def get_topic_sender(
             amqp_transport=self._amqp_transport,
             client_identifier=client_identifier,
             socket_timeout=socket_timeout,
+            use_tls=self._config.use_tls,
             **kwargs,
         )
         self._handlers.add(handler)
@@ -630,6 +634,7 @@ def get_subscription_receiver(
                 connection_verify=self._connection_verify,
                 ssl_context=self._ssl_context,
                 amqp_transport=self._amqp_transport,
+                use_tls=self._config.use_tls,
                 **kwargs,
             )
         except ValueError:
@@ -659,6 +664,7 @@ def get_subscription_receiver(
                 connection_verify=self._connection_verify,
                 ssl_context=self._ssl_context,
                 amqp_transport=self._amqp_transport,
+                use_tls=self._config.use_tls,
                 **kwargs,
             )
         self._handlers.add(handler)

sdk/servicebus/azure-servicebus/azure/servicebus/_transport/_pyamqp_transport.py

@@ -467,6 +467,7 @@ def create_send_client(config: "Configuration", **kwargs: Any) -> "SendClient":
             transport_type=config.transport_type,
             http_proxy=config.http_proxy,
             socket_timeout=config.socket_timeout,
+            use_tls=config.use_tls,
             **kwargs,
         )
 
@@ -596,6 +597,7 @@ def create_receive_client(receiver: "ServiceBusReceiver", **kwargs: "Any") -> "R
             ),
             on_attach=functools.partial(PyamqpTransport.on_attach, receiver),
             socket_timeout=config.socket_timeout,
+            use_tls=config.use_tls,
             **kwargs,
         )
 

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_client_async.py

@@ -226,7 +226,8 @@ def from_connection_string(
                 :caption: Create a new instance of the ServiceBusClient from connection string.
 
         """
-        host, policy, key, entity_in_conn_str, token, token_expiry = _parse_conn_str(conn_str)
+        host, policy, key, entity_in_conn_str, token, token_expiry, emulator = _parse_conn_str(conn_str)
+        kwargs["use_tls"] = not emulator
         if token and token_expiry:
             credential = ServiceBusSASTokenCredential(token, token_expiry)
         elif policy and key:
@@ -312,6 +313,7 @@ def get_queue_sender(self, queue_name: str, **kwargs: Any) -> ServiceBusSender:
             connection_verify=self._connection_verify,
             ssl_context=self._ssl_context,
             amqp_transport=self._amqp_transport,
+            use_tls=self._config.use_tls,
             **kwargs,
         )
         self._handlers.add(handler)
@@ -433,6 +435,7 @@ def get_queue_receiver(
             connection_verify=self._connection_verify,
             ssl_context=self._ssl_context,
             amqp_transport=self._amqp_transport,
+            use_tls=self._config.use_tls,
             **kwargs,
         )
         self._handlers.add(handler)
@@ -486,6 +489,7 @@ def get_topic_sender(self, topic_name: str, **kwargs: Any) -> ServiceBusSender:
             connection_verify=self._connection_verify,
             ssl_context=self._ssl_context,
             amqp_transport=self._amqp_transport,
+            use_tls=self._config.use_tls,
             **kwargs,
         )
         self._handlers.add(handler)
@@ -609,6 +613,7 @@ def get_subscription_receiver(
                 connection_verify=self._connection_verify,
                 ssl_context=self._ssl_context,
                 amqp_transport=self._amqp_transport,
+                use_tls=self._config.use_tls,
                 **kwargs,
             )
         except ValueError:
@@ -638,6 +643,7 @@ def get_subscription_receiver(
                 connection_verify=self._connection_verify,
                 ssl_context=self._ssl_context,
                 amqp_transport=self._amqp_transport,
+                use_tls=self._config.use_tls,
                 **kwargs,
             )
         self._handlers.add(handler)

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_transport/_pyamqp_transport_async.py

@@ -116,6 +116,7 @@ def create_send_client_async(config: "Configuration", **kwargs: Any) -> "SendCli
             transport_type=config.transport_type,
             http_proxy=config.http_proxy,
             socket_timeout=config.socket_timeout,
+            use_tls=config.use_tls,
             **kwargs,
         )
 
@@ -219,6 +220,7 @@ def create_receive_client_async(
             ),
             on_attach=functools.partial(PyamqpTransportAsync.on_attach_async, receiver),
             socket_timeout=config.socket_timeout,
+            use_tls=config.use_tls,
             **kwargs,
         )
 

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

@@ -248,14 +248,10 @@ def from_connection_string(
          in reduced feature compatibility.
         :paramtype api_version: str or ApiVersion
         """
-        (
-            endpoint,
-            shared_access_key_name,
-            shared_access_key,
-            _,
-            token,
-            token_expiry,
-        ) = _parse_conn_str(conn_str)
+        # pylint: disable=unused-variable
+        (endpoint, shared_access_key_name, shared_access_key, _, token, token_expiry, emulator) = _parse_conn_str(
+            conn_str
+        )
         if token and token_expiry:
             credential = ServiceBusSASTokenCredential(token, token_expiry)
         elif shared_access_key_name and shared_access_key:

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

@@ -244,14 +244,10 @@ def from_connection_string(
         :returns: Returns a ServiceBusAdministrationClient.
         :rtype: ~azure.servicebus.management.ServiceBusAdministrationClient
         """
-        (
-            endpoint,
-            shared_access_key_name,
-            shared_access_key,
-            _,
-            token,
-            token_expiry,
-        ) = _parse_conn_str(conn_str)
+        # pylint: disable=unused-variable
+        (endpoint, shared_access_key_name, shared_access_key, _, token, token_expiry, emulator) = _parse_conn_str(
+            conn_str
+        )
         if token and token_expiry:
             credential = ServiceBusSASTokenCredential(token, token_expiry)
         elif shared_access_key_name and shared_access_key: