CH-295 Support Kubernetes Secret references in workflow tasks

libraries/cloudharness-common/cloudharness/workflows/tasks.py

@@ -176,11 +176,17 @@ def command(self):
 
 
 class CustomTask(ContainerizedTask):
-    def __init__(self, name, image_name, resources={}, image_pull_policy='IfNotPresent', command=None, **env_args):
+    def __init__(self, name, image_name, resources={}, image_pull_policy='IfNotPresent', command=None,
+                 secret_envs=None, **env_args):
+        self.secret_envs = list(secret_envs or ())
         super().__init__(name, resources=resources,
                          image_pull_policy=image_pull_policy, command=command, **env_args)
         self.__image_name = get_image_full_tag(image_name)
 
+    @property
+    def envs(self):
+        return super().envs + self.secret_envs
+
     @property
     def image_name(self):
         return self.__image_name

libraries/cloudharness-common/tests/test_workflow.py

@@ -26,6 +26,36 @@ def check_wf(wf):
     assert "spec" in wf
 
 
+def test_custom_task_supports_secret_environment_entries():
+    secret_environment = [
+        {
+            "name": name,
+            "valueFrom": {
+                "secretKeyRef": {
+                    "name": "database-credentials",
+                    "key": key,
+                }
+            },
+        }
+        for name, key in (("DB_USER", "username"), ("DB_PASS", "password"))
+    ]
+    task = tasks.CustomTask(
+        'secret-task', 'busybox', secret_envs=secret_environment,
+        DB_HOST='database',
+    )
+    workflow = operations.SingleTaskOperation('secret-operation-', task).to_workflow()
+    environment = {
+        item['name']: item
+        for item in workflow['spec']['templates'][0]['container']['env']
+    }
+
+    assert environment['DB_HOST']['value'] == 'database'
+    assert environment['DB_USER'] == secret_environment[0]
+    assert environment['DB_PASS'] == secret_environment[1]
+    assert 'secret_envs' not in environment
+    assert 'value' not in environment['DB_PASS']
+
+
 def test_volume_affinity_check():
     assert not utils.volume_requires_affinity("a")
     assert utils.volume_requires_affinity("a:b")