[KeyVault] KV Certificates to test proxy

sdk/keyvault/azure-keyvault-certificates/tests/_async_test_case.py

@@ -0,0 +1,58 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+
+import os
+
+from azure.keyvault.certificates import ApiVersion
+from azure.keyvault.certificates._shared.client_base import DEFAULT_VERSION
+from devtools_testutils import AzureRecordedTestCase, is_live
+import pytest
+
+
+def get_decorator(**kwargs):
+    """returns a test decorator for test parameterization"""
+    versions = kwargs.pop("api_versions", None) or ApiVersion
+    params = [pytest.param(api_version) for api_version in versions]
+    return params
+
+
+class AsyncCertificatesClientPreparer(AzureRecordedTestCase):
+    def __init__(self, **kwargs) -> None:
+        self.azure_keyvault_url = "https://vaultname.vault.azure.net"
+
+        if is_live():
+            self.azure_keyvault_url = os.environ["AZURE_KEYVAULT_URL"]
+
+        self.is_logging_enabled = kwargs.pop("logging_enable", True)
+
+        if is_live():
+            os.environ["AZURE_TENANT_ID"] = os.environ["KEYVAULT_TENANT_ID"]
+            os.environ["AZURE_CLIENT_ID"] = os.environ["KEYVAULT_CLIENT_ID"]
+            os.environ["AZURE_CLIENT_SECRET"] = os.environ["KEYVAULT_CLIENT_SECRET"]
+
+    def __call__(self, fn):
+        async def _preparer(test_class, api_version, **kwargs):
+
+            self._skip_if_not_configured(api_version)
+            if not self.is_logging_enabled:
+                kwargs.update({"logging_enable": False})
+            client = self.create_client(self.azure_keyvault_url, api_version=api_version, **kwargs)
+
+            async with client:
+                await fn(test_class, client)
+        return _preparer
+
+    def create_client(self, vault_uri, **kwargs):
+        from azure.keyvault.certificates.aio import CertificateClient
+
+        credential = self.get_credential(CertificateClient, is_async = True)
+
+        return self.create_client_from_credential(
+            CertificateClient, credential=credential, vault_url=vault_uri, **kwargs
+        )
+
+    def _skip_if_not_configured(self, api_version, **kwargs):
+        if self.is_live and api_version != DEFAULT_VERSION:
+            pytest.skip("This test only uses the default API version for live tests")

sdk/keyvault/azure-keyvault-certificates/tests/_shared/test_case.py

@@ -3,23 +3,11 @@
 # Licensed under the MIT License.
 # ------------------------------------
 import time
+from azure.keyvault.certificates._shared import HttpChallengeCache
+from devtools_testutils import AzureRecordedTestCase
 
-from azure_devtools.scenario_tests.patches import patch_time_sleep_api
-from devtools_testutils import AzureTestCase
-
-
-class KeyVaultTestCase(AzureTestCase):
-    def __init__(self, *args, **kwargs):
-        if "match_body" not in kwargs:
-            kwargs["match_body"] = True
-
-        super(KeyVaultTestCase, self).__init__(*args, **kwargs)
-        self.replay_patches.append(patch_time_sleep_api)
-
-    def setUp(self):
-        self.list_test_size = 7
-        super(KeyVaultTestCase, self).setUp()
 
+class KeyVaultTestCase(AzureRecordedTestCase):
     def get_resource_name(self, name):
         """helper to create resources with a consistent, test-indicative prefix"""
         return super(KeyVaultTestCase, self).get_resource_name("livekvtest{}".format(name))
@@ -48,3 +36,7 @@ def _poll_until_exception(self, fn, expected_exception, max_retries=20, retry_de
                 return
 
         self.fail("expected exception {expected_exception} was not raised")
+
+    def tear_down(self):
+        HttpChallengeCache.clear()
+        assert len(HttpChallengeCache._cache) == 0

sdk/keyvault/azure-keyvault-certificates/tests/_shared/test_case_async.py

@@ -5,7 +5,9 @@
 import asyncio
 
 from azure_devtools.scenario_tests.patches import mock_in_unit_test
-from devtools_testutils import AzureTestCase
+from devtools_testutils import AzureRecordedTestCase
+
+from azure.keyvault.certificates._shared import HttpChallengeCache
 
 
 def skip_sleep(unit_test):
@@ -15,15 +17,7 @@ async def immediate_return(_):
     return mock_in_unit_test(unit_test, "asyncio.sleep", immediate_return)
 
 
-class KeyVaultTestCase(AzureTestCase):
-    def __init__(self, *args, match_body=True, **kwargs):
-        super().__init__(*args, match_body=match_body, **kwargs)
-        self.replay_patches.append(skip_sleep)
-
-    def setUp(self):
-        self.list_test_size = 7
-        super(KeyVaultTestCase, self).setUp()
-
+class KeyVaultTestCase(AzureRecordedTestCase):
     def get_resource_name(self, name):
         """helper to create resources with a consistent, test-indicative prefix"""
         return super(KeyVaultTestCase, self).get_resource_name("livekvtest{}".format(name))
@@ -51,3 +45,8 @@ async def _poll_until_exception(self, fn, expected_exception, max_retries=20, re
             except expected_exception:
                 return
         self.fail("expected exception {expected_exception} was not raised")
+
+    def tear_down(self):
+        HttpChallengeCache.clear()
+        assert len(HttpChallengeCache._cache) == 0
+

sdk/keyvault/azure-keyvault-certificates/tests/_test_case.py

@@ -2,59 +2,53 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-import functools
+import os
 
 from azure.keyvault.certificates import ApiVersion
-from azure.keyvault.certificates._shared import HttpChallengeCache
 from azure.keyvault.certificates._shared.client_base import DEFAULT_VERSION
-from devtools_testutils import AzureTestCase, PowerShellPreparer
-from parameterized import parameterized, param
+from devtools_testutils import AzureRecordedTestCase, is_live
 import pytest
 
 
-def client_setup(testcase_func):
-    """decorator that creates a client to be passed in to a test method"""
-    @PowerShellPreparer("keyvault", azure_keyvault_url="https://vaultname.vault.azure.net")
-    @functools.wraps(testcase_func)
-    def wrapper(test_class_instance, azure_keyvault_url, api_version, **kwargs):
-        test_class_instance._skip_if_not_configured(api_version)
-        client = test_class_instance.create_client(azure_keyvault_url, api_version=api_version, **kwargs)
-
-        if kwargs.get("is_async"):
-            import asyncio
-
-            coroutine = testcase_func(test_class_instance, client)
-            loop = asyncio.get_event_loop()
-            loop.run_until_complete(coroutine)
-        else:
-            testcase_func(test_class_instance, client)
-    return wrapper
-
 
 def get_decorator(**kwargs):
     """returns a test decorator for test parameterization"""
     versions = kwargs.pop("api_versions", None) or ApiVersion
-    params = [param(api_version=api_version, **kwargs) for api_version in versions]
-    return functools.partial(parameterized.expand, params, name_func=suffixed_test_name)
-
-
-def suffixed_test_name(testcase_func, param_num, param):
-    return "{}_{}".format(testcase_func.__name__, parameterized.to_safe_name(param.kwargs.get("api_version")))
-
-
-class CertificatesTestCase(AzureTestCase):
-    def tearDown(self):
-        HttpChallengeCache.clear()
-        assert len(HttpChallengeCache._cache) == 0
-        super(CertificatesTestCase, self).tearDown()
-
+    params = [pytest.param(api_version) for api_version in versions]
+    return params
+
+
+class CertificatesClientPreparer(AzureRecordedTestCase):
+    def __init__(self, **kwargs) -> None:
+        self.azure_keyvault_url = "https://vaultname.vault.azure.net"
+
+        if is_live():
+            self.azure_keyvault_url = os.environ["AZURE_KEYVAULT_URL"]
+
+        self.is_logging_enabled = kwargs.pop("logging_enable", True)
+
+        if is_live():
+            os.environ["AZURE_TENANT_ID"] = os.environ["KEYVAULT_TENANT_ID"]
+            os.environ["AZURE_CLIENT_ID"] = os.environ["KEYVAULT_CLIENT_ID"]
+            os.environ["AZURE_CLIENT_SECRET"] = os.environ["KEYVAULT_CLIENT_SECRET"]
+
+    def __call__(self, fn):
+        def _preparer(test_class, api_version, **kwargs):
+
+            self._skip_if_not_configured(api_version)
+            if not self.is_logging_enabled:
+                kwargs.update({"logging_enable": False})
+            client = self.create_client(self.azure_keyvault_url, api_version=api_version, **kwargs)
+
+            with client:
+                fn(test_class, client)
+        return _preparer
+
     def create_client(self, vault_uri, **kwargs):
-        if kwargs.pop("is_async", False):
-            from azure.keyvault.certificates.aio import CertificateClient
-            credential = self.get_credential(CertificateClient, is_async=True)
-        else:
-            from azure.keyvault.certificates import CertificateClient
-            credential = self.get_credential(CertificateClient)
+        from azure.keyvault.certificates import CertificateClient
+
+        credential = self.get_credential(CertificateClient)
+
         return self.create_client_from_credential(
             CertificateClient, credential=credential, vault_url=vault_uri, **kwargs
         )

sdk/keyvault/azure-keyvault-certificates/tests/conftest.py

@@ -0,0 +1,75 @@
+# --------------------------------------------------------------------------
+#
+# Copyright (c) Microsoft Corporation. All rights reserved.
+#
+# The MIT License (MIT)
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the ""Software""), to
+# deal in the Software without restriction, including without limitation the
+# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+# sell copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+# IN THE SOFTWARE.
+#
+# --------------------------------------------------------------------------
+import asyncio
+import os
+import pytest
+from unittest import mock
+from devtools_testutils import is_live, test_proxy, add_oauth_response_sanitizer, add_general_regex_sanitizer
+
+
+@pytest.fixture(scope="session", autouse=True)
+def add_sanitizers(test_proxy):
+    azure_keyvault_url = os.getenv("azure_keyvault_url", "https://vaultname.vault.azure.net")
+    azure_keyvault_url = azure_keyvault_url.rstrip("/")
+    keyvault_tenant_id = os.getenv("keyvault_tenant_id", "keyvault_tenant_id")
+    keyvault_subscription_id = os.getenv("keyvault_subscription_id", "keyvault_subscription_id")
+
+    add_general_regex_sanitizer(regex=azure_keyvault_url, value="https://vaultname.vault.azure.net")
+    add_general_regex_sanitizer(regex=keyvault_tenant_id, value="00000000-0000-0000-0000-000000000000")
+    add_general_regex_sanitizer(regex=keyvault_subscription_id, value="00000000-0000-0000-0000-000000000000")
+    add_oauth_response_sanitizer()
+
+
+@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
+
+@pytest.fixture(scope="session")
+def event_loop(request):
+    loop = asyncio.get_event_loop()
+    yield loop
+    loop.close()