sync from main (#125)

* sync from main
conductor-sdk · Nov 12, 2023 · d1508c5 · d1508c5
1 parent 4d69012
commit d1508c5
Show file tree

Hide file tree

Showing 28 changed files with 3,865 additions and 248 deletions.
diff --git a/docs/README.md b/docs/README.md
@@ -2,4 +2,5 @@
 ### [Workflow Management](executor.md)
 ### [Task Worker](worker.md)
 ### [Workflow Authoring](workflow.md)
-### [Settings](settings.md)
+### [Settings](settings.md)
+### [Conductor Client](client.md)
diff --git a/docs/client.md b/docs/client.md
diff --git a/internal/testdata/kitchensink.go b/internal/testdata/kitchensink.go
@@ -58,6 +58,18 @@ func NewKitchenSinkWorkflow(executor *executor.WorkflowExecutor) *workflow.Condu
 			workflow.NewSimpleTask("simple_task", "simple_task_5"),
 		},
 	)
+	join := workflow.NewJoinTask("new_join_ref", "simple_task_fork_ref2", "simple_task_fork_ref4")
+	join.InputMap(map[string]interface{}{
+		"param1": "value",
+	})
+	forkWithJoin := workflow.NewForkTaskWithJoin("fork_with_join_fork_ref", join,
+		[]workflow.TaskInterface{
+			workflow.NewSimpleTask("simple_task", "simple_task_fork_ref1"),
+			workflow.NewSimpleTask("simple_task", "simple_task_fork_ref2"),
+		}, []workflow.TaskInterface{
+			workflow.NewSimpleTask("simple_task", "simple_task_fork_ref3"),
+			workflow.NewSimpleTask("simple_task", "simple_task_fork_ref4"),
+		})
 	dynamicFork := workflow.NewDynamicForkTask(
 		"dynamic_fork",
 		workflow.NewSimpleTask("dynamic_fork_prep", "dynamic_fork_prep"),
@@ -90,7 +102,8 @@ func NewKitchenSinkWorkflow(executor *executor.WorkflowExecutor) *workflow.Condu
 		Add(setVariable).
 		Add(subWorkflow).
 		Add(dynamicFork).
-		Add(fork)
+		Add(fork).
+		Add(forkWithJoin)
 
 	return workflow
 }

diff --git a/internal/testdata/util.go b/internal/testdata/util.go
@@ -50,6 +50,9 @@ var (
 	EventClient = client.EventResourceApiService{
 		APIClient: apiClient,
 	}
+	TagsClient = client.TagsApiService{
+		APIClient: apiClient,
+	}
 )
 
 var TaskRunner = worker.NewTaskRunnerWithApiClient(apiClient)
@@ -59,7 +62,7 @@ var WorkflowExecutor = executor.NewWorkflowExecutor(apiClient)
 func init() {
 	log.SetFormatter(&log.JSONFormatter{})
 	log.SetOutput(os.Stdout)
-	log.SetLevel(log.DebugLevel)
+	log.SetLevel(log.ErrorLevel)
 }
 
 func ValidateWorkflowDaemon(waitTime time.Duration, outputChannel chan error, workflowId string, expectedOutput map[string]interface{}, expectedStatus model.WorkflowStatus) {
@@ -224,6 +227,19 @@ func ValidateWorkflowRegistration(workflow *workflow.ConductorWorkflow) error {
 	return fmt.Errorf("exhausted retries")
 }
 
+func ValidateWorkflowDeletion(workflow *workflow.ConductorWorkflow) error {
+	for attempt := 0; attempt < 5; attempt += 1 {
+		err := workflow.UnRegister()
+		if err != nil {
+			time.Sleep(time.Duration(attempt+2) * time.Second)
+			fmt.Println("Failed to validate workflow deletion, reason: " + err.Error())
+			continue
+		}
+		return nil
+	}
+	return fmt.Errorf("exhausted retries")
+}
+
 func isWorkflowCompleted(workflow *model.Workflow, expectedStatus model.WorkflowStatus) bool {
 	return workflow.Status == expectedStatus
 }