[draft] Fix bundle/config/root.go

bundle/config/root.go

@@ -3,6 +3,7 @@ package config
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
@@ -15,6 +16,7 @@ import (
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/convert"
 	"github.com/databricks/cli/libs/dyn/dynloc"
+	"github.com/databricks/cli/libs/dyn/dynvar"
 	"github.com/databricks/cli/libs/dyn/merge"
 	"github.com/databricks/cli/libs/dyn/yamlloader"
 	"github.com/databricks/cli/libs/log"
@@ -130,6 +132,10 @@ func LoadFromBytes(path string, raw []byte) (*Root, diag.Diagnostics) {
 	if err != nil {
 		return nil, diag.Errorf("failed to rewrite %s: %v", path, err)
 	}
+	v, err = rewriteForEachTaskInputs(v)
+	if err != nil {
+		return nil, diag.FromErr(fmt.Errorf("failed to rewrite for_each_task.inputs in %s: %w", path, err))
+	}
 
 	// Normalize dynamic configuration tree according to configuration type.
 	v, diags := convert.Normalize(r, v)
@@ -552,6 +558,36 @@ func rewriteShorthands(v dyn.Value) (dyn.Value, error) {
 	}))
 }
 
+func rewriteForEachTaskInputs(v dyn.Value) (dyn.Value, error) {
+	return dyn.Walk(v, func(p dyn.Path, input dyn.Value) (dyn.Value, error) {
+		if !isForEachTaskInputsPath(p) || input.Kind() != dyn.KindSequence {
+			return input, nil
+		}
+
+		data, err := json.Marshal(input.AsAny())
+		if err != nil {
+			return dyn.InvalidValue, fmt.Errorf("%s at %s: %w", p, input.Location(), err)
+		}
+		// Leave unresolved references for normal validation instead of interpolating into JSON text.
+		if dynvar.ContainsVariableReference(string(data)) {
+			return input, nil
+		}
+		return dyn.NewValue(string(data), input.Locations()), nil
+	})
+}
+
+func isForEachTaskInputsPath(p dyn.Path) bool {
+	if len(p) == 9 && (p[0] == dyn.Key("targets") || p[0] == dyn.Key("environments")) && p[1].Key() != "" {
+		p = p[2:]
+	}
+	return len(p) == 7 &&
+		p.HasPrefix(dyn.NewPath(dyn.Key("resources"), dyn.Key("jobs"))) &&
+		p[2].Key() != "" &&
+		p[3] == dyn.Key("tasks") &&
+		p[4].Key() == "" &&
+		p.HasSuffix(dyn.NewPath(dyn.Key("for_each_task"), dyn.Key("inputs")))
+}
+
 // validateVariableOverrides checks that all variables specified
 // in the target override are also defined in the root.
 func validateVariableOverrides(root, target dyn.Value) (err error) {

bundle/config/root_test.go

@@ -31,6 +31,58 @@ func TestRootLoad(t *testing.T) {
 	assert.Equal(t, "basic", root.Bundle.Name)
 }
 
+func TestLoadForEachTaskInputs(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		inputs string
+		want   string
+	}{
+		{name: "YAML array", inputs: "[input_1, input_2]", want: `["input_1","input_2"]`},
+		{name: "YAML object array", inputs: "[{name: input_1, enabled: true}]", want: `[{"enabled":true,"name":"input_1"}]`},
+		{name: "empty YAML array", inputs: "[]", want: `[]`},
+		{name: "JSON string", inputs: `'[ "input_1", "input_2" ]'`, want: `[ "input_1", "input_2" ]`},
+		{name: "variable reference string", inputs: `${var.items}`, want: `${var.items}`},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			config := "bundle:\n  name: foreach\nresources:\n  jobs:\n    fan:\n      tasks:\n        - task_key: fan\n          for_each_task:\n            inputs: " + tc.inputs + "\n            task:\n              task_key: inner\n"
+			root, diags := LoadFromBytes("databricks.yml", []byte(config))
+			require.NoError(t, diags.Error())
+			inputs := root.Resources.Jobs["fan"].Tasks[0].ForEachTask.Inputs
+			require.Equal(t, tc.want, inputs)
+
+			body, err := json.Marshal(root.Resources.Jobs["fan"].Tasks[0].ForEachTask)
+			require.NoError(t, err)
+			var apiTask struct {
+				Inputs string `json:"inputs"`
+			}
+			require.NoError(t, json.Unmarshal(body, &apiTask))
+			assert.Equal(t, tc.want, apiTask.Inputs)
+		})
+	}
+}
+
+func TestLoadForEachTaskInputsInTarget(t *testing.T) {
+	config := `bundle:
+  name: foreach
+targets:
+  dev:
+    resources:
+      jobs:
+        fan:
+          tasks:
+            - task_key: fan
+              for_each_task:
+                inputs: [input_1, input_2]
+                task:
+                  task_key: inner
+`
+	root, diags := LoadFromBytes("databricks.yml", []byte(config))
+	require.NoError(t, diags.Error())
+	require.Equal(t, `["input_1","input_2"]`, root.Targets["dev"].Resources.Jobs["fan"].Tasks[0].ForEachTask.Inputs)
+	require.NoError(t, root.MergeTargetOverrides("dev"))
+	assert.Equal(t, `["input_1","input_2"]`, root.Resources.Jobs["fan"].Tasks[0].ForEachTask.Inputs)
+}
+
 func TestInitializeVariables(t *testing.T) {
 	fooDefault := "abc"
 	root := &Root{