Added tests

Author: Nicola Camillucci

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

@@ -22,15 +22,18 @@ class AsyncKeysClientPreparer(AzureRecordedTestCase):
     def __init__(self, *args, **kwargs):
         vault_playback_url = "https://vaultname.vault.azure.net"
         hsm_playback_url = "https://managedhsmvaultname.managedhsm.azure.net"
+        playback_ekm_external_id = "fake-ekm-host"
         self.is_logging_enabled = kwargs.pop("logging_enable", True)
 
         if self.is_live:
             self.vault_url = os.environ["AZURE_KEYVAULT_URL"]
             hsm = os.environ.get("AZURE_MANAGEDHSM_URL")
             self.managed_hsm_url = hsm if hsm else None
+            self.ekm_external_id = os.getenv("EKM_EXTERNAL_ID")
         else:
             self.vault_url = vault_playback_url
             self.managed_hsm_url = hsm_playback_url
+            self.ekm_external_id = playback_ekm_external_id
 
         self._set_mgmt_settings_real_values()
 
@@ -44,7 +47,12 @@ async def _preparer(test_class, api_version, is_hsm, **kwargs):
             client = self.create_key_client(endpoint_url, api_version=api_version, **kwargs)
             async with client:
                 await fn(
-                    test_class, client, is_hsm=is_hsm, managed_hsm_url=self.managed_hsm_url, vault_url=self.vault_url
+                    test_class,
+                    client,
+                    is_hsm=is_hsm,
+                    managed_hsm_url=self.managed_hsm_url,
+                    vault_url=self.vault_url,
+                    ekm_external_id=self.ekm_external_id,
                 )
 
         return _preparer

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

@@ -61,6 +61,7 @@ class KeysClientPreparer(AzureRecordedTestCase):
     def __init__(self, *args, **kwargs):
         vault_playback_url = "https://vaultname.vault.azure.net"
         hsm_playback_url = "https://managedhsmvaultname.managedhsm.azure.net"
+        playback_ekm_external_id = "fake-ekm-host"
         self.is_logging_enabled = kwargs.pop("logging_enable", True)
 
         if self.is_live:
@@ -70,9 +71,11 @@ def __init__(self, *args, **kwargs):
             self.managed_hsm_url = hsm if hsm else None
             if self.managed_hsm_url:
                 self.managed_hsm_url = self.managed_hsm_url.rstrip("/")
+            self.ekm_external_id = os.getenv("EKM_EXTERNAL_ID")
         else:
             self.vault_url = vault_playback_url
             self.managed_hsm_url = hsm_playback_url
+            self.ekm_external_id = playback_ekm_external_id
 
         self._set_mgmt_settings_real_values()
 
@@ -86,7 +89,14 @@ def _preparer(test_class, api_version, is_hsm, **kwargs):
             client = self.create_key_client(endpoint_url, api_version=api_version, **kwargs)
 
             with client:
-                fn(test_class, client, is_hsm=is_hsm, managed_hsm_url=self.managed_hsm_url, vault_url=self.vault_url)
+                fn(
+                    test_class,
+                    client,
+                    is_hsm=is_hsm,
+                    managed_hsm_url=self.managed_hsm_url,
+                    vault_url=self.vault_url,
+                    ekm_external_id=self.ekm_external_id,
+                )
 
         return _preparer
 

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

@@ -17,6 +17,7 @@
 from azure.core.rest import HttpRequest
 from azure.keyvault.keys import (
     ApiVersion,
+    ExternalKey,
     JsonWebKey,
     KeyClient,
     KeyProperties,
@@ -807,6 +808,33 @@ def test_send_request(self, client, is_hsm, **kwargs):
         response = client.send_request(request)
         assert response.json()["key"]["kid"] == key.id
 
+    @pytest.mark.parametrize("api_version,is_hsm", only_hsm_default)
+    @KeysClientPreparer()
+    @recorded_by_proxy
+    def test_create_external_key(self, client, **kwargs):
+        """Register an external HSM key and verify the external_key reference round-trips."""
+        external_id = kwargs.pop("ekm_external_id")
+        if not external_id:
+            pytest.skip(
+                "No external key ID provided. This test requires an EKM-connected HSM and an existing external key."
+            )
+
+        key_name = self.get_resource_name("ext-key")
+        external_key = ExternalKey(id=external_id)
+
+        created = client.create_external_key(key_name, external_key=external_key)
+        assert created is not None
+        assert created.name == key_name
+        assert created.properties.external_key is not None
+        assert created.properties.external_key.id == external_id
+        assert created.key_type is not None
+
+        # Verify the external_key reference is also returned by a subsequent get_key.
+        fetched = client.get_key(key_name)
+        assert fetched.properties.external_key is not None
+        assert fetched.properties.external_key.id == external_id
+        assert fetched.key_type is not None
+
     @pytest.mark.parametrize("api_version,is_hsm", only_hsm_default)
     @KeysClientPreparer()
     @recorded_by_proxy

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

@@ -17,6 +17,7 @@
 from azure.core.rest import HttpRequest
 from azure.keyvault.keys import (
     ApiVersion,
+    ExternalKey,
     JsonWebKey,
     KeyProperties,
     KeyReleasePolicy,
@@ -824,6 +825,34 @@ async def test_send_request(self, client, is_hsm, **kwargs):
         response = await client.send_request(request)
         assert response.json()["key"]["kid"] == key.id
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("api_version,is_hsm", only_hsm_default)
+    @AsyncKeysClientPreparer()
+    @recorded_by_proxy_async
+    async def test_create_external_key(self, client, **kwargs):
+        """Register an external HSM key and verify the external_key reference round-trips."""
+        external_id = kwargs.pop("ekm_external_id")
+        if not external_id:
+            pytest.skip(
+                "No external key ID provided. This test requires an EKM-connected HSM and an existing external key."
+            )
+
+        key_name = self.get_resource_name("ext-key")
+        external_key = ExternalKey(id=external_id)
+
+        created = await client.create_external_key(key_name, external_key=external_key)
+        assert created is not None
+        assert created.name == key_name
+        assert created.properties.external_key is not None
+        assert created.properties.external_key.id == external_id
+        assert created.key_type is not None
+
+        # Verify the external_key reference is also returned by a subsequent get_key.
+        fetched = await client.get_key(key_name)
+        assert fetched.properties.external_key is not None
+        assert fetched.properties.external_key.id == external_id
+        assert fetched.key_type is not None
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("api_version,is_hsm", only_hsm_default)
     @AsyncKeysClientPreparer()

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

@@ -17,6 +17,7 @@
 all_api_versions = get_decorator(only_vault=True)
 default_version = get_decorator(api_versions=[DEFAULT_VERSION])
 only_hsm = get_decorator(only_hsm=True)
+only_hsm_default = get_decorator(only_hsm=True, api_versions=[DEFAULT_VERSION])
 
 
 def print(*args):
@@ -148,6 +149,30 @@ def test_example_create_oct_key(self, key_client, **kwargs):
         print(key.properties.key_size)
         # [END create_oct_key]
 
+    @pytest.mark.parametrize("api_version,is_hsm", only_hsm_default)
+    @KeysClientPreparer()
+    @recorded_by_proxy
+    def test_example_create_external_key(self, key_client, **kwargs):
+        external_id = kwargs.pop("ekm_external_id")
+        if not external_id:
+            pytest.skip(
+                "No external key ID provided. This test requires an EKM-connected HSM and an existing external key."
+            )
+        key_name = self.get_resource_name("ext-key")
+
+        # [START create_external_key]
+        from azure.keyvault.keys import ExternalKey
+
+        # the external_key.id refers to the key material managed by an external HSM
+        external_key = ExternalKey(id=external_id)
+        key = key_client.create_external_key(key_name, external_key=external_key)
+
+        print(key.id)
+        print(key.name)
+        print(key.properties.external_key.id)
+        print(key.key_type)
+        # [END create_external_key]
+
     @pytest.mark.parametrize("api_version,is_hsm", all_api_versions)
     @KeysClientPreparer()
     @recorded_by_proxy

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

@@ -17,6 +17,7 @@
 all_api_versions = get_decorator(is_async=True, only_vault=True)
 default_version = get_decorator(is_async=True, api_versions=[DEFAULT_VERSION])
 only_hsm = get_decorator(only_hsm=True, is_async=True)
+only_hsm_default = get_decorator(only_hsm=True, is_async=True, api_versions=[DEFAULT_VERSION])
 
 
 def print(*args):
@@ -150,6 +151,31 @@ async def test_example_create_oct_key(self, key_client, **kwargs):
         print(key.properties.key_size)
         # [END create_oct_key]
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("api_version,is_hsm", only_hsm_default)
+    @AsyncKeysClientPreparer()
+    @recorded_by_proxy_async
+    async def test_example_create_external_key(self, key_client, **kwargs):
+        external_id = kwargs.pop("ekm_external_id")
+        if not external_id:
+            pytest.skip(
+                "No external key ID provided. This test requires an EKM-connected HSM and an existing external key."
+            )
+        key_name = self.get_resource_name("ext-key")
+
+        # [START create_external_key]
+        from azure.keyvault.keys import ExternalKey
+
+        # the external_key.id refers to the key material managed by an external HSM
+        external_key = ExternalKey(id=external_id)
+        key = await key_client.create_external_key(key_name, external_key=external_key)
+
+        print(key.id)
+        print(key.name)
+        print(key.properties.external_key.id)
+        print(key.key_type)
+        # [END create_external_key]
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("api_version,is_hsm", all_api_versions)
     @AsyncKeysClientPreparer()