Add tests for pipeline selector in elasticsearch output

libbeat/outputs/elasticsearch/client.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"strings"
 	"time"
 
 	"go.elastic.co/apm"
@@ -352,7 +353,7 @@ func getPipeline(event *beat.Event, pipelineSel *outil.Selector) (string, error)
 			return "", errors.New("pipeline metadata is no string")
 		}
 
-		return pipeline, nil
+		return strings.ToLower(pipeline), nil
 	}
 
 	if pipelineSel != nil {

libbeat/outputs/elasticsearch/elasticsearch.go

@@ -133,13 +133,7 @@ func buildSelectors(
 		return index, pipeline, err
 	}
 
-	pipelineSel, err := outil.BuildSelectorFromConfig(cfg, outil.Settings{
-		Key:              "pipeline",
-		MultiKey:         "pipelines",
-		EnableSingleOnly: true,
-		FailEmpty:        false,
-		Case:             outil.SelectorLowerCase,
-	})
+	pipelineSel, err := buildPipelineSelector(cfg)
 	if err != nil {
 		return index, pipeline, err
 	}
@@ -150,3 +144,13 @@ func buildSelectors(
 
 	return index, pipeline, err
 }
+
+func buildPipelineSelector(cfg *common.Config) (outil.Selector, error) {
+	return outil.BuildSelectorFromConfig(cfg, outil.Settings{
+		Key:              "pipeline",
+		MultiKey:         "pipelines",
+		EnableSingleOnly: true,
+		FailEmpty:        false,
+		Case:             outil.SelectorLowerCase,
+	})
+}

libbeat/outputs/elasticsearch/elasticsearch_test.go

@@ -21,6 +21,8 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/esleg/eslegclient"
 )
 
@@ -73,3 +75,59 @@ func TestGlobalConnectCallbacksManagement(t *testing.T) {
 		t.Fatalf("third callback cannot be retrieved")
 	}
 }
+
+func TestPipelineSelection(t *testing.T) {
+	cases := map[string]struct {
+		cfg   map[string]interface{}
+		event beat.Event
+		want  string
+	}{
+		"no pipline configured": {},
+		"pipeline configured": {
+			cfg:  map[string]interface{}{"pipeline": "test"},
+			want: "test",
+		},
+		"pipeline must be lowercase": {
+			cfg:  map[string]interface{}{"pipeline": "Test"},
+			want: "test",
+		},
+		"pipeline via event meta": {
+			event: beat.Event{Meta: common.MapStr{"pipeline": "test"}},
+			want:  "test",
+		},
+		"pipeline via event meta must be lowercase": {
+			event: beat.Event{Meta: common.MapStr{"pipeline": "Test"}},
+			want:  "test",
+		},
+		"pipelines setting": {
+			cfg: map[string]interface{}{
+				"pipelines": []map[string]interface{}{{"pipeline": "test"}},
+			},
+			want: "test",
+		},
+		"pipelines setting must be lowercase": {
+			cfg: map[string]interface{}{
+				"pipelines": []map[string]interface{}{{"pipeline": "Test"}},
+			},
+			want: "test",
+		},
+	}
+
+	for name, test := range cases {
+		t.Run(name, func(t *testing.T) {
+			selector, err := buildPipelineSelector(common.MustNewConfigFrom(test.cfg))
+			if err != nil {
+				t.Fatalf("Failed to parse configuration: %v", err)
+			}
+
+			got, err := getPipeline(&test.event, &selector)
+			if err != nil {
+				t.Fatalf("Failed to create pipeline name: %v", err)
+			}
+
+			if test.want != got {
+				t.Errorf("Pipeline name missmatch (want: %v, got: %v)", test.want, got)
+			}
+		})
+	}
+}