Allow oneof in wait_for (#236)

* Upgraded go SDK and use oneof for wait-for OR

* Update comment

* Fix test case to account for new features with latest go SDK

go.mod

@@ -11,7 +11,7 @@ require (
 	go.flow.arcalot.io/dockerdeployer v0.7.3
 	go.flow.arcalot.io/expressions v0.4.5
 	go.flow.arcalot.io/kubernetesdeployer v0.10.0
-	go.flow.arcalot.io/pluginsdk v0.13.0
+	go.flow.arcalot.io/pluginsdk v0.14.0
 	go.flow.arcalot.io/podmandeployer v0.11.4
 	go.flow.arcalot.io/pythondeployer v0.6.2
 	go.flow.arcalot.io/testdeployer v0.6.2

go.sum

@@ -137,8 +137,8 @@ go.flow.arcalot.io/expressions v0.4.5 h1:GHRDHMkYIj2SN/TMc4aRApCewkJjl6moqhXjpdh
 go.flow.arcalot.io/expressions v0.4.5/go.mod h1:0Y2LgynO1SWA4bqsnKlCxqLME9zOR8tWKg3g+RG+FFQ=
 go.flow.arcalot.io/kubernetesdeployer v0.10.0 h1:4Fc6TsmM5pu1E0r9PzNu5AR1Q222B3Uo1Tqka1qqt8k=
 go.flow.arcalot.io/kubernetesdeployer v0.10.0/go.mod h1:OZTuKevUWxEHrlVYDE8+M5bHx1lJO6uwepKq+d3/k1w=
-go.flow.arcalot.io/pluginsdk v0.13.0 h1:bZqohrDkyAHsWmFJbyvPkjqUALPNJqObefVQrmYqUTw=
-go.flow.arcalot.io/pluginsdk v0.13.0/go.mod h1:YPVTOQ0BGn72RR4YkhsFXznaejfR5HN+or05t23Nqns=
+go.flow.arcalot.io/pluginsdk v0.14.0 h1:GtMaVjhTLIDHfeqnUn2CCJM5BdJF7OVO+/CEcxArDrE=
+go.flow.arcalot.io/pluginsdk v0.14.0/go.mod h1:TOuJdxpyCcLYW+yNUBVe2vs6wFBaJj/z9+44IR1GqCU=
 go.flow.arcalot.io/podmandeployer v0.11.4 h1:Oj0n1iW3X26dfTM2cgDUswtXNggVAmLpY5BUQqB8zBs=
 go.flow.arcalot.io/podmandeployer v0.11.4/go.mod h1:pc1gGXUAS8YeNkXLaVKApwmurqZ8VgrqZPakH8F/uUk=
 go.flow.arcalot.io/pythondeployer v0.6.2 h1:hIbDpdEhILArf8jMkqJplbxEnYYVcVslbH+IMdtXNy0=

internal/step/plugin/provider_test.go

@@ -293,12 +293,14 @@ func TestProvider_HappyError(t *testing.T) {
 	// unserialize malformed input schema
 	assert.Error(t, running.ProvideStageInput(
 		string(plugin.StageIDStarting),
-		map[string]any{"input": 1},
+		// The schema must be for the object with the field wait_time_ms, which is an integer.
+		map[string]any{"input": "not a valid value"},
 	))
 
 	waitTimeMs := 50
 	assert.NoError(t, running.ProvideStageInput(
 		string(plugin.StageIDStarting),
+		// It is also valid to inline waitTimeMs.
 		map[string]any{"input": map[string]any{"wait_time_ms": waitTimeMs}},
 	))
 

workflow/workflow_test.go

@@ -1410,6 +1410,90 @@ func TestGracefullyDisabledStepWorkflow(t *testing.T) {
 	assert.Equals(t, outputDataMap["result"], "disabled_wait_output")
 }
 
+var multiStepGracefullyDisabledStepWorkflow = `
+version: v0.2.0
+input:
+  root: WorkflowInput
+  objects:
+    WorkflowInput:
+      id: WorkflowInput
+      properties:
+        a_enabled:
+          type:
+            type_id: bool
+        b_enabled:
+          type:
+            type_id: bool
+steps:
+  a:
+    plugin:
+      src: "n/a"
+      deployment_type: "builtin"
+    step: wait
+    input:
+      wait_time_ms: 0
+    enabled: !expr $.input.a_enabled
+  b:
+    plugin:
+      src: "n/a"
+      deployment_type: "builtin"
+    step: wait
+    input:
+      wait_time_ms: 0
+    enabled: !expr $.input.b_enabled
+  simple_wait:
+    plugin:
+      src: "n/a"
+      deployment_type: "builtin"
+    step: wait
+    input:
+      wait_time_ms: 0
+    wait_for: !oneof
+      discriminator: "n/a"
+      one_of:
+        a: !expr $.steps.a.outputs.success
+        b: !expr $.steps.b.outputs.success
+outputs:
+  success:
+    a_output: !wait-optional $.steps.a.outputs.success
+    b_output: !wait-optional $.steps.b.outputs.success
+    simple_wait_output: !expr $.steps.simple_wait.outputs.success
+  all_disabled:
+    a_disabled: !expr $.steps.a.disabled.output
+    b_disabled: !expr $.steps.b.disabled.output
+`
+
+func TestMultiStepWaitForGracefullyDisabledStepWorkflow(t *testing.T) {
+	// Run a workflow where two steps can be disabled, and the second step only waits for one of either.
+	preparedWorkflow := assert.NoErrorR[workflow.ExecutableWorkflow](t)(
+		getTestImplPreparedWorkflow(t, multiStepGracefullyDisabledStepWorkflow),
+	)
+	outputID, _, err := preparedWorkflow.Execute(context.Background(), map[string]any{
+		"a_enabled": false,
+		"b_enabled": false,
+	})
+	assert.NoError(t, err)
+	assert.Equals(t, outputID, "all_disabled")
+	outputID, _, err = preparedWorkflow.Execute(context.Background(), map[string]any{
+		"a_enabled": true,
+		"b_enabled": false,
+	})
+	assert.NoError(t, err)
+	assert.Equals(t, outputID, "success")
+	outputID, _, err = preparedWorkflow.Execute(context.Background(), map[string]any{
+		"a_enabled": false,
+		"b_enabled": true,
+	})
+	assert.NoError(t, err)
+	assert.Equals(t, outputID, "success")
+	outputID, _, err = preparedWorkflow.Execute(context.Background(), map[string]any{
+		"a_enabled": true,
+		"b_enabled": true,
+	})
+	assert.NoError(t, err)
+	assert.Equals(t, outputID, "success")
+}
+
 var gracefullyDisabledForeachStepWorkflow = `
 version: v0.2.0
 input: