vdk-plugin-control-cli: add secrets command

projects/vdk-plugins/vdk-plugin-control-cli/src/vdk/plugin/control_cli_plugin/control_service_secrets_client.py

@@ -0,0 +1,38 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+import logging
+from typing import Dict
+
+from vdk.api.plugin.plugin_input import ISecretsServiceClient
+from vdk.internal.control.rest_lib.factory import ApiClientFactory
+from vdk.plugin.control_cli_plugin.control_service_api_error_decorator import (
+    ConstrolServiceApiErrorDecorator,
+)
+
+log = logging.getLogger(__name__)
+
+
+class ControlServiceSecretsServiceClient(ISecretsServiceClient):
+    """Implementation of SecretsServiceClient which connects to VDK Control
+    Service Secrets API."""
+
+    def __init__(self, rest_api_url: str):
+        self.__secrets_api = ApiClientFactory(rest_api_url).get_secrets_api()
+        log.debug(f"Initialized Secrets against {rest_api_url}.")
+
+    @ConstrolServiceApiErrorDecorator()
+    def read_secrets(self, job_name: str, team_name: str):
+        data = self.__secrets_api.data_job_secrets_read(
+            team_name=team_name, job_name=job_name, deployment_id="TODO"
+        )
+        return data
+
+    @ConstrolServiceApiErrorDecorator()
+    def write_secrets(self, job_name: str, team_name: str, secrets: Dict) -> Dict:
+        self.__secrets_api.data_job_secrets_update(
+            team_name=team_name,
+            job_name=job_name,
+            deployment_id="TODO",
+            request_body=secrets,
+        )
+        return secrets

projects/vdk-plugins/vdk-plugin-control-cli/src/vdk/plugin/control_cli_plugin/secrets_plugin.py

@@ -0,0 +1,33 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+import logging
+
+from vdk.api.plugin.hook_markers import hookimpl
+from vdk.internal.builtin_plugins.run.job_context import JobContext
+from vdk.plugin.control_cli_plugin.control_service_configuration import (
+    ControlServiceConfiguration,
+)
+from vdk.plugin.control_cli_plugin.control_service_secrets_client import (
+    ControlServiceSecretsServiceClient,
+)
+
+log = logging.getLogger(__name__)
+
+
+@hookimpl
+def initialize_job(context: JobContext) -> None:
+    conf = ControlServiceConfiguration(context.core_context.configuration)
+    url = conf.control_service_rest_api_url()
+    if url:
+        log.info("Initialize Control Service based Secrets client implementation.")
+        context.secrets.set_secrets_factory_method(
+            "default", lambda: ControlServiceSecretsServiceClient(url)
+        )
+        context.secrets.set_secrets_factory_method(
+            "control-service", lambda: ControlServiceSecretsServiceClient(url)
+        )
+    else:
+        log.info(
+            "Control Service REST API URL is not configured. "
+            "Will not initialize Control Service based Secrets client implementation."
+        )

projects/vdk-plugins/vdk-plugin-control-cli/src/vdk/plugin/control_cli_plugin/vdk_plugin_control_cli.py

@@ -17,6 +17,7 @@
 from vdk.internal.control.command_groups.job.execute import execute
 from vdk.internal.control.command_groups.job.list import list_command
 from vdk.internal.control.command_groups.job.properties import properties_command
+from vdk.internal.control.command_groups.job.secrets import secrets_command
 from vdk.internal.control.command_groups.job.show import show_command
 from vdk.internal.control.command_groups.login_group.login import login
 from vdk.internal.control.command_groups.logout_group.logout import logout
@@ -45,6 +46,7 @@ def vdk_command_line(root_command: click.Group):
     root_command.add_command(show_command)
     root_command.add_command(properties_command)
     root_command.add_command(info)
+    root_command.add_command(secrets_command)
 
     plugins = control_plugin_manager.Plugins()
     default_options = DefaultOptions(plugins)

projects/vdk-plugins/vdk-plugin-control-cli/tests/jobs/secrets-job/10_secrets_step.py

@@ -0,0 +1,9 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from vdk.api.job_input import IJobInput
+
+
+def run(job_input: IJobInput):
+    secrets = job_input.get_all_secrets()
+    secrets["new"] = secrets["original"] + 1
+    job_input.set_all_secrets(secrets)

projects/vdk-plugins/vdk-plugin-control-cli/tests/jobs/secrets-job/config.ini

@@ -0,0 +1,2 @@
+[owner]
+team = test-team

projects/vdk-plugins/vdk-plugin-control-cli/tests/test_secrets_plugin.py

@@ -0,0 +1,74 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+import json
+import os
+from unittest import mock
+
+from click.testing import Result
+from pytest_httpserver.pytest_plugin import PluginHTTPServer
+from vdk.plugin.control_cli_plugin import secrets_plugin
+from vdk.plugin.control_cli_plugin import vdk_plugin_control_cli
+from vdk.plugin.test_utils.util_funcs import cli_assert_equal
+from vdk.plugin.test_utils.util_funcs import CliEntryBasedTestRunner
+from vdk.plugin.test_utils.util_funcs import jobs_path_from_caller_directory
+from werkzeug import Request
+from werkzeug import Response
+
+
+def test_secrets_plugin(httpserver: PluginHTTPServer):
+    api_url = httpserver.url_for("")
+    secrets_data = {"original": 1}
+
+    httpserver.expect_request(
+        method="GET",
+        uri=f"/data-jobs/for-team/test-team/jobs/secrets-job/deployments/TODO/secrets",
+    ).respond_with_handler(
+        lambda r: Response(status=200, response=json.dumps(secrets_data, indent=4))
+    )
+
+    def update_secrets_data(req: Request):
+        secrets_data.clear()
+        secrets_data.update(req.get_json())
+        return Response(status=200)
+
+    httpserver.expect_request(
+        method="PUT",
+        uri=f"/data-jobs/for-team/test-team/jobs/secrets-job/deployments/TODO/secrets",
+    ).respond_with_handler(update_secrets_data)
+
+    with mock.patch.dict(
+        os.environ,
+        {
+            "VDK_CONTROL_SERVICE_REST_API_URL": api_url,
+            "VDK_secrets_DEFAULT_TYPE": "control-service",
+        },
+    ):
+        runner = CliEntryBasedTestRunner(vdk_plugin_control_cli, secrets_plugin)
+
+        result: Result = runner.invoke(
+            ["run", jobs_path_from_caller_directory("secrets-job")]
+        )
+        cli_assert_equal(0, result)
+        assert secrets_data == {"original": 1, "new": 2}
+
+
+def test_secrets_plugin_no_url_configured():
+    with mock.patch.dict(
+        os.environ,
+        {
+            "VDK_secrets_DEFAULT_TYPE": "control-service",
+        },
+    ):
+        runner = CliEntryBasedTestRunner(vdk_plugin_control_cli, secrets_plugin)
+
+        result: Result = runner.invoke(
+            ["run", jobs_path_from_caller_directory("simple-job")]
+        )
+        # job that do not use secrets should succeed
+        cli_assert_equal(0, result)
+
+        result: Result = runner.invoke(
+            ["run", jobs_path_from_caller_directory("secrets-job")]
+        )
+        # but jobs that do use secrets should fail.
+        cli_assert_equal(1, result)