[Key Vault] Test keys library against a shared vault

sdk/keyvault/.gitignore

@@ -0,0 +1,3 @@
+*.cer
+*.key
+*.pfx

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

@@ -5,10 +5,12 @@
 import time
 
 from azure_devtools.scenario_tests.patches import patch_time_sleep_api
-from devtools_testutils import AzureMgmtTestCase
+from devtools_testutils import AzureTestCase
 
+from azure.keyvault.keys import KeyClient
 
-class KeyVaultTestCase(AzureMgmtTestCase):
+
+class KeyVaultTestCase(AzureTestCase):
     def __init__(self, *args, **kwargs):
         if "match_body" not in kwargs:
             kwargs["match_body"] = True
@@ -20,6 +22,16 @@ def setUp(self):
         self.list_test_size = 7
         super(KeyVaultTestCase, self).setUp()
 
+    def create_kv_client(self, vault_uri, **kwargs):
+        credential = self.get_credential(KeyClient)
+        return self.create_client_from_credential(
+            KeyClient, credential=credential, vault_url=vault_uri, **kwargs
+        )
+
+    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))
+
     def _poll_until_no_exception(self, fn, expected_exception, max_retries=20, retry_delay=3):
         """polling helper for live tests because some operations take an unpredictable amount of time to complete"""
 

sdk/keyvault/azure-keyvault-keys/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 AzureMgmtTestCase
+from devtools_testutils import AzureTestCase
+
+from azure.keyvault.keys import KeyClient
 
 
 def skip_sleep(unit_test):
@@ -15,7 +17,7 @@ async def immediate_return(_):
     return mock_in_unit_test(unit_test, "asyncio.sleep", immediate_return)
 
 
-class KeyVaultTestCase(AzureMgmtTestCase):
+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)
@@ -24,6 +26,16 @@ def setUp(self):
         self.list_test_size = 7
         super(KeyVaultTestCase, self).setUp()
 
+    def create_kv_client(self, vault_uri, **kwargs):
+        credential = self.get_credential(KeyClient)
+        return self.create_client_from_credential(
+            KeyClient, credential=credential, vault_url=vault_uri, **kwargs
+        )
+
+    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))
+
     async def _poll_until_no_exception(self, fn, expected_exception, max_retries=20, retry_delay=3):
         """polling helper for live tests because some operations take an unpredictable amount of time to complete"""
 

sdk/keyvault/azure-keyvault-keys/tests/test_key_client.py

@@ -11,13 +11,15 @@
 
 from azure.core.exceptions import ResourceExistsError, ResourceNotFoundError
 from azure.keyvault.keys import JsonWebKey, KeyClient
-from devtools_testutils import ResourceGroupPreparer, KeyVaultPreparer
+from devtools_testutils import PowerShellPreparer
 
-from _shared.preparer import KeyVaultClientPreparer as _KeyVaultClientPreparer
 from _shared.test_case import KeyVaultTestCase
 
-# pre-apply the client_cls positional argument so it needn't be explicitly passed below
-KeyVaultClientPreparer = functools.partial(_KeyVaultClientPreparer, KeyClient)
+KeyVaultPreparer = functools.partial(
+    PowerShellPreparer,
+    "keyvault",
+    azure_keyvault_url="https://vaultname.vault.azure.net"
+)
 
 # used for logging tests
 class MockHandler(logging.Handler):
@@ -140,24 +142,26 @@ def _to_bytes(hex):
         self._validate_rsa_key_bundle(imported_key, client.vault_url, name, key.kty, key.key_ops)
         return imported_key
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_key_crud_operations(self, client, **kwargs):
-
+    def test_key_crud_operations(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url)
         self.assertIsNotNone(client)
 
         # create ec key
-        self._create_ec_key(client, key_name="crud-ec-key", hsm=True)
+        ec_key_name = self.get_resource_name("crud-ec-key")
+        self._create_ec_key(client, key_name=ec_key_name, hsm=True)
         # create ec with curve
-        created_ec_key_curve = client.create_ec_key(name="crud-P-256-ec-key", curve="P-256")
+        ec_key_curve_name = self.get_resource_name("crud-P-256-ec-key")
+        created_ec_key_curve = client.create_ec_key(name=ec_key_curve_name, curve="P-256")
         self.assertEqual("P-256", created_ec_key_curve.key.crv)
 
         # import key
-        self._import_test_key(client, "import-test-key")
+        import_test_key_name = self.get_resource_name("import-test-key")
+        self._import_test_key(client, import_test_key_name)
 
         # create rsa key
-        created_rsa_key = self._create_rsa_key(client, key_name="crud-rsa-key")
+        rsa_key_name = self.get_resource_name("crud-rsa-key")
+        created_rsa_key = self._create_rsa_key(client, key_name=rsa_key_name)
 
         # get the created key with version
         key = client.get_key(created_rsa_key.name, created_rsa_key.properties.version)
@@ -190,12 +194,11 @@ def test_key_crud_operations(self, client, **kwargs):
         self.assertIsNotNone(deleted_key)
         self.assertEqual(created_rsa_key.id, deleted_key.id)
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_backup_restore(self, client, **kwargs):
-
+    def test_backup_restore(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url)
         self.assertIsNotNone(client)
+
         key_name = self.get_resource_name("keybak")
         key_type = "RSA"
 
@@ -218,19 +221,17 @@ def test_backup_restore(self, client, **kwargs):
         restored_key = self._poll_until_no_exception(restore_function, ResourceExistsError)
         self._assert_key_attributes_equal(created_bundle.properties, restored_key.properties)
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_key_list(self, client, **kwargs):
-
+    def test_key_list(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url)
         self.assertIsNotNone(client)
 
         max_keys = self.list_test_size
         expected = {}
 
         # create many keys
         for x in range(max_keys):
-            key_name = "key{}".format(x)
+            key_name = self.get_resource_name("key{}".format(x))
             key = client.create_key(key_name, "RSA")
             expected[key.name] = key
 
@@ -242,12 +243,11 @@ def test_key_list(self, client, **kwargs):
                 del expected[key.name]
         self.assertEqual(len(expected), 0)
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_list_versions(self, client, **kwargs):
-
+    def test_list_versions(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url)
         self.assertIsNotNone(client)
+
         key_name = self.get_resource_name("testKey")
 
         max_keys = self.list_test_size
@@ -268,17 +268,16 @@ def test_list_versions(self, client, **kwargs):
                 self._assert_key_attributes_equal(expected_key.properties, key)
         self.assertEqual(0, len(expected))
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_list_deleted_keys(self, client, **kwargs):
+    def test_list_deleted_keys(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url)
         self.assertIsNotNone(client)
 
         expected = {}
 
         # create keys
         for i in range(self.list_test_size):
-            key_name = "key{}".format(i)
+            key_name = self.get_resource_name("key{}".format(i))
             key_value = "value{}".format(i)
             expected[key_name] = client.create_key(key_name, "RSA")
 
@@ -299,16 +298,15 @@ def test_list_deleted_keys(self, client, **kwargs):
                 self._assert_key_attributes_equal(expected[key.name].properties, key.properties)
                 del expected[key.name]
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_recover(self, client, **kwargs):
+    def test_recover(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url)
         self.assertIsNotNone(client)
 
         # create keys
         keys = {}
         for i in range(self.list_test_size):
-            key_name = "key{}".format(i)
+            key_name = self.get_resource_name("key{}".format(i))
             keys[key_name] = client.create_key(key_name, "RSA")
 
         # delete them
@@ -325,14 +323,13 @@ def test_recover(self, client, **kwargs):
             expected_key = keys[key_name]
             self._assert_key_attributes_equal(expected_key.properties, recovered_key.properties)
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_purge(self, client, **kwargs):
+    def test_purge(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url)
         self.assertIsNotNone(client)
 
         # create keys
-        key_names = ["key{}".format(i) for i in range(self.list_test_size)]
+        key_names = [self.get_resource_name("key{}".format(i)) for i in range(self.list_test_size)]
         for name in key_names:
             client.create_key(name, "RSA")
 
@@ -356,17 +353,17 @@ def test_purge(self, client, **kwargs):
         deleted = [s.name for s in client.list_deleted_keys()]
         self.assertTrue(not any(s in deleted for s in key_names))
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer(client_kwargs={"logging_enable": True})
-    def test_logging_enabled(self, client, **kwargs):
+    def test_logging_enabled(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url, logging_enable=True)
         mock_handler = MockHandler()
 
         logger = logging.getLogger("azure")
         logger.addHandler(mock_handler)
         logger.setLevel(logging.DEBUG)
 
-        client.create_rsa_key("rsa-key-name", size=2048)
+        rsa_key_name = self.get_resource_name("rsa-key-name")
+        client.create_rsa_key(rsa_key_name, size=2048)
 
         for message in mock_handler.messages:
             if message.levelname == "DEBUG" and message.funcName == "on_request":
@@ -380,17 +377,17 @@ def test_logging_enabled(self, client, **kwargs):
 
         assert False, "Expected request body wasn't logged"
 
-    @ResourceGroupPreparer(random_name_enabled=True)
     @KeyVaultPreparer()
-    @KeyVaultClientPreparer()
-    def test_logging_disabled(self, client, **kwargs):
+    def test_logging_disabled(self, azure_keyvault_url, **kwargs):
+        client = self.create_kv_client(azure_keyvault_url, logging_enable=False)
         mock_handler = MockHandler()
 
         logger = logging.getLogger("azure")
         logger.addHandler(mock_handler)
         logger.setLevel(logging.DEBUG)
 
-        client.create_rsa_key("rsa-key-name", size=2048)
+        rsa_key_name = self.get_resource_name("rsa-key-name")
+        client.create_rsa_key(rsa_key_name, size=2048)
 
         for message in mock_handler.messages:
             if message.levelname == "DEBUG" and message.funcName == "on_request":

sdk/keyvault/test-resources-cleanup.py

@@ -0,0 +1,47 @@
+import os
+
+from azure.identity import ClientSecretCredential
+from azure.keyvault.certificates import CertificateClient
+from azure.keyvault.keys import KeyClient
+from azure.keyvault.secrets import SecretClient
+
+if "AZURE_KEYVAULT_URL" not in os.environ:
+    raise EnvironmentError("Missing a Key Vault URL")
+if "KEYVAULT_TENANT_ID" not in os.environ:
+    raise EnvironmentError("Missing a tenant ID for Key Vault")
+if "KEYVAULT_CLIENT_ID" not in os.environ:
+    raise EnvironmentError("Missing a client ID for Key Vault")
+if "KEYVAULT_CLIENT_SECRET" not in os.environ:
+    raise EnvironmentError("Missing a client secret for Key Vault")
+
+credential = ClientSecretCredential(
+    tenant_id=os.environ.get("KEYVAULT_TENANT_ID"),
+    client_id=os.environ.get("KEYVAULT_CLIENT_ID"),
+    client_secret=os.environ.get("KEYVAULT_CLIENT_SECRET")
+)
+
+if "AZURE_KEYVAULT_URL" in os.environ:
+    cert_client = CertificateClient(os.environ.get("AZURE_KEYVAULT_URL"), credential)
+    key_client = KeyClient(os.environ.get("AZURE_KEYVAULT_URL"), credential)
+    secret_client = SecretClient(os.environ.get("AZURE_KEYVAULT_URL"), credential)
+
+    test_certificates = [c for c in cert_client.list_properties_of_certificates() if c.name.startswith("livekvtest")]
+    for certificate in test_certificates:
+        cert_client.begin_delete_certificate(certificate.name).wait()
+    deleted_test_certificates = [c for c in cert_client.list_deleted_certificates() if c.name.startswith("livekvtest")]
+    for certificate in deleted_test_certificates:
+        cert_client.purge_deleted_certificate(certificate.name)
+
+    test_keys = [k for k in key_client.list_properties_of_keys() if k.name.startswith("livekvtest")]
+    for key in test_keys:
+        key_client.begin_delete_key(key.name).wait()
+    deleted_test_keys = [k for k in key_client.list_deleted_keys() if k.name.startswith("livekvtest")]
+    for key in deleted_test_keys:
+        key_client.purge_deleted_key(key.name)
+
+    test_secrets = [s for s in secret_client.list_properties_of_secrets() if s.name.startswith("livekvtest")]
+    for secret in test_secrets:
+        secret_client.begin_delete_secret(secret.name).wait()
+    deleted_test_secrets = [s for s in secret_client.list_deleted_secrets() if s.name.startswith("livekvtest")]
+    for secret in deleted_test_secrets:
+        secret_client.purge_deleted_secret(secret.name)