[pyAMQP] Align pyAMQP (Azure#37095)

* align

* updates

* pylint

Author: l0lawrence

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_client_async.py

@@ -389,6 +389,9 @@ async def mgmt_request_async(
                 self._mgmt_links[node] = mgmt_link
                 await mgmt_link.open()
 
+        while not await self.client_ready_async():
+            await asyncio.sleep(0.05)
+
         while not await mgmt_link.ready():
             await self._connection.listen(wait=False)
 

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_link_async.py

@@ -100,7 +100,7 @@ def __init__(
         self._is_closed = False
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
-        self._error = None
+        self._error: Optional[AMQPLinkError] = None
 
     async def __aenter__(self) -> "Link":
         await self.attach()
@@ -244,6 +244,11 @@ async def _incoming_detach(self, frame) -> None:
             self._error = error_cls(condition=frame[2][0], description=frame[2][1], info=frame[2][2])
             await self._set_state(LinkState.ERROR)
         else:
+            if self.state != LinkState.DETACH_SENT:
+                # Handle the case of when the remote side detaches without sending an error.
+                # We should detach as per the spec but then retry connecting
+                self._error = AMQPLinkError(condition=ErrorCondition.UnknownError,
+                                          description="Link detached unexpectedly.", retryable=True)
             await self._set_state(LinkState.DETACHED)
 
     async def attach(self) -> None:

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_receiver_async.py

@@ -16,7 +16,7 @@
     DispositionFrame,
 )
 from ..outcomes import Received, Accepted, Rejected, Released, Modified
-from ..error import AMQPLinkError, ErrorCondition
+from ..error import AMQPException, ErrorCondition
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -113,7 +113,7 @@ async def _outgoing_disposition(
             role=self.role, first=first, last=last, settled=settled, state=state, batchable=batchable
         )
         if delivery_tag not in self._received_delivery_tags:
-            raise AMQPLinkError(condition=ErrorCondition.InternalError, description = "Delivery tag not found.")
+            raise AMQPException(condition=ErrorCondition.IllegalState, description = "Delivery tag not found.")
 
         if self.network_trace:
             _LOGGER.debug("-> %r", DispositionFrame(*disposition_frame), extra=self.network_trace_params)

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

@@ -467,6 +467,9 @@ def mgmt_request(
                 self._mgmt_links[node] = mgmt_link
                 mgmt_link.open()
 
+        while not self.client_ready():
+            time.sleep(0.05)
+
         while not mgmt_link.ready():
             self._connection.listen(wait=False)
         operation_type = operation_type or b"empty"

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

@@ -96,7 +96,7 @@ def __init__(
         self._is_closed = False
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
-        self._error = None
+        self._error: Optional[AMQPLinkError] = None
 
     def __enter__(self) -> "Link":
         self.attach()
@@ -239,6 +239,11 @@ def _incoming_detach(self, frame) -> None:
             self._error = error_cls(condition=frame[2][0], description=frame[2][1], info=frame[2][2])
             self._set_state(LinkState.ERROR)
         else:
+            if self.state != LinkState.DETACH_SENT:
+                # Handle the case of when the remote side detaches without sending an error.
+                # We should detach as per the spec but then retry connecting
+                self._error = AMQPLinkError(condition=ErrorCondition.UnknownError,
+                                          description="Link detached unexpectedly.", retryable=True)
             self._set_state(LinkState.DETACHED)
 
     def attach(self) -> None:

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

@@ -13,7 +13,7 @@
 from .constants import LinkState, Role
 from .performatives import TransferFrame, DispositionFrame
 from .outcomes import Received, Accepted, Rejected, Released, Modified
-from .error import AMQPLinkError, ErrorCondition
+from .error import AMQPException, ErrorCondition
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -109,7 +109,7 @@ def _outgoing_disposition(
         batchable: Optional[bool],
     ):
         if delivery_tag not in self._received_delivery_tags:
-            raise AMQPLinkError(condition=ErrorCondition.InternalError, description = "Delivery tag not found.")
+            raise AMQPException(condition=ErrorCondition.IllegalState, description = "Delivery tag not found.")
 
         disposition_frame = DispositionFrame(
             role=self.role, first=first, last=last, settled=settled, state=state, batchable=batchable

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

@@ -126,6 +126,10 @@ def __init__( # pylint:disable=too-many-locals,too-many-statements
             self._port = PORT
         self.state: Optional[ConnectionState] = None
 
+        # Set the port for AmqpOverWebsocket
+        if transport_type.value == TransportType.AmqpOverWebsocket.value:
+            self._port = WEBSOCKET_PORT
+
         # Custom Endpoint
         custom_endpoint_address = kwargs.get("custom_endpoint_address")
         custom_endpoint = None
@@ -157,6 +161,7 @@ def __init__( # pylint:disable=too-many-locals,too-many-statements
             self._transport = sasl_transport(
                 host=endpoint,
                 credential=kwargs["sasl_credential"],
+                port=self._port,
                 custom_endpoint=custom_endpoint,
                 socket_timeout=self._socket_timeout,
                 network_trace_params=self._network_trace_params,

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

@@ -171,6 +171,7 @@ def __init__(
         socket_timeout=SOCKET_TIMEOUT,
         socket_settings=None,
         raise_on_initial_eintr=True,
+        use_tls: bool =True,
         **kwargs
     ):
         self._quick_recv = None
@@ -186,6 +187,8 @@ def __init__(
         self.socket_settings = socket_settings
         self.socket_lock = Lock()
 
+        self._use_tls = use_tls
+
     def connect(self):
         try:
             # are we already connected?
@@ -509,7 +512,8 @@ def __init__(
 
     def _setup_transport(self):
         """Wrap the socket in an SSL object."""
-        self.sock = self._wrap_socket(self.sock, **self.sslopts)
+        if self._use_tls:
+            self.sock = self._wrap_socket(self.sock, **self.sslopts)
         self._quick_recv = self.sock.recv
 
     def _wrap_socket(self, sock, context=None, **sslopts):
@@ -599,7 +603,8 @@ def _shutdown_transport(self):
         """Unwrap a SSL socket, so we can call shutdown()."""
         if self.sock is not None:
             try:
-                self.sock = self.sock.unwrap()
+                if self._use_tls:
+                    self.sock = self.sock.unwrap()
             except OSError:
                 pass
 
@@ -737,11 +742,12 @@ def connect(self):
             ) from None
         try:
             self.sock = create_connection(
-                url="wss://{}".format(self._custom_endpoint or self._host),
+                url="wss://{}".format(self._custom_endpoint or self._host) if self._use_tls
+                    else "ws://{}".format(self._custom_endpoint or self._host),
                 subprotocols=[AMQP_WS_SUBPROTOCOL],
                 timeout=self.socket_timeout,    # timeout for read/write operations
                 skip_utf8_validation=True,
-                sslopt=self.sslopts,
+                sslopt=self.sslopts if self._use_tls else None,
                 http_proxy_host=http_proxy_host,
                 http_proxy_port=http_proxy_port,
                 http_proxy_auth=http_proxy_auth,

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_client_async.py

@@ -244,7 +244,7 @@ async def open_async(self, connection=None):
             self._external_connection = True
         if not self._connection:
             self._connection = Connection(
-                "amqps://" + self._hostname,
+                "amqps://" + self._hostname if self._use_tls else "amqp://" + self._hostname,
                 sasl_credential=self._auth.sasl,
                 ssl_opts={'ca_certs': self._connection_verify or certifi.where()},
                 container_id=self._name,
@@ -257,6 +257,7 @@ async def open_async(self, connection=None):
                 http_proxy=self._http_proxy,
                 custom_endpoint_address=self._custom_endpoint_address,
                 socket_timeout=self._socket_timeout,
+                use_tls=self._use_tls,
             )
             await self._connection.open()
         if not self._session:

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_connection_async.py

@@ -107,6 +107,10 @@ def __init__(# pylint:disable=too-many-locals,too-many-statements
             self._port = PORT
         self.state: Optional[ConnectionState] = None
 
+        # Set the port for AmqpOverWebsocket
+        if transport_type.value == TransportType.AmqpOverWebsocket.value:
+            self._port = WEBSOCKET_PORT
+
         # Custom Endpoint
         custom_endpoint_address = kwargs.get("custom_endpoint_address")
         custom_endpoint = None
@@ -141,6 +145,7 @@ def __init__(# pylint:disable=too-many-locals,too-many-statements
             self._transport: Union[SASLTransport, SASLWithWebSocket, AsyncTransport] = sasl_transport(
                 host=endpoint,
                 credential=kwargs["sasl_credential"],
+                port=self._port,
                 custom_endpoint=custom_endpoint,
                 socket_timeout=self._socket_timeout,
                 network_trace_params=self._network_trace_params,