Make Namespace Behavior Public (#199)

* refactor Signed-off-by: Matthew F Leader <[email protected]> * add function description * add fn to exectuable workflow iface --------- Signed-off-by: Matthew F Leader <[email protected]>
arcalot · Jul 16, 2024 · c284ae0 · c284ae0
1 parent ee454cc
commit c284ae0
Show file tree

Hide file tree

Showing 4 changed files with 51 additions and 22 deletions.
diff --git a/internal/util/stringers.go b/internal/util/stringers.go
@@ -0,0 +1,17 @@
+package util
+
+import "go.flow.arcalot.io/pluginsdk/schema"
+
+// BuildNamespaceString creates a human-readable string representation of the
+// namespace scoped objects.
+func BuildNamespaceString(allNamespaces map[string]map[string]*schema.ObjectSchema) string {
+	availableObjects := ""
+	for namespace, objects := range allNamespaces {
+		availableObjects += "\n\t" + namespace + ":"
+		for objectID := range objects {
+			availableObjects += " " + objectID
+		}
+	}
+	availableObjects += "\n"
+	return availableObjects
+}
diff --git a/workflow/executor.go b/workflow/executor.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"go.flow.arcalot.io/engine/internal/util"
 	"reflect"
 	"strings"
 
@@ -71,6 +72,11 @@ type ExecutableWorkflow interface {
 
 	// OutputSchema returns the schema for the possible outputs of this workflow.
 	OutputSchema() map[string]*schema.StepOutputSchema
+
+	// Namespaces returns a namespaced collection of objects for the inputs
+	// and outputs of each stage in the step's lifecycles.
+	// It maps namespace id (path) to object id to object schema.
+	Namespaces() map[string]map[string]*schema.ObjectSchema
 }
 
 type executor struct {
@@ -121,7 +127,7 @@ func (e *executor) Prepare(workflow *Workflow, workflowContext map[string][]byte
 	}
 
 	// Apply step lifecycle objects to the input scope
-	err = applyLifecycleScopes(stepLifecycles, typedInput)
+	err = applyLifecycleNamespaces(stepLifecycles, typedInput)
 	if err != nil {
 		return nil, err
 	}
@@ -313,23 +319,30 @@ func (e *executor) processSteps(
 	return runnableSteps, stepOutputProperties, stepLifecycles, stepRunData, nil
 }
 
-func applyLifecycleScopes(
-	stepLifecycles map[string]step.Lifecycle[step.LifecycleStageWithSchema],
-	typedInput schema.Scope,
-) error {
+// BuildNamespaces creates a namespaced collection of objects for the inputs
+// and outputs of each stage in the step's lifecycles.
+// It maps namespace id (path) to object id to object schema.
+func BuildNamespaces(stepLifecycles map[string]step.Lifecycle[step.LifecycleStageWithSchema]) map[string]map[string]*schema.ObjectSchema {
 	allNamespaces := make(map[string]map[string]*schema.ObjectSchema)
 	for workflowStepID, stepLifecycle := range stepLifecycles {
 		for _, stage := range stepLifecycle.Stages {
 			prefix := "$.steps." + workflowStepID + "." + stage.ID + "."
 			// Apply inputs
 			// Example with stage "starting": $.steps.wait_step.starting.inputs.
-			addInputNamespacedScopes(allNamespaces, stage, prefix+"inputs.")
+			addInputNamespaces(allNamespaces, stage, prefix+"inputs.")
 			// Apply outputs
 			// Example with stage "outputs": $.steps.wait_step.outputs.outputs.
-			addOutputNamespacedScopes(allNamespaces, stage, prefix+"outputs.")
+			addOutputNamespaces(allNamespaces, stage, prefix+"outputs.")
 		}
 	}
-	return applyAllNamespaces(allNamespaces, typedInput)
+	return allNamespaces
+}
+
+func applyLifecycleNamespaces(
+	stepLifecycles map[string]step.Lifecycle[step.LifecycleStageWithSchema],
+	typedInput schema.Scope,
+) error {
+	return applyAllNamespaces(BuildNamespaces(stepLifecycles), typedInput)
 }
 
 // applyAllNamespaces applies all namespaces to the given scope.
@@ -344,28 +357,20 @@ func applyAllNamespaces(allNamespaces map[string]map[string]*schema.ObjectSchema
 		return nil // Success
 	}
 	// Now on the error path. Provide useful debug info.
-	availableObjects := ""
-	for namespace, objects := range allNamespaces {
-		availableObjects += "\n\t" + namespace + ":"
-		for objectID := range objects {
-			availableObjects += " " + objectID
-		}
-	}
-	availableObjects += "\n" // Since this is a multi-line error message, ending with a newline is clearer.
 	return fmt.Errorf(
 		"error validating references for workflow input (%w)\nAvailable namespaces and objects:%s",
 		err,
-		availableObjects,
+		util.BuildNamespaceString(allNamespaces),
 	)
 }
 
-func addOutputNamespacedScopes(allNamespaces map[string]map[string]*schema.ObjectSchema, stage step.LifecycleStageWithSchema, prefix string) {
+func addOutputNamespaces(allNamespaces map[string]map[string]*schema.ObjectSchema, stage step.LifecycleStageWithSchema, prefix string) {
 	for outputID, outputSchema := range stage.Outputs {
 		addScopesWithReferences(allNamespaces, outputSchema.Schema(), prefix+outputID)
 	}
 }
 
-func addInputNamespacedScopes(allNamespaces map[string]map[string]*schema.ObjectSchema, stage step.LifecycleStageWithSchema, prefix string) {
+func addInputNamespaces(allNamespaces map[string]map[string]*schema.ObjectSchema, stage step.LifecycleStageWithSchema, prefix string) {
 	for inputID, inputSchemaProperty := range stage.InputSchema {
 		var inputSchemaType = inputSchemaProperty.Type()
 		// Extract item values from lists (like for ForEach)

diff --git a/workflow/executor_unit_test.go b/workflow/executor_unit_test.go
@@ -61,7 +61,7 @@ var testLifecycleStage = step.LifecycleStageWithSchema{
 func TestAddOutputNamespacedScopes(t *testing.T) {
 	allNamespaces := make(map[string]map[string]*schema.ObjectSchema)
 	namespacePrefix := "TEST_PREFIX_"
-	addOutputNamespacedScopes(allNamespaces, testLifecycleStage, namespacePrefix)
+	addOutputNamespaces(allNamespaces, testLifecycleStage, namespacePrefix)
 	expectedNamespace := namespacePrefix + "outputC"
 	assert.Equals(t, len(allNamespaces), 1)
 	assert.MapContainsKey(t, expectedNamespace, allNamespaces)
@@ -76,7 +76,7 @@ func TestAddInputNamespacedScopes(t *testing.T) {
 		"listA":  "testObjectB",
 	}
 
-	addInputNamespacedScopes(allNamespaces, testLifecycleStage, namespacePrefix)
+	addInputNamespaces(allNamespaces, testLifecycleStage, namespacePrefix)
 	assert.Equals(t, len(allNamespaces), 2)
 	for expectedInput, expectedObject := range expectedInputs {
 		expectedNamespace := namespacePrefix + expectedInput
@@ -459,7 +459,7 @@ func TestApplyLifecycleScopes_Basic(t *testing.T) {
 			},
 		),
 	)
-	assert.NoError(t, applyLifecycleScopes(stepLifecycles, testScope))
+	assert.NoError(t, applyLifecycleNamespaces(stepLifecycles, testScope))
 	assert.NoError(t, testScope.ValidateReferences())
 	assert.NotNil(t, ref1Schema.GetObject())
 	assert.NotNil(t, ref2Schema.GetObject())

diff --git a/workflow/workflow.go b/workflow/workflow.go
@@ -55,6 +55,13 @@ func (e *executableWorkflow) DAG() dgraph.DirectedGraph[*DAGItem] {
 	return e.dag
 }
 
+// Namespaces returns a namespaced collection of objects for the inputs
+// and outputs of each stage in the step's lifecycles.
+// It maps namespace id (path) to object id to object schema.
+func (e *executableWorkflow) Namespaces() map[string]map[string]*schema.ObjectSchema {
+	return BuildNamespaces(e.lifecycles)
+}
+
 // Execute runs the workflow with the specified input. You can use the context variable to abort the workflow execution
 // (e.g. when the user presses Ctrl+C).
 func (e *executableWorkflow) Execute(ctx context.Context, serializedInput any) (outputID string, outputData any, err error) { //nolint:gocognit