fix: Load step settings from prefixed environment variables in Argo backend (#194)

Author: sam-hey

tests/step/backend_test.py

@@ -14,7 +14,8 @@
 
 from wurzel.backend.backend_argo import ArgoBackend, ArgoBackendSettings, EnvVar
 from wurzel.datacontract.common import MarkdownDataContract
-from wurzel.step import Settings, TypedStep
+from wurzel.step import NoSettings, Settings, TypedStep
+from wurzel.steps.manual_markdown import ManualMarkdownStep
 from wurzel.utils.meta_settings import WZ
 
 
@@ -102,6 +103,34 @@ def test_env_vars_in_task_container(argo_backend: ArgoBackend):
     assert len(templates[2]["container"]["env"]) == 3
 
 
+def test_create_envs_from_step_settings_loads_prefixed_env(argo_backend: ArgoBackend, env, tmp_path):
+    step = WZ(ManualMarkdownStep)
+    folder = tmp_path / "docs"
+    folder.mkdir()
+    env.set("MANUALMARKDOWNSTEP__FOLDER_PATH", str(folder))
+
+    envs = argo_backend._create_envs_from_step_settings(step)
+    env_dict = {e.name: e.value for e in envs}
+
+    assert env_dict["MANUALMARKDOWNSTEP__FOLDER_PATH"] == str(folder)
+
+
+def test_create_envs_from_step_with_no_settings(argo_backend: ArgoBackend):
+    """Verify that steps with NoSettings don't cause a crash and return no envs."""
+
+    class NoSettingsStep(TypedStep[NoSettings, None, MarkdownDataContract]):
+        def run(self, inpt: None) -> MarkdownDataContract:
+            return super().run(inpt)
+
+    step = WZ(NoSettingsStep)
+
+    # This call should not raise an exception
+    envs = argo_backend._create_envs_from_step_settings(step)
+
+    # Expect an empty list of environment variables
+    assert envs == []
+
+
 def test_argo_settings(env):
     settings = ArgoBackendSettings()
     assert settings.IMAGE == "ghcr.io/telekom/wurzel"

wurzel/backend/backend_argo.py

@@ -5,16 +5,18 @@
 import logging
 from functools import cache
 from pathlib import Path
+from types import NoneType
 from typing import Any
 
 from hera.workflows import DAG, ConfigMapEnvFrom, Container, CronWorkflow, S3Artifact, SecretEnvFrom, Task, Workflow
 from hera.workflows.archive import NoneArchiveStrategy
 from hera.workflows.models import EnvVar, SecurityContext
-from pydantic import Field, SecretStr
+from pydantic import Field, SecretStr, ValidationError
 from pydantic_settings import SettingsConfigDict
 
 from wurzel.backend.backend import Backend
 from wurzel.cli import generate_cli_call
+from wurzel.exceptions import EnvSettingsError
 from wurzel.step import TypedStep
 from wurzel.step.settings import SettingsBase, SettingsLeaf
 from wurzel.step_executor import BaseStepExecutor, PrometheusStepExecutor
@@ -98,7 +100,18 @@ def _create_envs_from_step_settings(self, step: "TypedStep[Any, Any, Any]") -> l
 
         env_vars = []
 
-        for field_name, field_value in step.settings_class().model_dump().items():
+        if step.settings_class == NoneType:
+            return env_vars
+
+        settings_cls = step.settings_class.with_prefix(f"{step.__class__.__name__.upper()}__")
+        try:
+            settings_instance = settings_cls()
+        except ValidationError as err:  # type: ignore [attr-defined]
+            e = EnvSettingsError(f"could not create {step.settings_class.__name__} from env for {step.__class__.__name__}")
+            e.add_note("To fix these issues setup env vars in the format <step_name>__<var>")
+            raise e from err
+
+        for field_name, field_value in settings_instance.model_dump().items():
             # Skip fields with sensitive keywords in their names
             if isinstance(field_value, SecretStr):
                 log.info(f"skipped config {field_name} due to secret detection")