Skip to content

Commit

Permalink
run black on devtools_testutils (#16264)
Browse files Browse the repository at this point in the history 
cleans up the code in the `devtools_testutils` folder
  • Loading branch information
@seankane-msft
seankane-msft authored Jan 20, 2021
1 parent a78be2e commit ce3bfd5
Show file tree
Hide file tree
Showing 12 changed files with 492 additions and 337 deletions.
36 changes: 26 additions & 10 deletions tools/azure-sdk-tools/devtools_testutils/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,16 +1,32 @@
from .mgmt_testcase import (AzureMgmtTestCase, AzureMgmtPreparer)
from .mgmt_testcase import AzureMgmtTestCase, AzureMgmtPreparer
from .azure_testcase import AzureTestCase, is_live, get_region_override
from .resource_testcase import (FakeResource, ResourceGroupPreparer, RandomNameResourceGroupPreparer, CachedResourceGroupPreparer)
from .storage_testcase import FakeStorageAccount, StorageAccountPreparer, CachedStorageAccountPreparer
from .resource_testcase import (
FakeResource,
ResourceGroupPreparer,
RandomNameResourceGroupPreparer,
CachedResourceGroupPreparer,
)
from .storage_testcase import (
FakeStorageAccount,
StorageAccountPreparer,
CachedStorageAccountPreparer,
)
from .keyvault_preparer import KeyVaultPreparer
from .powershell_preparer import PowerShellPreparer

__all__ = [
'AzureMgmtTestCase', 'AzureMgmtPreparer',
'FakeResource', 'ResourceGroupPreparer',
'StorageAccountPreparer', 'CachedStorageAccountPreparer',
'FakeStorageAccount',
'AzureTestCase', 'is_live', 'get_region_override',
'KeyVaultPreparer', 'RandomNameResourceGroupPreparer',
'CachedResourceGroupPreparer', 'PowerShellPreparer'
"AzureMgmtTestCase",
"AzureMgmtPreparer",
"FakeResource",
"ResourceGroupPreparer",
"StorageAccountPreparer",
"CachedStorageAccountPreparer",
"FakeStorageAccount",
"AzureTestCase",
"is_live",
"get_region_override",
"KeyVaultPreparer",
"RandomNameResourceGroupPreparer",
"CachedResourceGroupPreparer",
"PowerShellPreparer",
]
144 changes: 86 additions & 58 deletions tools/azure-sdk-tools/devtools_testutils/azure_testcase.py
View file
Original file line number Diff line number Diff line change
@@ -1,14 +1,15 @@
#-------------------------------------------------------------------------
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
# --------------------------------------------------------------------------
import functools
import inspect
import logging
import os.path
import sys
import zlib

try:
from inspect import getfullargspec as get_arg_spec
except ImportError:
Expand All @@ -18,9 +19,12 @@
from dotenv import load_dotenv, find_dotenv

from azure_devtools.scenario_tests import (
ReplayableTest, AzureTestError,
GeneralNameReplacer, RequestUrlNormalizer,
AuthenticationMetadataFilter, OAuthRequestResponsesFilter
ReplayableTest,
AzureTestError,
GeneralNameReplacer,
RequestUrlNormalizer,
AuthenticationMetadataFilter,
OAuthRequestResponsesFilter,
)
from azure_devtools.scenario_tests.config import TestConfig
from azure_devtools.scenario_tests.utilities import trim_kwargs_from_test_function
Expand All @@ -34,6 +38,7 @@
except SyntaxError:
pass


class HttpStatusCode(object):
OK = 200
Created = 201
Expand All @@ -48,9 +53,9 @@ def get_resource_name(name_prefix, identifier):
# resource names, but each test will get the same name on repeat runs,
# which is needed for playback.
# Most resource names have a length limit, so we use a crc32
checksum = zlib.adler32(identifier) & 0xffffffff
name = '{}{}'.format(name_prefix, hex(checksum)[2:]).rstrip('L')
if name.endswith('L'):
checksum = zlib.adler32(identifier) & 0xFFFFFFFF
name = "{}{}".format(name_prefix, hex(checksum)[2:]).rstrip("L")
if name.endswith("L"):
name = name[:-1]
return name

Expand All @@ -60,46 +65,56 @@ def get_qualified_method_name(obj, method_name):
# test_mgmt_network.test_public_ip_addresses
_, filename = os.path.split(inspect.getsourcefile(type(obj)))
module_name, _ = os.path.splitext(filename)
return '{0}.{1}'.format(module_name, method_name)
return "{0}.{1}".format(module_name, method_name)


def is_live():
"""A module version of is_live, that could be used in pytest marker.
"""
if not hasattr(is_live, '_cache'):
"""A module version of is_live, that could be used in pytest marker."""
if not hasattr(is_live, "_cache"):
config_file = os.path.join(os.path.dirname(__file__), TEST_SETTING_FILENAME)
if not os.path.exists(config_file):
config_file = None
is_live._cache = TestConfig(config_file=config_file).record_mode
return is_live._cache


def get_region_override(default='westus'):
region = os.environ.get('RESOURCE_REGION', None) or default
def get_region_override(default="westus"):
region = os.environ.get("RESOURCE_REGION", None) or default
if not region:
raise ValueError('Region should not be None; set a non-empty-string region to either the RESOURCE_REGION environment variable or the default parameter to this function.')
raise ValueError(
"Region should not be None; set a non-empty-string region to either the RESOURCE_REGION environment variable or the default parameter to this function."
)
return region


def _is_autorest_v3(client_class):
""" IS this client a autorestv3/track2 one?.
"""IS this client a autorestv3/track2 one?.
Could be refined later if necessary.
"""
args = get_arg_spec(client_class.__init__).args
return "credential" in args


class AzureTestCase(ReplayableTest):
def __init__(self, method_name, config_file=None,
recording_dir=None, recording_name=None,
recording_processors=None, replay_processors=None,
recording_patches=None, replay_patches=None,
**kwargs):
def __init__(
self,
method_name,
config_file=None,
recording_dir=None,
recording_name=None,
recording_processors=None,
replay_processors=None,
recording_patches=None,
replay_patches=None,
**kwargs
):
self.working_folder = os.path.dirname(__file__)
self.qualified_test_name = get_qualified_method_name(self, method_name)
self._fake_settings, self._real_settings = self._load_settings()
self.scrubber = GeneralNameReplacer()
config_file = config_file or os.path.join(self.working_folder, TEST_SETTING_FILENAME)
config_file = config_file or os.path.join(
self.working_folder, TEST_SETTING_FILENAME
)
if not os.path.exists(config_file):
config_file = None
load_dotenv(find_dotenv())
Expand All @@ -108,7 +123,8 @@ def __init__(self, method_name, config_file=None,
config_file=config_file,
recording_dir=recording_dir,
recording_name=recording_name or self.qualified_test_name,
recording_processors=recording_processors or self._get_recording_processors(),
recording_processors=recording_processors
or self._get_recording_processors(),
replay_processors=replay_processors or self._get_replay_processors(),
recording_patches=recording_patches,
replay_patches=replay_patches,
Expand All @@ -121,13 +137,16 @@ def settings(self):
if self._real_settings:
return self._real_settings
else:
raise AzureTestError('Need a mgmt_settings_real.py file to run tests live.')
raise AzureTestError(
"Need a mgmt_settings_real.py file to run tests live."
)
else:
return self._fake_settings

def _load_settings(self):
try:
from . import mgmt_settings_real as real_settings

return fake_settings, real_settings
except ImportError:
return fake_settings, None
Expand All @@ -137,22 +156,28 @@ def _get_recording_processors(self):
self.scrubber,
AuthenticationMetadataFilter(),
OAuthRequestResponsesFilter(),
RequestUrlNormalizer()
RequestUrlNormalizer(),
]

def _get_replay_processors(self):
return [
RequestUrlNormalizer()
]
return [RequestUrlNormalizer()]

def is_playback(self):
return not self.is_live

def get_settings_value(self, key):
key_value = os.environ.get("AZURE_"+key, None)

if key_value and self._real_settings and getattr(self._real_settings, key) != key_value:
raise ValueError("You have both AZURE_{key} env variable and mgmt_settings_real.py for {key} to different values".format(key=key))
key_value = os.environ.get("AZURE_" + key, None)

if (
key_value
and self._real_settings
and getattr(self._real_settings, key) != key_value
):
raise ValueError(
"You have both AZURE_{key} env variable and mgmt_settings_real.py for {key} to different values".format(
key=key
)
)

if not key_value:
try:
Expand All @@ -170,7 +195,6 @@ def set_value_to_scrub(self, key, default_value):
else:
return default_value


def setUp(self):
# Every test uses a different resource group name calculated from its
# qualified test name.
Expand All @@ -194,35 +218,43 @@ def tearDown(self):

def get_credential(self, client_class, **kwargs):

tenant_id = os.environ.get("AZURE_TENANT_ID", getattr(self._real_settings, "TENANT_ID", None))
client_id = os.environ.get("AZURE_CLIENT_ID", getattr(self._real_settings, "CLIENT_ID", None))
secret = os.environ.get("AZURE_CLIENT_SECRET", getattr(self._real_settings, "CLIENT_SECRET", None))
tenant_id = os.environ.get(
"AZURE_TENANT_ID", getattr(self._real_settings, "TENANT_ID", None)
)
client_id = os.environ.get(
"AZURE_CLIENT_ID", getattr(self._real_settings, "CLIENT_ID", None)
)
secret = os.environ.get(
"AZURE_CLIENT_SECRET", getattr(self._real_settings, "CLIENT_SECRET", None)
)
is_async = kwargs.pop("is_async", False)

if tenant_id and client_id and secret and self.is_live:
if _is_autorest_v3(client_class):
# Create azure-identity class
from azure.identity import ClientSecretCredential

if is_async:
from azure.identity.aio import ClientSecretCredential
return ClientSecretCredential(
tenant_id=tenant_id,
client_id=client_id,
client_secret=secret
tenant_id=tenant_id, client_id=client_id, client_secret=secret
)
else:
# Create msrestazure class
from msrestazure.azure_active_directory import ServicePrincipalCredentials
from msrestazure.azure_active_directory import (
ServicePrincipalCredentials,
)

return ServicePrincipalCredentials(
tenant=tenant_id,
client_id=client_id,
secret=secret
tenant=tenant_id, client_id=client_id, secret=secret
)
else:
if _is_autorest_v3(client_class):
if is_async:
if self.is_live:
raise ValueError("Async live doesn't support mgmt_setting_real, please set AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET")
raise ValueError(
"Async live doesn't support mgmt_setting_real, please set AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET"
)
return AsyncFakeCredential()
else:
return self.settings.get_azure_core_credentials()
Expand All @@ -237,19 +269,15 @@ def create_client_from_credential(self, client_class, credential, **kwargs):
# kwargs.setdefault("polling_interval", 0)
if _is_autorest_v3(client_class):
kwargs.setdefault("logging_enable", True)
client = client_class(
credential=credential,
**kwargs
)
client = client_class(credential=credential, **kwargs)
else:
client = client_class(
credentials=credential,
**kwargs
)
client = client_class(credentials=credential, **kwargs)

if self.is_playback():
try:
client._config.polling_interval = 0 # FIXME in azure-mgmt-core, make this a kwargs
client._config.polling_interval = (
0 # FIXME in azure-mgmt-core, make this a kwargs
)
except AttributeError:
pass

Expand Down Expand Up @@ -281,18 +309,17 @@ def get_replayable_random_resource_name(self, name):
"""In a replay scenario, (is not live) gives the static moniker. In the random scenario, gives generated name."""
if self.is_live:
created_name = self.create_random_name(name)
self.scrubber.register_name_pair(
created_name,
name
)
self.scrubber.register_name_pair(created_name, name)
return name

def get_preparer_resource_name(self, prefix):
"""Random name generation for use by preparers.
If prefix is a blank string, use the fully qualified test name instead.
This is what legacy tests do for resource groups."""
return self.get_resource_name(prefix or self.qualified_test_name.replace('.', '_'))
return self.get_resource_name(
prefix or self.qualified_test_name.replace(".", "_")
)

@staticmethod
def await_prepared_test(test_fn):
Expand All @@ -308,6 +335,7 @@ def await_prepared_test(test_fn):
raise ImportError("Async wrapper is not needed for Python 2.7 code.")

import asyncio

@functools.wraps(test_fn)
def run(test_class_instance, *args, **kwargs):
trim_kwargs_from_test_function(test_fn, kwargs)
Expand Down
Loading

0 comments on commit ce3bfd5

Please sign in to comment.