添加argo的监听demo

test/approve_former_test.go

@@ -2,8 +2,13 @@ package test
 
 import (
 	"fmt"
+	"git.sxidc.com/go-tools/argo-api"
 	"git.sxidc.com/go-tools/argo-api/approve_former"
+	"git.sxidc.com/go-tools/argo-api/client"
+	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"sync"
 	"testing"
+	"time"
 )
 
 func TestApproveFormer(t *testing.T) {
@@ -27,3 +32,263 @@ func TestApproveFormer(t *testing.T) {
 
 	fmt.Println(flowDefinition)
 }
+
+func TestUseApproveFormer(t *testing.T) {
+	err := argo.Init(kubeConfig)
+	if err != nil {
+		t.Fatalf("%+v\n", err)
+	}
+
+	defer argo.Destroy()
+
+	flowName := "flow-" + time.Now().Format("20060102150405")
+
+	flow := approve_former.NewFlow(flowName,
+		approve_former.NewSign("A"),
+		approve_former.NewOr(
+			approve_former.NewSign("B"),
+			approve_former.NewSign("C"),
+		),
+		approve_former.NewAnd(
+			approve_former.NewSign("D"),
+			approve_former.NewSign("E"),
+		),
+		approve_former.NewSign("F"),
+	)
+
+	flowDefinition, err := flow.Render(nil)
+	if err != nil {
+		t.Fatalf("%+v\n", err)
+	}
+
+	err = argo.GetInstance().CreateWorkflowTemplate(client.CreateWorkflowTemplateParams{
+		Namespace:          namespace,
+		TemplateDefinition: flowDefinition,
+	})
+	if err != nil {
+		t.Fatalf("%+v\n", err)
+	}
+
+	defer func() {
+		err := argo.GetInstance().DeleteWorkflowTemplate(client.DeleteWorkflowTemplateParams{
+			Namespace: namespace,
+			Name:      flowName,
+		})
+		if err != nil {
+			t.Fatalf("%+v\n", err)
+		}
+	}()
+
+	workflowName, err := argo.GetInstance().SubmitWorkflowFromWorkflowTemplate(client.SubmitWorkflowFromWorkflowTemplateParams{
+		Namespace:    namespace,
+		TemplateName: flowName,
+		Parameters: []string{
+			`data={"name": "yjp", "age": 37}`,
+		},
+	})
+	if err != nil {
+		t.Fatalf("%+v\n", err)
+	}
+
+	defer func() {
+		err := argo.GetInstance().DeleteWorkflow(client.DeleteWorkflowParams{
+			Namespace: namespace,
+			Name:      workflowName,
+		})
+		if err != nil {
+			t.Fatalf("%+v\n", err)
+		}
+	}()
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+
+	resumeMap := make(map[string]bool)
+
+	err = argo.GetInstance().RegisterWorkflowWatcher(namespace, workflowName, func(wf *v1alpha1.Workflow) {
+		if wf.Status.Nodes == nil {
+			return
+		}
+
+		for _, node := range wf.Status.Nodes {
+			if node.DisplayName == "A" && node.Phase == "Running" && node.Outputs.Parameters[0].Value == nil {
+				_, ok := resumeMap[node.DisplayName]
+				if ok {
+					return
+				}
+
+				err := argo.GetInstance().SetWorkflow(client.SetWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=A",
+					OutputParameters:  `{"approve": "通过"}`,
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				err = argo.GetInstance().ResumeWorkflow(client.ResumeWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=A",
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				resumeMap[node.DisplayName] = true
+			}
+
+			if node.DisplayName == "B" && node.Phase == "Running" && node.Outputs.Parameters[0].Value == nil {
+				_, ok := resumeMap[node.DisplayName]
+				if ok {
+					return
+				}
+
+				err := argo.GetInstance().SetWorkflow(client.SetWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=B",
+					OutputParameters:  `{"approve": "通过"}`,
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				err = argo.GetInstance().ResumeWorkflow(client.ResumeWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=B",
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				resumeMap[node.DisplayName] = true
+			}
+
+			if node.DisplayName == "C" && node.Phase == "Running" && node.Outputs.Parameters[0].Value == nil {
+				_, ok := resumeMap[node.DisplayName]
+				if ok {
+					return
+				}
+
+				err := argo.GetInstance().SetWorkflow(client.SetWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=C",
+					OutputParameters:  `{"approve": "拒绝"}`,
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				err = argo.GetInstance().ResumeWorkflow(client.ResumeWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=C",
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				resumeMap[node.DisplayName] = true
+			}
+
+			if node.DisplayName == "D" && node.Phase == "Running" && node.Outputs.Parameters[0].Value == nil {
+				_, ok := resumeMap[node.DisplayName]
+				if ok {
+					return
+				}
+
+				err := argo.GetInstance().SetWorkflow(client.SetWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=D",
+					OutputParameters:  `{"approve": "通过"}`,
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				err = argo.GetInstance().ResumeWorkflow(client.ResumeWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=D",
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				resumeMap[node.DisplayName] = true
+			}
+
+			if node.DisplayName == "E" && node.Phase == "Running" && node.Outputs.Parameters[0].Value == nil {
+				_, ok := resumeMap[node.DisplayName]
+				if ok {
+					return
+				}
+
+				err := argo.GetInstance().SetWorkflow(client.SetWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=E",
+					OutputParameters:  `{"approve": "通过"}`,
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				err = argo.GetInstance().ResumeWorkflow(client.ResumeWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=E",
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				resumeMap[node.DisplayName] = true
+			}
+
+			if node.DisplayName == "F" && node.Phase == "Running" && node.Outputs.Parameters[0].Value == nil {
+				_, ok := resumeMap[node.DisplayName]
+				if ok {
+					return
+				}
+
+				err := argo.GetInstance().SetWorkflow(client.SetWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=F",
+					OutputParameters:  `{"approve": "通过"}`,
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				err = argo.GetInstance().ResumeWorkflow(client.ResumeWorkflowParams{
+					Namespace:         namespace,
+					Name:              workflowName,
+					NodeFieldSelector: "displayName=F",
+				})
+				if err != nil {
+					t.Fatalf("%+v\n", err)
+				}
+
+				resumeMap[node.DisplayName] = true
+			}
+
+			if node.DisplayName == "F" && node.Phase == "Succeeded" && node.Outputs.Parameters[0].Value.String() == "通过" {
+				wg.Done()
+			}
+		}
+	})
+	if err != nil {
+		t.Fatalf("%+v\n", err)
+	}
+
+	defer argo.GetInstance().UnregisterWorkflowWatcher(namespace, workflowName)
+
+	wg.Wait()
+}