migrate identity tests to test proxy

sdk/identity/azure-identity/conftest.py

@@ -5,8 +5,10 @@
 import os
 import sys
 
+from unittest import mock
 import pytest
 import six
+from devtools_testutils import test_proxy, add_general_regex_sanitizer, is_live, add_body_key_sanitizer
 from azure.identity._constants import DEVELOPER_SIGN_ON_CLIENT_ID, EnvironmentVariables
 
 RECORD_IMDS = "--record-imds"
@@ -45,7 +47,7 @@ def record_imds_test(request):
     Recorded IMDS tests run as expected in playback. However, because they require particular live environments, a
     custom pytest option ("--record-imds") controls whether they're included in a live test run.
     """
-    if request.instance.is_live and not request.session.config.getoption(RECORD_IMDS):
+    if is_live() and not request.session.config.getoption(RECORD_IMDS):
         pytest.skip('Run "pytest {}" to record a live run of this test'.format(RECORD_IMDS))
 
 
@@ -159,3 +161,54 @@ def event_loop():
 
     yield loop
     loop.close()
+
+@pytest.fixture(scope="session", autouse=True)
+def add_sanitizers(test_proxy):
+    if EnvironmentVariables.MSI_ENDPOINT in os.environ:
+        url = os.environ.get(EnvironmentVariables.MSI_ENDPOINT)
+        PLAYBACK_URL = "https://msi-endpoint/token"
+        add_general_regex_sanitizer(regex=url, value=PLAYBACK_URL)
+    if "USER_ASSIGNED_IDENTITY_CLIENT_ID" in os.environ:
+        PLAYBACK_CLIENT_ID = "client-id"
+        user_assigned_identity_client_id = os.environ.get("USER_ASSIGNED_IDENTITY_CLIENT_ID")
+        add_general_regex_sanitizer(regex=user_assigned_identity_client_id, value=PLAYBACK_CLIENT_ID)
+    if "CAE_ARM_URL" in os.environ and "CAE_TENANT_ID" in os.environ and "CAE_USERNAME" in os.environ:
+        try:
+            from six.moves.urllib_parse import urlparse
+            arm_url = os.environ["CAE_ARM_URL"]
+            real = urlparse(arm_url)
+            add_general_regex_sanitizer(regex=real.netloc, value="management.azure.com")
+            add_general_regex_sanitizer(regex=os.environ["CAE_TENANT_ID"], value="tenant")
+            add_general_regex_sanitizer(regex=os.environ["CAE_USERNAME"], value="username")
+        except Exception:
+            pass
+    if "OBO_TENANT_ID" in os.environ and "OBO_USERNAME" in os.environ:
+        add_general_regex_sanitizer(regex=os.environ["OBO_TENANT_ID"], value="tenant")
+        add_general_regex_sanitizer(regex=os.environ["OBO_USERNAME"], value="username")
+    add_body_key_sanitizer(json_path="$..access_token", value="access_token")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def patch_async_sleep():
+    async def immediate_return(_):
+        return
+
+    if not is_live():
+        with mock.patch("asyncio.sleep", immediate_return):
+            yield
+
+    else:
+        yield
+
+
+@pytest.fixture(scope="session", autouse=True)
+def patch_sleep():
+    def immediate_return(_):
+        return
+
+    if not is_live():
+        with mock.patch("time.sleep", immediate_return):
+            yield
+
+    else:
+        yield

sdk/identity/azure-identity/tests/recorded_test_case.py

@@ -4,35 +4,20 @@
 # ------------------------------------
 import os
 
-from azure_devtools.scenario_tests import GeneralNameReplacer, patch_time_sleep_api, RequestUrlNormalizer
-from devtools_testutils.azure_testcase import AzureTestCase
+from devtools_testutils import AzureRecordedTestCase, is_live
 import pytest
 
-from recording_processors import IdTokenProcessor, RecordingRedactor
-
 PLAYBACK_CLIENT_ID = "client-id"
 
 
-class RecordedTestCase(AzureTestCase):
-    def __init__(self, *args, **kwargs):
-        scrubber = GeneralNameReplacer()
-        super(RecordedTestCase, self).__init__(
-            *args,
-            recording_processors=[RecordingRedactor(), scrubber],
-            replay_processors=[IdTokenProcessor(), RequestUrlNormalizer()],
-            **kwargs
-        )
-        self.scrubber = scrubber
-        self.replay_patches.append(patch_time_sleep_api)
-        self.user_assigned_identity_client_id = os.environ.get("USER_ASSIGNED_IDENTITY_CLIENT_ID", PLAYBACK_CLIENT_ID)
+class RecordedTestCase(AzureRecordedTestCase):
 
     @pytest.fixture()
     def user_assigned_identity_client_id(self):
-        if self.is_live:
+        self.user_assigned_identity_client_id = os.environ.get("USER_ASSIGNED_IDENTITY_CLIENT_ID", PLAYBACK_CLIENT_ID)
+        if is_live():
             if self.user_assigned_identity_client_id == PLAYBACK_CLIENT_ID:
                 pytest.skip("Set a value for $USER_ASSIGNED_IDENTITY_CLIENT_ID to record this test")
-            else:
-                self.scrubber.register_name_pair(self.user_assigned_identity_client_id, PLAYBACK_CLIENT_ID)
 
     @property
     def scope(self):