Skip to content
Merged
Show file tree
Hide file tree
Changes from 16 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 27 additions & 1 deletion sdk/identity/azure-identity/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

import pytest
import six
from devtools_testutils import test_proxy, add_general_regex_sanitizer, is_live
from azure.identity._constants import DEVELOPER_SIGN_ON_CLIENT_ID, EnvironmentVariables

RECORD_IMDS = "--record-imds"
Expand Down Expand Up @@ -45,7 +46,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))


Expand Down Expand Up @@ -159,3 +160,28 @@ 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:
Comment thread
xiangyan99 marked this conversation as resolved.
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.get("CAE_USERNAME"), value="username")
Comment thread
xiangyan99 marked this conversation as resolved.
Outdated
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")

23 changes: 4 additions & 19 deletions sdk/identity/azure-identity/tests/recorded_test_case.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Comment thread
xiangyan99 marked this conversation as resolved.
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):
Expand Down

This file was deleted.

This file was deleted.

This file was deleted.

This file was deleted.

Loading