Prevent AttributeError during get_certificate_version

sdk/keyvault/azure-keyvault-certificates/CHANGELOG.md

@@ -1,6 +1,7 @@
 # Release History
 
 ## 4.2.0b2 (Unreleased)
+- Fixed an `AttributeError` during `get_certificate_version`
 - Values of `x-ms-keyvault-region` and `x-ms-keyvault-service-version` headers
   are no longer redacted in logging output.
 - Updated minimum `azure-core` version to 1.4.0

sdk/keyvault/azure-keyvault-certificates/azure/keyvault/certificates/_models.py

@@ -320,11 +320,17 @@ def _from_certificate_bundle(cls, certificate_bundle):
         # type: (models.CertificateBundle) -> KeyVaultCertificate
         """Construct a certificate from an autorest-generated certificateBundle"""
         # pylint:disable=protected-access
+
+        if certificate_bundle.policy:
+            policy = CertificatePolicy._from_certificate_policy_bundle(certificate_bundle.policy)
+        else:
+            policy = None
+
         return cls(
             properties=CertificateProperties._from_certificate_item(certificate_bundle),
             key_id=certificate_bundle.kid,
             secret_id=certificate_bundle.sid,
-            policy=CertificatePolicy._from_certificate_policy_bundle(certificate_bundle.policy),
+            policy=policy,
             cer=certificate_bundle.cer,
         )
 

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

@@ -5,7 +5,6 @@
 import functools
 import json
 import logging
-import os
 import time
 
 from azure_devtools.scenario_tests import RecordingProcessor, RequestUrlNormalizer
@@ -21,7 +20,6 @@
     KeyUsageType,
     CertificateContentType,
     LifetimeAction,
-    WellKnownIssuerNames,
     CertificateIssuer,
     IssuerProperties,
 )
@@ -623,3 +621,29 @@ def test_logging_disabled(self, client, **kwargs):
                 except (ValueError, KeyError):
                     # this means the message is not JSON or has no kty property
                     pass
+
+    @ResourceGroupPreparer(random_name_enabled=True)
+    @KeyVaultPreparer()
+    @KeyVaultClientPreparer()
+    def test_get_certificate_version(self, client, **kwargs):
+        cert_name = self.get_resource_name("cert")
+        for _ in range(self.list_test_size):
+            client.begin_create_certificate(cert_name, CertificatePolicy.get_default()).wait()
+
+        for version_properties in client.list_properties_of_certificate_versions(cert_name):
+            cert = client.get_certificate_version(version_properties.name, version_properties.version)
+
+            # This isn't factored out into a helper method because the properties are not exactly equal.
+            # get_certificate_version sets "recovery_days" and "recovery_level" but the list method does not.
+            # (This is Key Vault's behavior, not an SDK limitation.)
+            assert version_properties.created_on == cert.properties.created_on
+            assert version_properties.enabled == cert.properties.enabled
+            assert version_properties.expires_on == cert.properties.expires_on
+            assert version_properties.id == cert.properties.id
+            assert version_properties.name == cert.properties.name
+            assert version_properties.not_before == cert.properties.not_before
+            assert version_properties.tags == cert.properties.tags
+            assert version_properties.updated_on == cert.properties.updated_on
+            assert version_properties.vault_url == cert.properties.vault_url
+            assert version_properties.version == cert.properties.version
+            assert version_properties.x509_thumbprint == cert.properties.x509_thumbprint

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

@@ -4,7 +4,6 @@
 # ------------------------------------
 import asyncio
 import functools
-import os
 import logging
 import json
 
@@ -19,7 +18,6 @@
     KeyUsageType,
     CertificateContentType,
     LifetimeAction,
-    WellKnownIssuerNames,
     CertificateIssuer,
     IssuerProperties,
 )
@@ -71,7 +69,9 @@ async def _import_common_certificate(self, client, cert_name):
             validity_in_months=12,
             key_usage=["digitalSignature", "keyEncipherment"],
         )
-        return await client.import_certificate(cert_name, CertificateClientTests.CERT_CONTENT_PASSWORD_ENODED, policy=cert_policy, password=cert_password)
+        return await client.import_certificate(
+            cert_name, CertificateClientTests.CERT_CONTENT_PASSWORD_ENODED, policy=cert_policy, password=cert_password
+        )
 
     def _validate_certificate_operation(self, pending_cert_operation, vault, cert_name, original_cert_policy):
         self.assertIsNotNone(pending_cert_operation)
@@ -235,7 +235,7 @@ async def test_import_certificate_password_encoded_no_policy(self, client):
         certificate = await client.import_certificate(
             certificate_name="importPasswordEncodedCertificate",
             certificate_bytes=CertificateClientTests.CERT_CONTENT_PASSWORD_ENODED,
-            password="123"
+            password="123",
         )
         self.assertIsNotNone(certificate.policy)
 
@@ -635,3 +635,29 @@ async def test_logging_disabled(self, client, **kwargs):
                 except (ValueError, KeyError):
                     # this means the message is not JSON or has no kty property
                     pass
+
+    @ResourceGroupPreparer(random_name_enabled=True)
+    @KeyVaultPreparer()
+    @KeyVaultClientPreparer()
+    async def test_get_certificate_version(self, client, **kwargs):
+        cert_name = self.get_resource_name("cert")
+        policy = CertificatePolicy.get_default()
+        await asyncio.gather(*[client.create_certificate(cert_name, policy) for _ in range(self.list_test_size)])
+
+        async for version_properties in client.list_properties_of_certificate_versions(cert_name):
+            cert = await client.get_certificate_version(version_properties.name, version_properties.version)
+
+            # This isn't factored out into a helper method because the properties are not exactly equal.
+            # get_certificate_version sets "recovery_days" and "recovery_level" but the list method does not.
+            # (This is Key Vault's behavior, not an SDK limitation.)
+            assert version_properties.created_on == cert.properties.created_on
+            assert version_properties.enabled == cert.properties.enabled
+            assert version_properties.expires_on == cert.properties.expires_on
+            assert version_properties.id == cert.properties.id
+            assert version_properties.name == cert.properties.name
+            assert version_properties.not_before == cert.properties.not_before
+            assert version_properties.tags == cert.properties.tags
+            assert version_properties.updated_on == cert.properties.updated_on
+            assert version_properties.vault_url == cert.properties.vault_url
+            assert version_properties.version == cert.properties.version
+            assert version_properties.x509_thumbprint == cert.properties.x509_thumbprint

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

@@ -4,8 +4,6 @@
 # ------------------------------------
 from __future__ import print_function
 import functools
-import hashlib
-import os
 
 from azure.keyvault.certificates import (
     CertificateClient,
@@ -139,12 +137,8 @@ def test_example_certificate_list_operations(self, client, **kwargs):
             validity_in_months=24,
         )
 
-        polling_interval = 0 if self.is_playback() else None
-
-        for i in range(4):
-            certificate_client.begin_create_certificate(
-                certificate_name="certificate{}".format(i), policy=cert_policy, _polling_interval=polling_interval
-            ).wait()
+        certificate_name = self.get_replayable_random_resource_name("cert")
+        certificate_client.begin_create_certificate(certificate_name, cert_policy).wait()
 
         # [START list_properties_of_certificates]
 
@@ -159,17 +153,24 @@ def test_example_certificate_list_operations(self, client, **kwargs):
             print(certificate.enabled)
 
         # [END list_properties_of_certificates]
+
+        # create a second version of the cert
+        certificate_client.begin_create_certificate(certificate_name, cert_policy).wait()
+
         # [START list_properties_of_certificate_versions]
 
         # get an iterator of a certificate's versions
-        certificate_versions = certificate_client.list_properties_of_certificate_versions("certificate-name")
+        certificate_versions = certificate_client.list_properties_of_certificate_versions(certificate_name)
 
         for certificate in certificate_versions:
             print(certificate.id)
             print(certificate.updated_on)
             print(certificate.version)
 
         # [END list_properties_of_certificate_versions]
+
+        certificate_client.begin_delete_certificate(certificate_name).wait()
+
         # [START list_deleted_certificates]
 
         # get an iterator of deleted certificates (requires soft-delete enabled for the vault)

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

@@ -123,14 +123,8 @@ async def test_example_certificate_list_operations(self, client, **kwargs):
             validity_in_months=24,
         )
 
-        create_certificate_pollers = []
-        for i in range(4):
-            create_certificate_pollers.append(
-                certificate_client.create_certificate(certificate_name="certificate{}".format(i), policy=cert_policy)
-            )
-
-        for poller in create_certificate_pollers:
-            await poller
+        certificate_name = self.get_replayable_random_resource_name("cert")
+        await certificate_client.create_certificate(certificate_name, cert_policy)
 
         # [START list_properties_of_certificates]
 
@@ -145,6 +139,10 @@ async def test_example_certificate_list_operations(self, client, **kwargs):
             print(certificate.enabled)
 
         # [END list_properties_of_certificates]
+
+        # create a second version of the cert
+        await certificate_client.create_certificate(certificate_name, cert_policy)
+
         # [START list_properties_of_certificate_versions]
 
         # get an iterator of all versions of a certificate
@@ -156,6 +154,9 @@ async def test_example_certificate_list_operations(self, client, **kwargs):
             print(certificate.properties.version)
 
         # [END list_properties_of_certificate_versions]
+
+        await certificate_client.delete_certificate(certificate_name)
+
         # [START list_deleted_certificates]
 
         # get an iterator of deleted certificates (requires soft-delete enabled for the vault)