[cognitive-services] cog services devtools test framework

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

@@ -1,9 +1,13 @@
 from .mgmt_testcase import (AzureMgmtTestCase, AzureMgmtPreparer)
+from .azure_testcase import AzureTestCase
 from .resource_testcase import (FakeResource, ResourceGroupPreparer)
 from .storage_testcase import (FakeStorageAccount, StorageAccountPreparer)
+from .cognitiveservices_testcase import CognitiveServiceTest, CognitiveServicesAccountPreparer
 
 __all__ = [
     'AzureMgmtTestCase', 'AzureMgmtPreparer',
     'FakeResource', 'ResourceGroupPreparer',
     'FakeStorageAccount', 'StorageAccountPreparer',
+    'CognitiveServiceTest', 'AzureTestCase',
+    'CognitiveServicesAccountPreparer'
 ]

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

@@ -0,0 +1,155 @@
+#-------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+#--------------------------------------------------------------------------
+
+import inspect
+import os.path
+import zlib
+
+from azure_devtools.scenario_tests import (
+    ReplayableTest, AzureTestError,
+    GeneralNameReplacer, RequestUrlNormalizer,
+)
+
+from .config import TEST_SETTING_FILENAME
+from . import mgmt_settings_fake as fake_settings
+
+
+class HttpStatusCode(object):
+    OK = 200
+    Created = 201
+    Accepted = 202
+    NoContent = 204
+    NotFound = 404
+
+
+def get_resource_name(name_prefix, identifier):
+    # Append a suffix to the name, based on the fully qualified test name
+    # We use a checksum of the test name so that each test gets different
+    # 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'):
+        name = name[:-1]
+    return name
+
+
+def get_qualified_method_name(obj, method_name):
+    # example of qualified test 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)
+
+
+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):
+        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)
+        if not os.path.exists(config_file):
+            config_file = None
+        super(AzureTestCase, self).__init__(
+            method_name,
+            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(),
+            replay_processors=replay_processors or self._get_replay_processors(),
+            recording_patches=recording_patches,
+            replay_patches=replay_patches,
+        )
+
+    @property
+    def settings(self):
+        if self.is_live:
+            if self._real_settings:
+                return self._real_settings
+            else:
+                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
+
+    def _get_recording_processors(self):
+        return [
+            self.scrubber,
+            RequestUrlNormalizer()
+        ]
+
+    def _get_replay_processors(self):
+        return [
+            RequestUrlNormalizer()
+        ]
+
+    def is_playback(self):
+        return not self.is_live
+
+
+    def setUp(self):
+        # Every test uses a different resource group name calculated from its
+        # qualified test name.
+        #
+        # When running all tests serially, this allows us to delete
+        # the resource group in teardown without waiting for the delete to
+        # complete. The next test in line will use a different resource group,
+        # so it won't have any trouble creating its resource group even if the
+        # previous test resource group hasn't finished deleting.
+        #
+        # When running tests individually, if you try to run the same test
+        # multiple times in a row, it's possible that the delete in the previous
+        # teardown hasn't completed yet (because we don't wait), and that
+        # would make resource group creation fail.
+        # To avoid that, we also delete the resource group in the
+        # setup, and we wait for that delete to complete.
+        super(AzureTestCase, self).setUp()
+
+    def tearDown(self):
+        return super(AzureTestCase, self).tearDown()
+
+    def create_basic_client(self, client_class, **kwargs):
+        # Whatever the client, if credentials is None, fail
+        with self.assertRaises(ValueError):
+            client = client_class(
+                credentials=None,
+                **kwargs
+            )
+
+        # Real client creation
+        client = client_class(
+            credentials=self.settings.get_credentials(),
+            **kwargs
+        )
+        if self.is_playback():
+            client.config.long_running_operation_timeout = 0
+        client.config.enable_http_logger = True
+        return client
+
+    def create_random_name(self, name):
+        return get_resource_name(name, self.qualified_test_name.encode())
+
+    def get_resource_name(self, name):
+        """Alias to create_random_name for back compatibility."""
+        return self.create_random_name(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('.', '_'))

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

@@ -0,0 +1,88 @@
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+import time
+from collections import namedtuple
+from azure_devtools.scenario_tests import ReplayableTest
+from devtools_testutils import AzureTestCase
+from azure_devtools.scenario_tests.exceptions import AzureTestError
+
+from . import AzureMgmtPreparer, ResourceGroupPreparer
+from .resource_testcase import RESOURCE_GROUP_PARAM
+
+from azure.mgmt.cognitiveservices import CognitiveServicesManagementClient
+from msrest.authentication import CognitiveServicesCredentials
+
+FakeCognitiveServicesAccount = namedtuple(
+    'FakeResource',
+    ['endpoint']
+)
+
+
+class CognitiveServiceTest(AzureTestCase):
+    FILTER_HEADERS = ReplayableTest.FILTER_HEADERS + ['Ocp-Apim-Subscription-Key']
+
+    def __init__(self, method_name):
+        super(CognitiveServiceTest, self).__init__(method_name)
+
+
+class CognitiveServicesAccountPreparer(AzureMgmtPreparer):
+    def __init__(self,
+                 name_prefix='',
+                 sku='S0', location='westus', kind='cognitiveservices',
+                 parameter_name='cognitiveservices_account',
+                 resource_group_parameter_name=RESOURCE_GROUP_PARAM,
+                 disable_recording=True, playback_fake_resource=None,
+                 client_kwargs=None):
+        super(CognitiveServicesAccountPreparer, self).__init__(name_prefix, 24,
+                                                     disable_recording=disable_recording,
+                                                     playback_fake_resource=playback_fake_resource,
+                                                     client_kwargs=client_kwargs)
+        self.location = location
+        self.sku = sku
+        self.kind = kind
+        self.resource_group_parameter_name = resource_group_parameter_name
+        self.parameter_name = parameter_name
+        self.cogsci_key = ''
+
+    def create_resource(self, name, **kwargs):
+        if self.is_live:
+            self.client = self.create_mgmt_client(CognitiveServicesManagementClient)
+            group = self._get_resource_group(**kwargs)
+            cogsci_account = self.client.accounts.create(
+                group.name,
+                name,
+                parameters=
+                {
+                    'sku': {'name': self.sku},
+                    'location': self.location,
+                    'kind': self.kind,
+                    'properties': {}
+                }
+            )
+            time.sleep(5)  # it takes a few seconds to create a cognitive services account
+            self.resource = cogsci_account
+            self.cogsci_key = self.client.accounts.list_keys(group.name, name).key1
+            # FIXME: LuisAuthoringClient and LuisRuntimeClient need authoring key from ARM API (coming soon-ish)
+        else:
+            self.resource = FakeCognitiveServicesAccount("https://{}.api.cognitive.microsoft.com".format(self.location))
+            self.cogsci_key = 'ZmFrZV9hY29jdW50X2tleQ=='
+        return {
+            self.parameter_name: self.resource.endpoint,
+            '{}_key'.format(self.parameter_name): CognitiveServicesCredentials(self.cogsci_key),
+        }
+
+    def remove_resource(self, name, **kwargs):
+        if self.is_live:
+            group = self._get_resource_group(**kwargs)
+            self.client.accounts.delete(group.name, name, polling=False)
+
+    def _get_resource_group(self, **kwargs):
+        try:
+            return kwargs.get(self.resource_group_parameter_name)
+        except KeyError:
+            template = 'To create a storage account a resource group is required. Please add ' \
+                       'decorator @{} in front of this storage account preparer.'
+            raise AzureTestError(template.format(ResourceGroupPreparer.__name__))

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

@@ -14,6 +14,7 @@
     OAuthRequestResponsesFilter, DeploymentNameReplacer,
     RequestUrlNormalizer
 )
+from .azure_testcase import AzureTestCase
 from .config import TEST_SETTING_FILENAME
 from . import mgmt_settings_fake as fake_settings
 
@@ -47,19 +48,12 @@ def get_qualified_method_name(obj, method_name):
     return '{0}.{1}'.format(module_name, method_name)
 
 
-class AzureMgmtTestCase(ReplayableTest):
+class AzureMgmtTestCase(AzureTestCase):
     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):
-        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.region = 'westus'
-        self.scrubber = GeneralNameReplacer()
-        config_file = config_file or os.path.join(self.working_folder, TEST_SETTING_FILENAME)
-        if not os.path.exists(config_file):
-            config_file = None
         super(AzureMgmtTestCase, self).__init__(
             method_name,
             config_file=config_file,
@@ -71,39 +65,6 @@ def __init__(self, method_name, config_file=None,
             replay_patches=replay_patches,
         )
 
-    @property
-    def settings(self):
-        if self.is_live:
-            if self._real_settings:
-                return self._real_settings
-            else:
-                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
-
-    def _get_recording_processors(self):
-        return [
-            self.scrubber,
-            OAuthRequestResponsesFilter(),
-            # DeploymentNameReplacer(), Not use this one, give me full control on deployment name
-            RequestUrlNormalizer()
-        ]
-
-    def _get_replay_processors(self):
-        return [
-            RequestUrlNormalizer()
-        ]
-
-    def is_playback(self):
-        return not self.is_live
-
     def _setup_scrubber(self):
         constants_to_scrub = ['SUBSCRIPTION_ID', 'AD_DOMAIN', 'TENANT_ID', 'CLIENT_OID', 'ADLA_JOB_ID']
         for key in constants_to_scrub:
@@ -133,24 +94,6 @@ def setUp(self):
     def tearDown(self):
         return super(AzureMgmtTestCase, self).tearDown()
 
-    def create_basic_client(self, client_class, **kwargs):
-        # Whatever the client, if credentials is None, fail
-        with self.assertRaises(ValueError):
-            client = client_class(
-                credentials=None,
-                **kwargs
-            )
-
-        # Real client creation
-        client = client_class(
-            credentials=self.settings.get_credentials(),
-            **kwargs
-        )
-        if self.is_playback():
-            client.config.long_running_operation_timeout = 0
-        client.config.enable_http_logger = True
-        return client
-
     def create_mgmt_client(self, client_class, **kwargs):
         # Whatever the client, if subscription_id is None, fail
         with self.assertRaises(ValueError):
@@ -166,20 +109,6 @@ def create_mgmt_client(self, client_class, **kwargs):
             **kwargs
         )
 
-    def create_random_name(self, name):
-        return get_resource_name(name, self.qualified_test_name.encode())
-
-    def get_resource_name(self, name):
-        """Alias to create_random_name for back compatibility."""
-        return self.create_random_name(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('.', '_'))
-
 
 class AzureMgmtPreparer(AbstractPreparer):
     def __init__(self, name_prefix, random_name_length,