add tests

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

@@ -49,7 +49,7 @@ def is_http_policy(policy) -> TypeGuard[HTTPPolicy]:
     return False
 
 def is_sansio_http_policy(policy) -> TypeGuard[SansIOHTTPPolicy]:
-    if hasattr(policy, "on_request") or hasattr(policy, "on_response"):
+    if hasattr(policy, "on_request") and hasattr(policy, "on_response"):
         return True
     return False
 

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
@@ -237,6 +238,23 @@ async def fake_send(*args, **kwargs):
     policy.on_challenge.assert_called_once()
     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()

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
 
 
@@ -250,6 +251,23 @@ def raise_the_second_time(*args, **kwargs):
     policy.on_challenge.assert_called_once()
     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"""