RSTUF Interface refactoring (IServices)

repository_service_tuf_worker/interfaces.py

@@ -1,12 +1,13 @@
 # SPDX-FileCopyrightText: 2022-2023 VMware Inc
 #
 # SPDX-License-Identifier: MIT
-
-
+import importlib
+import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, List, Optional
+from typing import Any, Dict, List, Optional
 
+from dynaconf import Dynaconf
 from securesystemslib.signer import Key, Signer
 from tuf.api.metadata import Metadata, T
 
@@ -15,7 +16,7 @@
 class ServiceSettings:
     """Dataclass for service settings."""
 
-    name: str
+    names: List[str]
     argument: str
     required: bool
     default: Optional[Any] = None
@@ -30,6 +31,13 @@ def configure(cls, settings) -> None:
         """
         pass  # pragma: no cover
 
+    @classmethod
+    def from_dynaconf(cls, settings: Dynaconf) -> None:
+        """
+        Run actions to test, configure using the settings.
+        """
+        _setup_service_dynaconf(cls, settings.KEYVAULT_BACKEND, settings)
+
     @classmethod
     @abstractmethod
     def settings(cls) -> List[ServiceSettings]:
@@ -47,18 +55,26 @@ def get(self, public_key: Key) -> Signer:
 class IStorage(ABC):
     @classmethod
     @abstractmethod
-    def configure(cls, settings: Any) -> None:
+    def configure(cls, settings: Dynaconf) -> None:
         """
         Run actions to test, configure using the settings.
         """
         raise NotImplementedError  # pragma: no cover
 
+    @classmethod
+    def from_dynaconf(cls, settings: Dynaconf) -> None:
+        """
+        Run actions to test and configure using the dynaconf settings.
+        """
+        _setup_service_dynaconf(cls, settings.STORAGE_BACKEND, settings)
+
     @classmethod
     @abstractmethod
     def settings(cls) -> List[ServiceSettings]:
         """
         Define all the ServiceSettings required in settings.
         """
+
         raise NotImplementedError  # pragma: no cover
 
     @abstractmethod
@@ -79,3 +95,78 @@ def put(
         Stores file bytes within a file with a specific filename.
         """
         raise NotImplementedError  # pragma: no cover
+
+
+def _setup_service_dynaconf(cls: Any, backend: Any, settings: Dynaconf):
+    """
+    Setup a Interface Service (IService) from settings Dynaconf (environment
+    variables)
+    """
+    # the 'service import is used to retrieve sublcasses (Implemented Services)
+    from repository_service_tuf_worker import services  # noqa
+
+    service_backends = [i.__name__.upper() for i in cls.__subclasses__()]
+    backend_name = f"RSTUF_{cls.__name__.replace('I', '').upper()}_BACKEND"
+
+    if type(backend) is not str and issubclass(
+        backend, tuple(cls.__subclasses__())
+    ):
+        logging.debug(f"{backend_name} is defined as {backend}")
+
+    elif backend.upper() not in service_backends:
+        raise ValueError(
+            f"Invalid {backend_name} {backend}. "
+            f"Supported {backend_name} {', '.join(service_backends)}"
+        )
+    else:
+        backend = getattr(
+            importlib.import_module("repository_service_tuf_worker.services"),
+            backend,
+        )
+        # look all required settings
+        if missing_settings := [
+            s.names
+            for s in backend.settings()
+            if s.required and all(n not in settings for n in s.names)
+        ]:
+            # add the prefix `RSTUF_` to attributes including as dynaconf
+            # removes it. It makes the message more clear to the user.
+            missing_stg: List = []
+            for missing in missing_settings:
+                missing_stg.append("RSTUF_" + " or RSTUF_".join(missing))
+
+            raise AttributeError(
+                "'Settings' object has no attribute(s) (environment variables)"
+                f": {', '.join(missing_stg)}"
+            )
+
+        # parse and define the keyargs from dynaconf
+        kwargs: Dict[str, Any] = {}
+        for s_var in backend.settings():
+            if all(
+                [
+                    settings.store.get(var_name) is None
+                    for var_name in s_var.names
+                ]
+            ):
+                for var_name in s_var.names:
+                    settings.store[var_name] = s_var.default
+                kwargs[s_var.argument] = settings.store[s_var.names[0]]
+            else:
+                for var_name in s_var.names:
+                    if settings.store.get(var_name) is not None:
+                        kwargs[s_var.argument] = settings.store[var_name]
+                        break
+
+        if cls.__name__ == "IStorage":
+            settings.STORAGE_BACKEND = backend
+            settings.STORAGE_BACKEND.configure(settings)
+            settings.STORAGE = settings.STORAGE_BACKEND(**kwargs)
+
+        elif cls.__name__ == "IKeyVault":
+            settings.KEYVAULT_BACKEND = backend
+            settings.KEYVAULT_BACKEND.configure(settings)
+            settings.KEYVAULT = settings.KEYVAULT_BACKEND(**kwargs)
+
+        else:
+            raise ValueError(f"Invalid Interface {cls.__name__}")

repository_service_tuf_worker/repository.py

@@ -3,7 +3,6 @@
 # SPDX-License-Identifier: MIT
 
 import enum
-import importlib
 import logging
 import time
 import warnings
@@ -41,12 +40,10 @@
 )
 from tuf.api.serialization.json import CanonicalJSONSerializer, JSONSerializer
 
-# the 'service import is used to retrieve sublcasses (Implemented Services)
 from repository_service_tuf_worker import (  # noqa
     Dynaconf,
     get_repository_settings,
     get_worker_settings,
-    services,
 )
 from repository_service_tuf_worker.interfaces import IKeyVault, IStorage
 from repository_service_tuf_worker.models import (
@@ -159,95 +156,12 @@ def refresh_settings(self, worker_settings: Optional[Dynaconf] = None):
         #
         # Backends
         #
-        storage_backends = [
-            storage.__name__.upper() for storage in IStorage.__subclasses__()
-        ]
-
-        if type(settings.STORAGE_BACKEND) is not str and issubclass(
-            settings.STORAGE_BACKEND, tuple(IStorage.__subclasses__())
-        ):
-            logging.debug(
-                f"STORAGE_BACKEND is defined as {settings.STORAGE_BACKEND}"
-            )
-
-        elif settings.STORAGE_BACKEND.upper() not in storage_backends:
-            raise ValueError(
-                f"Invalid Storage Backend {settings.STORAGE_BACKEND}. "
-                f"Supported Storage Backends {', '.join(storage_backends)}"
-            )
-        else:
-            settings.STORAGE_BACKEND = getattr(
-                importlib.import_module(
-                    "repository_service_tuf_worker.services"
-                ),
-                settings.STORAGE_BACKEND,
-            )
-
-            if missing := [
-                s.name
-                for s in settings.STORAGE_BACKEND.settings()
-                if s.required and s.name not in settings
-            ]:
-                raise AttributeError(
-                    "'Settings' object has not attribute(s) "
-                    f"{', '.join(missing)}"
-                )
-
-            storage_kwargs: Dict[str, Any] = {}
-            for s in settings.STORAGE_BACKEND.settings():
-                if settings.store.get(s.name) is None:
-                    settings.store[s.name] = s.default
-
-                storage_kwargs[s.argument] = settings.store[s.name]
-
-            settings.STORAGE_BACKEND.configure(settings)
-            settings.STORAGE = settings.STORAGE_BACKEND(**storage_kwargs)
-
-        keyvault_backends = [
-            keyvault.__name__.upper()
-            for keyvault in IKeyVault.__subclasses__()
-        ]
-
-        if type(settings.KEYVAULT_BACKEND) is not str and issubclass(
-            settings.KEYVAULT_BACKEND, tuple(IKeyVault.__subclasses__())
-        ):
-            logging.debug(
-                f"KEYVAULT_BACKEND is defined as {settings.KEYVAULT_BACKEND}"
-            )
-
-        elif settings.KEYVAULT_BACKEND.upper() not in keyvault_backends:
-            raise ValueError(
-                f"Invalid Key Vault Backend {settings.KEYVAULT_BACKEND}. "
-                "Supported Key Vault Backends :"
-                f"{', '.join(keyvault_backends)}"
-            )
-        else:
-            settings.KEYVAULT_BACKEND = getattr(
-                importlib.import_module(
-                    "repository_service_tuf_worker.services"
-                ),
-                settings.KEYVAULT_BACKEND,
-            )
-
-            if missing := [
-                s.name
-                for s in settings.KEYVAULT_BACKEND.settings()
-                if s.required and s.name not in settings
-            ]:
-                raise AttributeError(
-                    "'Settings' object has not attribute(s) "
-                    f"{', '.join(missing)}"
-                )
-
-            keyvault_kwargs: Dict[str, Any] = {}
-            for s in settings.KEYVAULT_BACKEND.settings():
-                if settings.store.get(s.name) is None:
-                    settings.store[s.name] = s.default
 
-                keyvault_kwargs[s.argument] = settings.store[s.name]
+        # storage
+        IStorage.from_dynaconf(settings)
 
-            settings.KEYVAULT_BACKEND.configure(settings)
-            settings.KEYVAULT = settings.KEYVAULT_BACKEND(**keyvault_kwargs)
+        # keyvault
+        IKeyVault.from_dynaconf(settings)
 
         self._worker_settings = settings
         return settings

repository_service_tuf_worker/services/keyvault/local.py

@@ -18,7 +18,11 @@
 from securesystemslib.interface import import_privatekey_from_file
 from securesystemslib.signer import Key, SSlibKey, SSlibSigner
 
-from repository_service_tuf_worker.interfaces import IKeyVault, ServiceSettings
+from repository_service_tuf_worker.interfaces import (
+    Dynaconf,
+    IKeyVault,
+    ServiceSettings,
+)
 
 
 @dataclass
@@ -130,7 +134,7 @@ def _raw_key_parser(cls, path: str, keys: str) -> List[LocalKey]:
         return parsed_keys
 
     @classmethod
-    def configure(cls, settings) -> None:
+    def configure(cls, settings: Dynaconf) -> None:
         """
         Run actions to check and configure the service using the settings.
         """
@@ -165,12 +169,12 @@ def settings(cls) -> List[ServiceSettings]:
         """Define the settings parameters."""
         return [
             ServiceSettings(
-                name="LOCAL_KEYVAULT_PATH",
+                names=["LOCAL_KEYVAULT_PATH"],
                 argument="path",
                 required=True,
             ),
             ServiceSettings(
-                name="LOCAL_KEYVAULT_KEYS",
+                names=["LOCAL_KEYVAULT_KEYS"],
                 argument="keys",
                 required=True,
             ),

repository_service_tuf_worker/services/storage/local.py

@@ -19,13 +19,16 @@ def __init__(self, path: str) -> None:
 
     @classmethod
     def configure(cls, settings) -> None:
-        os.makedirs(settings.LOCAL_STORAGE_BACKEND_PATH, exist_ok=True)
+        path = settings.get("LOCAL_STORAGE_BACKEND_PATH") or settings.get(
+            "LOCAL_STORAGE_PATH"
+        )
+        os.makedirs(path, exist_ok=True)
 
     @classmethod
     def settings(cls) -> List[ServiceSettings]:
         return [
             ServiceSettings(
-                name="LOCAL_STORAGE_BACKEND_PATH",
+                names=["LOCAL_STORAGE_BACKEND_PATH", "LOCAL_STORAGE_PATH"],
                 argument="path",
                 required=True,
             ),

tests/unit/tuf_repository_service_worker/services/keyvault/test_local.py

@@ -302,12 +302,12 @@ def test_settings(self):
 
         assert service_settings == [
             local.ServiceSettings(
-                name="LOCAL_KEYVAULT_PATH",
+                names=["LOCAL_KEYVAULT_PATH"],
                 argument="path",
                 required=True,
             ),
             local.ServiceSettings(
-                name="LOCAL_KEYVAULT_KEYS",
+                names=["LOCAL_KEYVAULT_KEYS"],
                 argument="keys",
                 required=True,
             ),

tests/unit/tuf_repository_service_worker/services/storage/test_local.py

@@ -15,7 +15,10 @@ def test_basic_init(self):
         assert service._path == "/path"
 
     def test_configure(self):
-        test_settings = pretend.stub(LOCAL_STORAGE_BACKEND_PATH="/path")
+        test_settings = pretend.stub(
+            LOCAL_STORAGE_BACKEND_PATH="/path",
+            get=pretend.call_recorder(lambda *a: "/path"),
+        )
         local.os = pretend.stub(
             makedirs=pretend.call_recorder(lambda *a, **kw: None)
         )
@@ -26,14 +29,17 @@ def test_configure(self):
         assert local.os.makedirs.calls == [
             pretend.call("/path", exist_ok=True)
         ]
+        assert test_settings.get.calls == [
+            pretend.call("LOCAL_STORAGE_BACKEND_PATH")
+        ]
 
     def test_settings(self):
         service = local.LocalStorage("/path")
         service_settings = service.settings()
 
         assert service_settings == [
             local.ServiceSettings(
-                name="LOCAL_STORAGE_BACKEND_PATH",
+                names=["LOCAL_STORAGE_BACKEND_PATH", "LOCAL_STORAGE_PATH"],
                 argument="path",
                 required=True,
             ),