fix(controller): Fix bug in util/RecoverWorkflowNameFromSelectorStrin…

…g. Add error handling (#3596)

server/workflow/workflow_server.go

@@ -143,7 +143,10 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
 	opts := &metav1.ListOptions{}
 	if req.ListOptions != nil {
 		opts = req.ListOptions
-		wfName := argoutil.RecoverWorkflowNameFromSelectorString(opts.FieldSelector)
+		wfName, err := argoutil.RecoverWorkflowNameFromSelectorString(opts.FieldSelector)
+		if err != nil {
+			return err
+		}
 		if len(wfName) > 0 {
 			wf, err := s.getWorkflow(wfClient, req.Namespace, wfName, metav1.GetOptions{})
 			if err != nil {

util/util.go

@@ -96,14 +96,22 @@ func RecoverIndexFromNodeName(name string) int {
 
 func GenerateFieldSelectorFromWorkflowName(wfName string) string {
 	result := fields.ParseSelectorOrDie(fmt.Sprintf("metadata.name=%s", wfName)).String()
-	if compare := RecoverWorkflowNameFromSelectorString(result); wfName != compare {
+	compare, err := RecoverWorkflowNameFromSelectorString(result)
+	if err != nil {
+		log.WithFields(log.Fields{"wfName": wfName}).Error(err)
+	}
+	if wfName != compare {
 		panic(fmt.Sprintf("Could not recover field selector from workflow name. Expected '%s' but got '%s'\n", wfName, compare))
 	}
 	return result
 }
 
-func RecoverWorkflowNameFromSelectorString(selector string) string {
+func RecoverWorkflowNameFromSelectorString(selector string) (string, error) {
 	nameIndex := strings.Index(selector, "=")
+	prefix := selector[:nameIndex]
+	if prefix != "metadata.name" {
+		return "", fmt.Errorf("Incorrect prefix. Expected 'metadata.name' but got '%s'", prefix)
+	}
 	name := selector[nameIndex+1:]
-	return name
+	return name, nil
 }

util/util_test.go

@@ -2,6 +2,8 @@ package util
 
 import (
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestGenerateFieldSelectorFromWorkflowName(t *testing.T) {
@@ -34,12 +36,21 @@ func TestRecoverWorkflowNameFromSelectorString(t *testing.T) {
 		want string
 	}{
 		{"TestRecoverWorkflowNameFromSelectorString", args{"metadata.name=whalesay"}, "whalesay"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"metadata.name="}, ""},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := RecoverWorkflowNameFromSelectorString(tt.args.selector); got != tt.want {
+			got, err := RecoverWorkflowNameFromSelectorString(tt.args.selector)
+			assert.NoError(t, err)
+			if got != tt.want {
 				t.Errorf("RecoverWorkflowNameFromSelectorString() = %v, want %v", got, tt.want)
 			}
 		})
 	}
 }
+
+func TestRecoverWorkflowNameFromSelectorStringError(t *testing.T) {
+	name, err := RecoverWorkflowNameFromSelectorString("whatever=whalesay")
+	assert.NotNil(t, err)
+	assert.Equal(t, name, "")
+}