[keyvault] add parse id feature for certs

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

@@ -2,6 +2,10 @@
 
 ## 4.2.1 (Unreleased)
 
+### Added
+- Added method `parse_certificate_id` that parses out a full ID returned by key vault, so users
+can easily access the certificate's `name`, `vault_url`, `version` etc.
+
 
 ## 4.2.0 (2020-08-11)
 ### Fixed

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

@@ -24,7 +24,9 @@
     LifetimeAction,
     KeyVaultCertificate
 )
+from ._parse_id import parse_certificate_id
 from ._shared.client_base import ApiVersion
+from ._shared import ParsedId
 
 __all__ = [
     "ApiVersion",
@@ -47,7 +49,9 @@
     "CertificateContentType",
     "WellKnownIssuerNames",
     "CertificateIssuer",
-    "IssuerProperties"
+    "IssuerProperties",
+    "parse_certificate_id",
+    "ParsedId",
 ]
 
 from ._version import VERSION

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

@@ -4,7 +4,7 @@
 # ------------------------------------
 
 # pylint: disable=too-many-lines,too-many-public-methods
-from ._shared import parse_vault_id
+from ._shared import parse_key_vault_identifier
 from ._generated.v7_1 import models
 from ._enums import(
     CertificatePolicyAction,
@@ -147,7 +147,7 @@ def __init__(self, **kwargs):
         # type: (**Any) -> None
         self._attributes = kwargs.pop("attributes", None)
         self._id = kwargs.pop("cert_id", None)
-        self._vault_id = parse_vault_id(self._id)
+        self._vault_id = parse_key_vault_identifier(self._id)
         self._x509_thumbprint = kwargs.pop("x509_thumbprint", None)
         self._tags = kwargs.pop("tags", None)
 
@@ -430,7 +430,7 @@ def __init__(
     ):
         # type: (...) -> None
         self._id = cert_operation_id
-        self._vault_id = parse_vault_id(cert_operation_id)
+        self._vault_id = parse_key_vault_identifier(cert_operation_id)
         self._issuer_name = issuer_name
         self._certificate_type = certificate_type
         self._certificate_transparency = certificate_transparency
@@ -1058,7 +1058,7 @@ class IssuerProperties(object):
     def __init__(self, provider=None, **kwargs):
         # type: (Optional[str], **Any) -> None
         self._id = kwargs.pop("issuer_id", None)
-        self._vault_id = parse_vault_id(self._id)
+        self._vault_id = parse_key_vault_identifier(self._id)
         self._provider = provider
 
     def __repr__(self):
@@ -1120,7 +1120,7 @@ def __init__(
         self._organization_id = organization_id
         self._admin_contacts = admin_contacts
         self._id = kwargs.pop("issuer_id", None)
-        self._vault_id = parse_vault_id(self._id)
+        self._vault_id = parse_key_vault_identifier(self._id)
 
     def __repr__(self):
         # type () -> str
@@ -1158,7 +1158,8 @@ def id(self):
     def name(self):
         # type: () -> str
         # Issuer name is listed under version under vault_id
-        # This is because the id we pass to parse_vault_id has an extra segment, so where most cases the version of
+        # This is because the id we pass to parse_key_vault_identifier has an extra segment,
+        # so where most cases the version of
         # The general pattern is certificates/name/version, but here we have certificates/issuers/name/version
         # Issuers are not versioned.
         """:rtype: str"""

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

@@ -0,0 +1,41 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+
+from ._shared import parse_key_vault_identifier, ParsedId
+
+
+def parse_certificate_id(original_id):
+    # type: (str) -> ParsedId
+    """Parses a full certificate's ID into a class.
+
+    :param str original_id: the full original identifier of a certificate
+    :returns: Returns a parsed certificate id
+    :rtype: ~azure.keyvault.certificates.ParsedId
+    :raises: ValueError
+
+    Example:
+        .. literalinclude:: ../tests/test_parse_id.py
+            :start-after: [START parse_certificate_id]
+            :end-before: [END parse_certificate_id]
+            :language: python
+            :caption: Parse a certificate's ID
+            :dedent: 8
+    """
+    parsed_id = parse_key_vault_identifier(original_id)
+
+    valid_collections = ["certificates", "deletedcertificates"]
+
+    if parsed_id.collection not in valid_collections:
+        raise ValueError(
+            "Collection '{}' is not a valid certificate collection. ".format(parsed_id.collection),
+            "Valid collections are: {}".format(", ".join(valid_collections))
+        )
+    return ParsedId(
+        collection=parsed_id.collection,
+        name=parsed_id.name,
+        original_id=parsed_id.original_id,
+        vault_url=parsed_id.vault_url,
+        version=parsed_id.version
+    )

sdk/keyvault/azure-keyvault-certificates/azure/keyvault/certificates/_shared/__init__.py

@@ -2,8 +2,6 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from collections import namedtuple
-
 try:
     import urllib.parse as parse
 except ImportError:
@@ -24,23 +22,47 @@
     "KeyVaultClientBase",
 ]
 
-_VaultId = namedtuple("VaultId", ["vault_url", "collection", "name", "version"])
+class ParsedId():
+    """Represents a key vault identifier and its parsed contents.
+
+    :param str original_id: The originally received complete identifier
+    :param str vault_url: The vault URL
+    :param str collection: The collection name of the id.
+    :param str name: The name extracted from the id
+    :param str version: The version extracted from the id
+    """
+
+    def __init__(
+        self,
+        original_id,  # type: str
+        vault_url,  # type: str
+        collection,  # type: str
+        name,  # type: str
+        version=None  # type: Optional[str]
+    ):
+        self.original_id = original_id
+        self.vault_url = vault_url
+        self.collection = collection
+        self.name = name
+        self.version = version
 
 
-def parse_vault_id(url):
+def parse_key_vault_identifier(original_id):
+    # type: (str) -> ParsedId
     try:
-        parsed_uri = parse.urlparse(url)
+        parsed_uri = parse.urlparse(original_id)
     except Exception:  # pylint: disable=broad-except
-        raise ValueError("'{}' is not not a valid url".format(url))
+        raise ValueError("'{}' is not not a valid ID".format(original_id))
     if not (parsed_uri.scheme and parsed_uri.hostname):
-        raise ValueError("'{}' is not not a valid url".format(url))
+        raise ValueError("'{}' is not not a valid ID".format(original_id))
 
     path = list(filter(None, parsed_uri.path.split("/")))
 
     if len(path) < 2 or len(path) > 3:
-        raise ValueError("'{}' is not not a valid vault url".format(url))
+        raise ValueError("'{}' is not not a valid vault ID".format(original_id))
 
-    return _VaultId(
+    return ParsedId(
+        original_id=original_id,
         vault_url="{}://{}".format(parsed_uri.scheme, parsed_uri.hostname),
         collection=path[0],
         name=path[1],

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

@@ -24,7 +24,7 @@
     CertificateIssuer,
     IssuerProperties,
 )
-from azure.keyvault.certificates._shared import parse_vault_id
+from azure.keyvault.certificates._shared import parse_key_vault_identifier
 from devtools_testutils import ResourceGroupPreparer, KeyVaultPreparer
 
 from _shared.preparer import KeyVaultClientPreparer as _KeyVaultClientPreparer
@@ -83,7 +83,7 @@ def _validate_certificate_operation(self, pending_cert_operation, vault, cert_na
         self.assertIsNotNone(pending_cert_operation)
         self.assertIsNotNone(pending_cert_operation.csr)
         self.assertEqual(original_cert_policy.issuer_name, pending_cert_operation.issuer_name)
-        pending_id = parse_vault_id(pending_cert_operation.id)
+        pending_id = parse_key_vault_identifier(pending_cert_operation.id)
         self.assertEqual(pending_id.vault_url.strip("/"), vault.strip("/"))
         self.assertEqual(pending_id.name, cert_name)
 
@@ -258,7 +258,7 @@ def test_list(self, client, **kwargs):
             error_count = 0
             try:
                 cert_bundle = self._import_common_certificate(client=client, cert_name=cert_name)
-                parsed_id = parse_vault_id(url=cert_bundle.id)
+                parsed_id = parse_key_vault_identifier(original_id=cert_bundle.id)
                 cid = parsed_id.vault_url + "/" + parsed_id.collection + "/" + parsed_id.name
                 expected[cid.strip("/")] = cert_bundle
             except Exception as ex:
@@ -287,7 +287,7 @@ def test_list_certificate_versions(self, client, **kwargs):
             error_count = 0
             try:
                 cert_bundle = self._import_common_certificate(client=client, cert_name=cert_name)
-                parsed_id = parse_vault_id(url=cert_bundle.id)
+                parsed_id = parse_key_vault_identifier(original_id=cert_bundle.id)
                 cid = parsed_id.vault_url + "/" + parsed_id.collection + "/" + parsed_id.name + "/" + parsed_id.version
                 expected[cid.strip("/")] = cert_bundle
             except Exception as ex:
@@ -355,7 +355,7 @@ def test_recover_and_purge(self, client, **kwargs):
             client.begin_delete_certificate(certificate_name=cert_name).wait()
 
         # validate all our deleted certificates are returned by list_deleted_certificates
-        deleted = [parse_vault_id(url=c.id).name for c in client.list_deleted_certificates()]
+        deleted = [parse_key_vault_identifier(original_id=c.id).name for c in client.list_deleted_certificates()]
         self.assertTrue(all(c in deleted for c in certs.keys()))
 
         # recover select certificates
@@ -370,7 +370,7 @@ def test_recover_and_purge(self, client, **kwargs):
             time.sleep(50)
 
         # validate none of our deleted certificates are returned by list_deleted_certificates
-        deleted = [parse_vault_id(url=c.id).name for c in client.list_deleted_certificates()]
+        deleted = [parse_key_vault_identifier(original_id=c.id).name for c in client.list_deleted_certificates()]
         self.assertTrue(not any(c in deleted for c in certs.keys()))
 
         # validate the recovered certificates

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

@@ -22,7 +22,7 @@
     IssuerProperties,
 )
 from azure.keyvault.certificates.aio import CertificateClient
-from azure.keyvault.certificates._shared import parse_vault_id
+from azure.keyvault.certificates._shared import parse_key_vault_identifier
 from devtools_testutils import ResourceGroupPreparer, KeyVaultPreparer
 import pytest
 
@@ -77,7 +77,7 @@ def _validate_certificate_operation(self, pending_cert_operation, vault, cert_na
         self.assertIsNotNone(pending_cert_operation)
         self.assertIsNotNone(pending_cert_operation.csr)
         self.assertEqual(original_cert_policy.issuer_name, pending_cert_operation.issuer_name)
-        pending_id = parse_vault_id(pending_cert_operation.id)
+        pending_id = parse_key_vault_identifier(pending_cert_operation.id)
         self.assertEqual(pending_id.vault_url.strip("/"), vault.strip("/"))
         self.assertEqual(pending_id.name, cert_name)
 
@@ -253,7 +253,7 @@ async def test_list(self, client, **kwargs):
             error_count = 0
             try:
                 cert_bundle = await self._import_common_certificate(client=client, cert_name=cert_name)
-                parsed_id = parse_vault_id(url=cert_bundle.id)
+                parsed_id = parse_key_vault_identifier(original_id=cert_bundle.id)
                 cid = parsed_id.vault_url + "/" + parsed_id.collection + "/" + parsed_id.name
                 expected[cid.strip("/")] = cert_bundle
             except Exception as ex:
@@ -282,7 +282,7 @@ async def test_list_certificate_versions(self, client, **kwargs):
             error_count = 0
             try:
                 cert_bundle = await self._import_common_certificate(client=client, cert_name=cert_name)
-                parsed_id = parse_vault_id(url=cert_bundle.id)
+                parsed_id = parse_key_vault_identifier(original_id=cert_bundle.id)
                 cid = parsed_id.vault_url + "/" + parsed_id.collection + "/" + parsed_id.name + "/" + parsed_id.version
                 expected[cid.strip("/")] = cert_bundle
             except Exception as ex:
@@ -355,7 +355,7 @@ async def test_recover_and_purge(self, client, **kwargs):
         deleted_certificates = client.list_deleted_certificates()
         deleted = []
         async for c in deleted_certificates:
-            deleted.append(parse_vault_id(url=c.id).name)
+            deleted.append(parse_key_vault_identifier(original_id=c.id).name)
         self.assertTrue(all(c in deleted for c in certs.keys()))
 
         # recover select certificates
@@ -373,7 +373,7 @@ async def test_recover_and_purge(self, client, **kwargs):
         deleted_certificates = client.list_deleted_certificates()
         deleted = []
         async for c in deleted_certificates:
-            deleted.append(parse_vault_id(url=c.id).name)
+            deleted.append(parse_key_vault_identifier(original_id=c.id).name)
         self.assertTrue(not any(c in deleted for c in certs.keys()))
 
         # validate the recovered certificates

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

@@ -0,0 +1,53 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# -------------------------------------
+import pytest
+from azure.keyvault.certificates import parse_certificate_id
+from _shared.test_case import KeyVaultTestCase
+
+
+class TestParseId(KeyVaultTestCase):
+    def test_parse_certificate_id_with_version(self):
+        # [START parse_certificate_id]
+        original_id = "https://keyvault-name.vault.azure.net/certificates/certificate-name/version"
+        parsed_certificate_id = parse_certificate_id(original_id)
+
+        print(parsed_certificate_id.name)
+        print(parsed_certificate_id.collection)
+        print(parsed_certificate_id.vault_url)
+        print(parsed_certificate_id.version)
+        print(parsed_certificate_id.original_id)
+        # [END parse_certificate_id]
+        assert parsed_certificate_id.name == "certificate-name"
+        assert parsed_certificate_id.collection == "certificates"
+        assert parsed_certificate_id.vault_url == "https://keyvault-name.vault.azure.net"
+        assert parsed_certificate_id.version == "version"
+        assert parsed_certificate_id.original_id == "https://keyvault-name.vault.azure.net/certificates/certificate-name/version"
+
+    def test_parse_certificate_id_with_pending_version(self):
+        original_id = "https://keyvault-name.vault.azure.net/certificates/certificate-name/pending"
+        parsed_certificate_id = parse_certificate_id(original_id)
+
+        assert parsed_certificate_id.name == "certificate-name"
+        assert parsed_certificate_id.collection == "certificates"
+        assert parsed_certificate_id.vault_url == "https://keyvault-name.vault.azure.net"
+        assert parsed_certificate_id.version == "pending"
+        assert parsed_certificate_id.original_id == "https://keyvault-name.vault.azure.net/certificates/certificate-name/pending"
+
+    def test_parse_deleted_certificate_id(self):
+        original_id = "https://keyvault-name.vault.azure.net/deletedcertificates/deleted-certificate"
+        parsed_certificate_id = parse_certificate_id(original_id)
+
+        assert parsed_certificate_id.name == "deleted-certificate"
+        assert parsed_certificate_id.collection == "deletedcertificates"
+        assert parsed_certificate_id.vault_url == "https://keyvault-name.vault.azure.net"
+        assert parsed_certificate_id.version == None
+        assert parsed_certificate_id.original_id == "https://keyvault-name.vault.azure.net/deletedcertificates/deleted-certificate"
+
+    def test_parse_certificate_id_with_invalid_collection(self):
+        original_id = "https://keyvault-name.vault.azure.net/invalidcollection/certificate"
+
+        with pytest.raises(ValueError) as excinfo:
+            parse_certificate_id(original_id)
+        assert "Collection 'invalidcollection' is not a valid certificate collection" in str(excinfo.value)