Application Registrations

Solution:
Added application registrations and regeneration endpoint.
Added appidp endpoint to get token.
Selectively run individual unit or functional tests.
Remove cert keyword from archivist.
Authtoken is now positional (not keyword) arg to Archivist.

Signed-off-by: Paul Hewlett <phewlett76@gmail.com>

Author: Paul Hewlett

README.rst

@@ -35,7 +35,7 @@ One can then use the examples code to create assets (see examples directory):
     # Initialize connection to Archivist - the URL for production will be different to this URL
     arch = Archivist(
         "https://app.rkvst.io",
-        auth=authtoken,
+        authtoken,
     )
     
     # Create a new asset

archivist/access_policies.py

@@ -15,7 +15,7 @@
       # Initialize connection to Archivist
       arch = Archivist(
           "https://app.rkvst.io",
-          auth=authtoken,
+          authtoken,
       )
       access_policy = arch.access_policies.create(...)
 

archivist/appidp.py

@@ -0,0 +1,80 @@
+"""Appidp interface
+
+   Access to the Appidp endpoint.
+
+   The user is not expected to use this class directly. It is an attribute of the
+   :class:`Archivist` class.
+
+   For example instantiate an Archivist instance and execute the methods of the class:
+
+   .. code-block:: python
+
+      with open(".auth_token", mode="r", encoding="utf-8") as tokenfile:
+          authtoken = tokenfile.read().strip()
+
+      # Initialize connection to Archivist
+      arch = Archivist(
+          "https://app.rkvst.io",
+          authtoken,
+      )
+      appidp = arch.appidp.token(...)
+
+"""
+
+import logging
+
+# pylint:disable=cyclic-import      # but pylint doesn't understand this feature
+# pylint:disable=unused-import      # To prevent cyclical import errors forward referencing is used
+# pylint:disable=too-few-public-methods
+from . import archivist as type_helper
+
+from .constants import (
+    APPIDP_SUBPATH,
+    APPIDP_LABEL,
+    APPIDP_TOKEN,
+)
+from .dictmerge import _deepmerge
+
+LOGGER = logging.getLogger(__name__)
+
+
+class AppIDP(dict):
+    """Appidp object"""
+
+
+class _AppIDPClient:
+    """AppIDP Client
+
+    Access to appidp entities. This class is usually
+    accessed as an attribute of the Archivist class.
+
+    Args:
+        archivist (Archivist): :class:`Archivist` instance
+
+    """
+
+    def __init__(self, archivist: "type_helper.Archivist"):
+        self._archivist = archivist
+
+    def token(self, client_id: str, client_secret: str) -> AppIDP:
+        """Create access token from client id and secret
+
+        Args:
+            client_id (str): client id
+            client_secret (str): client_secret
+
+        Returns:
+            :class:`AppIDP` instance
+
+        """
+        return AppIDP(
+            **self._archivist.post(
+                f"{APPIDP_SUBPATH}/{APPIDP_LABEL}/{APPIDP_TOKEN}",
+                {
+                    "grant_type": "client_credentials",
+                    "client_id": client_id,
+                    "client_secret": client_secret,
+                },
+                noheaders=True,
+            )
+        )

archivist/applications.py

@@ -0,0 +1,231 @@
+"""Applications interface
+
+   Access to the applications endpoint.
+
+   The user is not expected to use this class directly. It is an attribute of the
+   :class:`Archivist` class.
+
+   For example instantiate an Archivist instance and execute the methods of the class:
+
+   .. code-block:: python
+
+      with open(".auth_token", mode="r", encoding="utf-8") as tokenfile:
+          authtoken = tokenfile.read().strip()
+
+      # Initialize connection to Archivist
+      arch = Archivist(
+          "https://app.rkvst.io",
+          authtoken,
+      )
+      application = arch.applications.create(...)
+
+"""
+
+import logging
+from typing import Dict, Optional
+
+# pylint:disable=unused-import      # To prevent cyclical import errors forward referencing is used
+# pylint:disable=cyclic-import      # but pylint doesn't understand this feature
+from . import archivist as type_helper
+
+from .constants import (
+    APPLICATIONS_SUBPATH,
+    APPLICATIONS_LABEL,
+    APPLICATIONS_REGENERATE,
+)
+from .dictmerge import _deepmerge
+
+FIXTURE_LABEL = "applications"
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class Application(dict):
+    """Application object"""
+
+
+class _ApplicationsClient:
+    """ApplicationsClient
+
+    Access to applications entities using CRUD interface. This class is usually
+    accessed as an attribute of the Archivist class.
+
+    Args:
+        archivist (Archivist): :class:`Archivist` instance
+
+    """
+
+    def __init__(self, archivist: "type_helper.Archivist"):
+        self._archivist = archivist
+
+    def create(self, display_name: str, custom_claims: Dict) -> Application:
+        """Create application
+
+        Creates application with defined attributes.
+
+        Args:
+            display_name (str): display name of application.
+            custom_claims (dict): custom claims
+
+        Returns:
+            :class:`Application` instance
+
+        """
+        LOGGER.debug("Create Application %s", display_name)
+        return self.create_from_data(
+            self.__query(
+                display_name=display_name,
+                custom_claims=custom_claims,
+            ),
+        )
+
+    def create_from_data(self, data: Dict) -> Application:
+        """Create application
+
+        Creates application with request body from data stream.
+        Suitable for reading data from a file using json.load or yaml.load
+
+        Args:
+            data (dict): request body of application.
+
+        Returns:
+            :class:`Application` instance
+
+        """
+        return Application(
+            **self._archivist.post(
+                f"{APPLICATIONS_SUBPATH}/{APPLICATIONS_LABEL}",
+                data,
+            )
+        )
+
+    def read(self, identity: str) -> Application:
+        """Read Application
+
+        Reads application.
+
+        Args:
+            identity (str): applications identity e.g. applications/xxxxxxxxxxxxxxxxxxxxxxx
+
+        Returns:
+            :class:`Application` instance
+
+        """
+        return Application(
+            **self._archivist.get(
+                APPLICATIONS_SUBPATH,
+                identity,
+            )
+        )
+
+    def update(
+        self,
+        identity: str,
+        *,
+        display_name: str = None,
+        custom_claims: Optional[Dict] = None,
+    ) -> Application:
+        """Update Application
+
+        Update application.
+
+        Args:
+            identity (str): applications identity e.g. applications/xxxxxxxxxxxxxxxxxxxxxxx
+            display_name (str): display name of application.
+            custom_claims (dict): custom claims
+
+        Returns:
+            :class:`Application` instance
+
+        """
+        return Application(
+            **self._archivist.patch(
+                APPLICATIONS_SUBPATH,
+                identity,
+                self.__query(
+                    display_name=display_name,
+                    custom_claims=custom_claims,
+                ),
+            )
+        )
+
+    def delete(self, identity: str) -> Dict:
+        """Delete Application
+
+        Deletes application.
+
+        Args:
+            identity (str): applications identity e.g. applications/xxxxxxxxxxxxxxxxxxxxxxx
+
+        Returns:
+            :class:`Application` instance - empty?
+
+        """
+        return self._archivist.delete(APPLICATIONS_SUBPATH, identity)
+
+    def __query(
+        self,
+        *,
+        display_name: Optional[str] = None,
+        custom_claims: Optional[Dict] = None,
+    ) -> Dict:
+
+        query = {}
+
+        if display_name is not None:
+            query["display_name"] = display_name
+
+        if custom_claims is not None:
+            query["custom_claims"] = custom_claims
+
+        return _deepmerge(self._archivist.fixtures.get(FIXTURE_LABEL), query)
+
+    def list(
+        self,
+        *,
+        page_size: Optional[int] = None,
+        display_name: Optional[str] = None,
+    ):
+        """List applications.
+
+        List applications that match criteria.
+
+        Args:
+            display_name (str): display name (optional)
+            page_size (int): optional page size. (Rarely used).
+
+        Returns:
+            iterable that returns :class:`Application` instances
+
+        """
+        return (
+            Application(**a)
+            for a in self._archivist.list(
+                f"{APPLICATIONS_SUBPATH}/{APPLICATIONS_LABEL}",
+                APPLICATIONS_LABEL,
+                page_size=page_size,
+                query=self.__query(display_name=display_name),
+            )
+        )
+
+    def regenerate(self, identity: str) -> Application:
+        """Regenerate secret
+
+        Makes an SBOM public.
+
+        Args:
+            identity (str): identity of application
+
+        Returns:
+            :class:`Application` instance
+
+        """
+        LOGGER.debug("Regenerate %s", identity)
+        return Application(
+            **self._archivist.post(
+                f"{APPLICATIONS_SUBPATH}/{identity}",
+                None,
+                verb=APPLICATIONS_REGENERATE,
+            )
+        )