run black on devtools_testutils (Azure#16264)

cleans up the code in the `devtools_testutils` folder

Author: seankane-msft

tools/azure-sdk-tools/devtools_testutils/__init__.py

@@ -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",
 ]

tools/azure-sdk-tools/devtools_testutils/azure_testcase.py

@@ -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:
@@ -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
@@ -34,6 +38,7 @@
 except SyntaxError:
     pass
 
+
 class HttpStatusCode(object):
     OK = 200
     Created = 201
@@ -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
 
@@ -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())
@@ -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,
@@ -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
@@ -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:
@@ -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.
@@ -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()
@@ -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
 
@@ -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):
@@ -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)