fix: populate scope with empty values for outputs of skipped/omitted steps

Signed-off-by: Bartek Kowalczyk <kowalczyk.bartek.mail@gmail.com>
Co-Authored-By: Bartek Kowalczyk <kowalczyk.bartek.mail@gmail.com>
Signed-off-by: isubasinghe <isitha@pipekit.io>

Author: isubasinghe

test/e2e/functional/steps-skipped-output-ref.yaml

@@ -0,0 +1,26 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: steps-skipped-output-ref-
+spec:
+  entrypoint: main
+  templates:
+    - name: main
+      steps:
+        - - name: job1
+            template: run-job
+        - - name: job2
+            template: run-job
+            when: "\"{{steps.job1.outputs.parameters.status}}\" == \"FAILED\""
+        - - name: job2-error-handler
+            template: run-job
+            when: "\"{{steps.job2.outputs.parameters.status}}\" != \"SUCCESS\" && \"{{steps.job1.outputs.parameters.status}}\" == \"SUCCESS\""
+    - name: run-job
+      outputs:
+        parameters:
+          - name: status
+            valueFrom:
+              path: /tmp/status.txt
+      container:
+        image: argoproj/argosay:v2
+        args: ["echo", "SUCCESS", "/tmp/status.txt"]

test/e2e/functional_test.go

@@ -512,6 +512,32 @@ func (s *FunctionalSuite) TestDAGSkippedOutputRef() {
 		})
 }
 
+func (s *FunctionalSuite) TestStepsSkippedOutputRef() {
+	s.Given().
+		Workflow("@functional/steps-skipped-output-ref.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow().
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
+			nodeJob1 := status.Nodes.FindByDisplayName("job1")
+			if assert.NotNil(t, nodeJob1) {
+				assert.Equal(t, wfv1.NodeSucceeded, nodeJob1.Phase)
+			}
+			// job2 is skipped because job1 succeeded (not failed)
+			nodeJob2 := status.Nodes.FindByDisplayName("job2")
+			if assert.NotNil(t, nodeJob2) {
+				assert.Equal(t, wfv1.NodeSkipped, nodeJob2.Phase)
+			}
+			// job2-error-handler runs: job2 output resolves to "" so "" != "SUCCESS" is true
+			nodeHandler := status.Nodes.FindByDisplayName("job2-error-handler")
+			if assert.NotNil(t, nodeHandler) {
+				assert.Equal(t, wfv1.NodeSucceeded, nodeHandler.Phase)
+			}
+		})
+}
+
 func (s *FunctionalSuite) TestStepsWhenExprFilter() {
 	s.Given().
 		Workflow("@functional/steps-when-expr-filter.yaml").

workflow/controller/steps.go

@@ -163,6 +163,20 @@ func (woc *wfOperationCtx) executeSteps(ctx context.Context, nodeName string, tm
 				woc.buildLocalScope(stepsCtx.scope, prefix, sgNode)
 			} else {
 				woc.buildLocalScope(stepsCtx.scope, prefix, childNode)
+
+				if (childNode.Phase == wfv1.NodeSkipped || childNode.Phase == wfv1.NodeOmitted) && childNode.Outputs == nil {
+					var resolvedTmpl *wfv1.Template
+					_, resolvedTmpl, _, err = stepsCtx.tmplCtx.ResolveTemplate(ctx, &step)
+					if err == nil && resolvedTmpl != nil {
+						for _, param := range resolvedTmpl.Outputs.Parameters {
+							key := fmt.Sprintf("%s.outputs.parameters.%s", prefix, param.Name)
+							stepsCtx.scope.addParamToScope(key, "")
+						}
+						if resolvedTmpl.Outputs.Result != nil {
+							stepsCtx.scope.addParamToScope(fmt.Sprintf("%s.outputs.result", prefix), "")
+						}
+					}
+				}
 			}
 		}
 	}