[corehttp] use typeguard for HTTPPolicy/SansIOHTTPPolicy check not isinstance

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

@@ -26,6 +26,7 @@
 from __future__ import annotations
 import logging
 from typing import Generic, TypeVar, Union, Any, List, Optional, Iterable, ContextManager
+from typing_extensions import TypeGuard
 
 from . import (
     PipelineRequest,
@@ -42,6 +43,18 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+def is_http_policy(policy) -> TypeGuard[HTTPPolicy]:
+    if hasattr(policy, "send"):
+        return True
+    return False
+
+
+def is_sansio_http_policy(policy) -> TypeGuard[SansIOHTTPPolicy]:
+    if hasattr(policy, "on_request") and hasattr(policy, "on_response"):
+        return True
+    return False
+
+
 class _SansIOHTTPPolicyRunner(HTTPPolicy[HTTPRequestType, HTTPResponseType]):
     """Sync implementation of the SansIO policy.
 
@@ -123,10 +136,10 @@ def __init__(
         self._transport = transport
 
         for policy in policies or []:
-            if isinstance(policy, SansIOHTTPPolicy):
-                self._impl_policies.append(_SansIOHTTPPolicyRunner(policy))
-            elif policy:
+            if is_http_policy(policy):
                 self._impl_policies.append(policy)
+            elif is_sansio_http_policy(policy):
+                self._impl_policies.append(_SansIOHTTPPolicyRunner(policy))
         for index in range(len(self._impl_policies) - 1):
             self._impl_policies[index].next = self._impl_policies[index + 1]
         if self._impl_policies:

sdk/core/corehttp/corehttp/runtime/pipeline/_base_async.py

@@ -26,17 +26,24 @@
 from __future__ import annotations
 from types import TracebackType
 from typing import Any, Union, Generic, TypeVar, List, Optional, Iterable, Type
-from typing_extensions import AsyncContextManager
+from typing_extensions import AsyncContextManager, TypeGuard
 
 from . import PipelineRequest, PipelineResponse, PipelineContext
 from ..policies import AsyncHTTPPolicy, SansIOHTTPPolicy
+from ..pipeline._base import is_sansio_http_policy
 from ._tools_async import await_result as _await_result
 from ...transport import AsyncHttpTransport
 
 AsyncHTTPResponseType = TypeVar("AsyncHTTPResponseType")
 HTTPRequestType = TypeVar("HTTPRequestType")
 
 
+def is_async_http_policy(policy) -> TypeGuard[AsyncHTTPPolicy]:
+    if hasattr(policy, "send"):
+        return True
+    return False
+
+
 class _SansIOAsyncHTTPPolicyRunner(
     AsyncHTTPPolicy[HTTPRequestType, AsyncHTTPResponseType]
 ):  # pylint: disable=unsubscriptable-object
@@ -127,10 +134,10 @@ def __init__(
         self._transport = transport
 
         for policy in policies or []:
-            if isinstance(policy, SansIOHTTPPolicy):
-                self._impl_policies.append(_SansIOAsyncHTTPPolicyRunner(policy))
-            elif policy:
+            if is_async_http_policy(policy):
                 self._impl_policies.append(policy)
+            elif is_sansio_http_policy(policy):
+                self._impl_policies.append(_SansIOAsyncHTTPPolicyRunner(policy))
         for index in range(len(self._impl_policies) - 1):
             self._impl_policies[index].next = self._impl_policies[index + 1]
         if self._impl_policies:

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

@@ -16,6 +16,7 @@
     SansIOHTTPPolicy,
 )
 from corehttp.rest import HttpRequest
+from azure.core.pipeline.policies import AzureKeyCredentialPolicy
 import pytest
 
 pytestmark = pytest.mark.asyncio
@@ -238,6 +239,27 @@ async def fake_send(*args, **kwargs):
     policy.on_exception.assert_called_once_with(policy.request)
 
 
+async def test_azure_core_sans_io_policy():
+    """Tests to see that we can use an azure.core SansIOHTTPPolicy with the corehttp Pipeline"""
+
+    class TestPolicy(AzureKeyCredentialPolicy):
+        def __init__(self, *args, **kwargs):
+            super(TestPolicy, self).__init__(*args, **kwargs)
+            self.on_exception = Mock(return_value=False)
+            self.on_request = Mock()
+
+    credential = Mock(
+        get_token=Mock(return_value=get_completed_future(AccessToken("***", int(time.time()) + 3600))), key="key"
+    )
+    policy = TestPolicy(credential, "scope")
+    transport = Mock(send=Mock(return_value=get_completed_future(Mock(status_code=200))))
+
+    pipeline = AsyncPipeline(transport=transport, policies=[policy])
+    await pipeline.run(HttpRequest("GET", "https://localhost"))
+
+    policy.on_request.assert_called_once()
+
+
 def get_completed_future(result=None):
     fut = asyncio.Future()
     fut.set_result(result)

sdk/core/corehttp/tests/test_authentication.py

@@ -15,6 +15,7 @@
     ServiceKeyCredentialPolicy,
 )
 from corehttp.rest import HttpRequest
+from azure.core.pipeline.policies import AzureKeyCredentialPolicy
 import pytest
 
 
@@ -251,6 +252,25 @@ def raise_the_second_time(*args, **kwargs):
     policy.on_exception.assert_called_once_with(policy.request)
 
 
+def test_azure_core_sans_io_policy():
+    """Tests to see that we can use an azure.core SansIOHTTPPolicy with the corehttp Pipeline"""
+
+    class TestPolicy(AzureKeyCredentialPolicy):
+        def __init__(self, *args, **kwargs):
+            super(TestPolicy, self).__init__(*args, **kwargs)
+            self.on_exception = Mock(return_value=False)
+            self.on_request = Mock()
+
+    credential = Mock(get_token=Mock(return_value=AccessToken("***", int(time.time()) + 3600)), key="key")
+    policy = TestPolicy(credential, "scope")
+    transport = Mock(send=Mock(return_value=Mock(status_code=200)))
+
+    pipeline = Pipeline(transport=transport, policies=[policy])
+    pipeline.run(HttpRequest("GET", "https://localhost"))
+
+    policy.on_request.assert_called_once()
+
+
 def test_service_key_credential_policy():
     """Tests to see if we can create an ServiceKeyCredentialPolicy"""