vdk-test-utils: add vdk sdk secrets api - part 2 (#2320)

Incremental change for adding the secrets capability.

Adopt secrets interface in test utils.

Actual tests and full implementation will follow in next MR.

Signed-off-by: Dako Dakov <ddakov@vmware.com>

Author: dakodakov

projects/vdk-plugins/vdk-test-utils/src/vdk/plugin/test_utils/util_funcs.py

@@ -41,6 +41,7 @@
 from vdk.internal.core.statestore import StateStore
 from vdk.internal.plugin.plugin import PluginRegistry
 from vdk.plugin.test_utils.util_plugins import TestPropertiesPlugin
+from vdk.plugin.test_utils.util_plugins import TestSecretsPlugin
 
 
 def cli_assert(
@@ -121,7 +122,7 @@ def __init__(self, *plugins):
         :param plugins: the list of plugins that should be loaded during this test run.
         """
         self._plugins = plugins
-        self._default_plugins = [TestPropertiesPlugin()]
+        self._default_plugins = [TestPropertiesPlugin(), TestSecretsPlugin()]
 
     def clear_default_plugins(self):
         self._default_plugins.clear()

projects/vdk-plugins/vdk-test-utils/src/vdk/plugin/test_utils/util_plugins.py

@@ -13,6 +13,7 @@
 from vdk.api.plugin.hook_markers import hookimpl
 from vdk.api.plugin.plugin_input import IIngesterPlugin
 from vdk.api.plugin.plugin_input import IPropertiesServiceClient
+from vdk.api.plugin.plugin_input import ISecretsServiceClient
 from vdk.internal.builtin_plugins.connection.decoration_cursor import DecorationCursor
 from vdk.internal.builtin_plugins.connection.managed_connection_base import (
     ManagedConnectionBase,
@@ -148,6 +149,46 @@ def initialize_job(self, context: JobContext) -> None:
         )
 
 
+class TestSecretsServiceClient(ISecretsServiceClient):
+    """Testing secrets client that keeps in memory per job properties."""
+
+    def __init__(self):
+        self._secrets = {}
+
+    def read_secrets(self, job_name: str, team_name: str) -> Dict:
+        res = deepcopy(self._secrets.get(job_name, {}))
+        return res
+
+    def write_secrets(self, job_name: str, team_name: str, secrets: Dict) -> Dict:
+        self._secrets[job_name] = deepcopy(secrets)
+        return self._secrets[job_name]
+
+
+class TestSecretsPlugin:
+    def __init__(self):
+        self.secrets_client = TestSecretsServiceClient()
+
+    @hookimpl
+    def initialize_job(self, context: JobContext) -> None:
+        context.secrets.set_secrets_factory_method(
+            "default", lambda: self.secrets_client
+        )
+
+
+class TestSecretsDecoratedPlugin(ISecretsServiceClient):
+    def read_secrets(self, job_name: str, team_name: str) -> Dict:
+        raise NotImplementedError()
+
+    def write_secrets(self, job_name: str, team_name: str, secrets: Dict) -> Dict:
+        return {**secrets, **{"test": "True"}}
+
+    @hookimpl
+    def initialize_job(self, context: JobContext) -> None:
+        context.secrets.set_secrets_factory_method(
+            "test-secret-decorated", lambda: self
+        )
+
+
 class IngestIntoMemoryPlugin(IIngesterPlugin):
     """
     Create a new ingestion mechanism to ingest data into memory