[corehttp] Misc updates (#34394)

These updates are similar updates that were made to azure-core
previously.

- Changed ServiceNamedKey to be a class instead of a namedtuple to fix a verifytypes issue.
- Typing/linting updates.
- Async auth policy lock refactoring.
- Refactored streaming tests to remove requirement on external endpoint.

Signed-off-by: Paul Van Eck <[email protected]>

Author: pvaneck

sdk/core/corehttp/corehttp/credentials.py

@@ -4,7 +4,6 @@
 # license information.
 # -------------------------------------------------------------------------
 from __future__ import annotations
-from collections import namedtuple
 from types import TracebackType
 from typing import Any, NamedTuple, Optional, AsyncContextManager, Type
 from typing_extensions import Protocol, runtime_checkable
@@ -40,7 +39,12 @@ def get_token(self, *scopes: str, claims: Optional[str] = None, **kwargs: Any) -
         ...
 
 
-ServiceNamedKey = namedtuple("ServiceNamedKey", ["name", "key"])
+class ServiceNamedKey(NamedTuple):
+    """Represents a name and key pair."""
+
+    name: str
+    key: str
+
 
 __all__ = [
     "AccessToken",

sdk/core/corehttp/corehttp/runtime/_base.py

@@ -23,11 +23,11 @@
 # IN THE SOFTWARE.
 #
 # --------------------------------------------------------------------------
-from typing import Any
+from typing import Any, Dict
 from urllib.parse import urlparse
 
 
-def _format_url_section(template, **kwargs):
+def _format_url_section(template, **kwargs: Dict[str, str]) -> str:
     """String format the template with the kwargs, auto-skip sections of the template that are NOT in the kwargs.
 
     By default in Python, "format" will raise a KeyError if a template element is not found. Here the section between
@@ -53,6 +53,7 @@ def _format_url_section(template, **kwargs):
                     f"The value provided for the url part '{template}' was incorrect, and resulted in an invalid url"
                 ) from key
             last_template = template
+    return last_template
 
 
 def _urljoin(base_url: str, stub_url: str) -> str:
@@ -71,7 +72,7 @@ def _urljoin(base_url: str, stub_url: str) -> str:
     stub_url_path = split_url.pop(0)
     stub_url_query = split_url.pop() if split_url else None
 
-    # Note that _replace is a public API named that way to avoid to avoid conflicts in namedtuple
+    # Note that _replace is a public API named that way to avoid conflicts in namedtuple
     # https://docs.python.org/3/library/collections.html?highlight=namedtuple#collections.namedtuple
     parsed_base_url = parsed_base_url._replace(
         path=parsed_base_url.path.rstrip("/") + "/" + stub_url_path,

sdk/core/corehttp/corehttp/runtime/policies/_authentication_async.py

@@ -4,7 +4,6 @@
 # license information.
 # -------------------------------------------------------------------------
 from __future__ import annotations
-import asyncio
 import time
 from typing import TYPE_CHECKING, Any, Awaitable, Optional, cast, TypeVar
 
@@ -14,6 +13,7 @@
 from ._base_async import AsyncHTTPPolicy
 from ._authentication import _BearerTokenCredentialPolicyBase
 from ...rest import AsyncHttpResponse, HttpRequest
+from ...utils._utils import get_running_async_lock
 
 if TYPE_CHECKING:
     from ...credentials import AsyncTokenCredential
@@ -36,10 +36,16 @@ def __init__(
     ) -> None:
         super().__init__()
         self._credential = credential
-        self._lock = asyncio.Lock()
+        self._lock_instance = None
         self._scopes = scopes
         self._token: Optional["AccessToken"] = None
 
+    @property
+    def _lock(self):
+        if self._lock_instance is None:
+            self._lock_instance = get_running_async_lock()
+        return self._lock_instance
+
     async def on_request(self, request: PipelineRequest[HTTPRequestType]) -> None:
         """Adds a bearer token Authorization header to request and sends request to next policy.
 

sdk/core/corehttp/corehttp/runtime/policies/_universal.py

@@ -35,7 +35,7 @@
 import xml.etree.ElementTree as ET
 import types
 import re
-from typing import IO, cast, Union, Optional, AnyStr, Dict, Any, Mapping, TYPE_CHECKING
+from typing import IO, cast, Union, Optional, AnyStr, Dict, Any, MutableMapping, TYPE_CHECKING
 
 from ... import __version__ as core_version
 from ...exceptions import DecodeError
@@ -449,12 +449,12 @@ class ProxyPolicy(SansIOHTTPPolicy[HTTPRequestType, HTTPResponseType]):
     Dictionary mapping protocol or protocol and host to the URL of the proxy
     to be used on each Request.
 
-    :param dict proxies: Maps protocol or protocol and hostname to the URL
+    :param MutableMapping proxies: Maps protocol or protocol and hostname to the URL
      of the proxy.
     """
 
     def __init__(
-        self, proxies: Optional[Mapping[str, str]] = None, **kwargs: Any
+        self, proxies: Optional[MutableMapping[str, str]] = None, **kwargs: Any
     ):  # pylint: disable=unused-argument,super-init-not-called
         self.proxies = proxies
 

sdk/core/corehttp/corehttp/transport/requests/_bigger_block_size_http_adapters.py

@@ -25,18 +25,21 @@
 # --------------------------------------------------------------------------
 
 import sys
+from typing import MutableMapping, Optional
+
 from requests.adapters import HTTPAdapter  # pylint: disable=all
+from urllib3.connectionpool import ConnectionPool
 
 
 class BiggerBlockSizeHTTPAdapter(HTTPAdapter):
-    def get_connection(self, url, proxies=None):
+    def get_connection(self, url: str, proxies: Optional[MutableMapping[str, str]] = None) -> ConnectionPool:
         """Returns a urllib3 connection for the given URL. This should not be
         called from user code, and is only exposed for use when subclassing the
         :class:`HTTPAdapter <requests.adapters.HTTPAdapter>`.
 
         :param str url: The URL to connect to.
         :param dict proxies: (optional) A Requests-style dictionary of proxies used on this request.
-        :rtype: urllib3.ConnectionPool
+        :rtype: urllib3.connectionpool.ConnectionPool
         :returns: The urllib3 ConnectionPool for the given URL.
         """
         conn = super(BiggerBlockSizeHTTPAdapter, self).get_connection(url, proxies)

sdk/core/corehttp/corehttp/utils/_utils.py

@@ -7,6 +7,7 @@
 import datetime
 from typing import (
     Any,
+    AsyncContextManager,
     Iterable,
     Iterator,
     Mapping,
@@ -164,3 +165,21 @@ def get_file_items(files: "FilesType") -> Sequence[Tuple[str, "FileType"]]:
         # though realistically it is ordered python 3.7 and after
         return cast(Sequence[Tuple[str, "FileType"]], files.items())
     return files
+
+
+def get_running_async_lock() -> AsyncContextManager:
+    """Get a lock instance from the async library that the current context is running under.
+    :return: An instance of the running async library's Lock class.
+    :rtype: AsyncContextManager
+    :raises: RuntimeError if the current context is not running under an async library.
+    """
+
+    try:
+        import asyncio
+
+        # Check if we are running in an asyncio event loop.
+        asyncio.get_running_loop()
+        return asyncio.Lock()
+    except RuntimeError as err:
+        # Otherwise, assume we are running in a trio event loop, but this currently isn't supported.
+        raise RuntimeError("An asyncio event loop is required.") from err

sdk/core/corehttp/tests/async_tests/test_authentication_async.py

@@ -252,3 +252,14 @@ async def get_token(self, *scopes, **kwargs):
 
     cred = TestTokenCredential()
     await cred.get_token("scope")
+
+
+@pytest.mark.asyncio
+async def test_async_token_credential_asyncio_lock():
+    auth_policy = AsyncBearerTokenCredentialPolicy(Mock(), "scope")
+    assert isinstance(auth_policy._lock, asyncio.Lock)
+
+
+def test_async_token_credential_sync():
+    """Verify that AsyncBearerTokenCredentialPolicy can be constructed in a synchronous context."""
+    AsyncBearerTokenCredentialPolicy(Mock(), "scope")