CH-297 Handle missing registry Secrets when generating workflows

libraries/cloudharness-common/cloudharness/utils/config.py

@@ -70,7 +70,10 @@ def get_registry_name(cls):
 
     @classmethod
     def get_registry_secret(cls):
-        return cls.get_configuration()['registry'].get('secret', {}).get('name', '')
+        secret = cls.get_configuration()['registry'].get('secret') or {}
+        if not isinstance(secret, dict):
+            return ''
+        return secret.get('name') or ''
 
     @classmethod
     def is_secured(cls):

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

@@ -120,13 +120,13 @@ def to_workflow(self, **arguments):
         }
 
     def spec(self):
+        registry_secret = config.CloudharnessConfig.get_registry_secret()
         spec = {
             'entrypoint': self.entrypoint,
             'ttlStrategy': self.ttl_strategy,
             'templates': [self.modify_template(template) for template in self.templates],
             'tolerations': [V1Toleration(key='cloudharness/temporary-job', operator='Equal', value='true').to_dict()],
             'serviceAccountName': SERVICE_ACCOUNT,
-            'imagePullSecrets': [{'name': config.CloudharnessConfig.get_registry_secret()}],
             'podGC': self.pod_gc or {
                 'strategy': 'OnWorkflowSuccess',
                 'deleteDelayDuration': self.ttl_strategy['secondsAfterSuccess'] if self.ttl_strategy else "600s",
@@ -140,6 +140,9 @@ def spec(self):
             }]
         }
 
+        if registry_secret:
+            spec['imagePullSecrets'] = [{'name': registry_secret}]
+
         if is_accounts_present():
             spec['volumes'].append({
                 'name': 'cloudharness-kc-accounts',

libraries/cloudharness-common/tests/test_workflow.py

@@ -1,4 +1,8 @@
 """Notice, this test needs a fully operating kubernetes with argo environment in the container running the test"""
+from unittest.mock import patch
+
+import pytest
+
 from cloudharness.utils.config import CloudharnessConfig
 from cloudharness.workflows import argo_service
 from cloudharness import set_debug
@@ -26,6 +30,42 @@ def check_wf(wf):
     assert "spec" in wf
 
 
+@pytest.mark.parametrize('secret', [None, {}, {'name': ''}, {'name': None}, ''])
+def test_unconfigured_registry_secret(secret):
+    with patch.object(
+        CloudharnessConfig, 'get_configuration',
+        return_value={'registry': {'secret': secret}},
+    ):
+        assert CloudharnessConfig.get_registry_secret() == ''
+
+
+def test_missing_registry_secret():
+    with patch.object(CloudharnessConfig, 'get_configuration', return_value={'registry': {}}):
+        assert CloudharnessConfig.get_registry_secret() == ''
+
+
+def test_configured_registry_secret():
+    with patch.object(
+        CloudharnessConfig, 'get_configuration',
+        return_value={'registry': {'secret': {'name': 'private-registry'}}},
+    ):
+        assert CloudharnessConfig.get_registry_secret() == 'private-registry'
+
+
+@pytest.mark.parametrize('registry_secret', ['', 'private-registry'])
+def test_workflow_optional_image_pull_secret(registry_secret):
+    task = tasks.CustomTask('registry-task', 'busybox')
+    operation = operations.SingleTaskOperation('registry-operation-', task)
+
+    with patch.object(CloudharnessConfig, 'get_registry_secret', return_value=registry_secret):
+        workflow = operation.to_workflow()
+
+    if registry_secret:
+        assert workflow['spec']['imagePullSecrets'] == [{'name': registry_secret}]
+    else:
+        assert 'imagePullSecrets' not in workflow['spec']
+
+
 def test_volume_affinity_check():
     assert not utils.volume_requires_affinity("a")
     assert utils.volume_requires_affinity("a:b")

tools/deployment-cli-tools/tests/test_helm.py

@@ -1704,6 +1704,8 @@ def test_find_chart_images_resolves_empty_tag_from_app_version():
     assert find_chart_images({'image': {'repository': 'a/b', 'tag': ''}}) == [
         ChartImageRef(path=('image',), value={'repository': 'a/b'}),
     ]
+
+
 def test_instances_expand_into_applications(tmp_path):
     """An instance is deployed as an application of its own, inheriting the parent's configuration."""
     out_folder = tmp_path / 'test_instances_expand_into_applications'