From 665e2072f51c4c5095ee1baf7ddc405121d8f598 Mon Sep 17 00:00:00 2001
From: urso <steffen.siering@elastic.co>
Date: Tue, 9 Jun 2020 13:37:19 +0200
Subject: [PATCH] Add tests for redis key selection

---
 libbeat/outputs/redis/redis.go      | 18 +++++----
 libbeat/outputs/redis/redis_test.go | 61 +++++++++++++++++++++++++++++
 2 files changed, 72 insertions(+), 7 deletions(-)

diff --git a/libbeat/outputs/redis/redis.go b/libbeat/outputs/redis/redis.go
index 23b5f137550a..910b69d9f580 100644
--- a/libbeat/outputs/redis/redis.go
+++ b/libbeat/outputs/redis/redis.go
@@ -92,13 +92,7 @@ func makeRedis(
 		return outputs.Fail(errors.New("Bad Redis data type"))
 	}
 
-	key, err := outil.BuildSelectorFromConfig(cfg, outil.Settings{
-		Key:              "key",
-		MultiKey:         "keys",
-		EnableSingleOnly: true,
-		FailEmpty:        true,
-		Case:             outil.SelectorKeepCase,
-	})
+	key, err := buildKeySelector(cfg)
 	if err != nil {
 		return outputs.Fail(err)
 	}
@@ -175,3 +169,13 @@ func makeRedis(
 
 	return outputs.SuccessNet(config.LoadBalance, config.BulkMaxSize, config.MaxRetries, clients)
 }
+
+func buildKeySelector(cfg *common.Config) (outil.Selector, error) {
+	return outil.BuildSelectorFromConfig(cfg, outil.Settings{
+		Key:              "key",
+		MultiKey:         "keys",
+		EnableSingleOnly: true,
+		FailEmpty:        true,
+		Case:             outil.SelectorKeepCase,
+	})
+}
diff --git a/libbeat/outputs/redis/redis_test.go b/libbeat/outputs/redis/redis_test.go
index 5ca91d3fef01..deffb1d8de52 100644
--- a/libbeat/outputs/redis/redis_test.go
+++ b/libbeat/outputs/redis/redis_test.go
@@ -118,3 +118,64 @@ func TestMakeRedis(t *testing.T) {
 		})
 	}
 }
+
+func TestKeySelection(t *testing.T) {
+	cases := map[string]struct {
+		cfg   map[string]interface{}
+		event beat.Event
+		want  string
+	}{
+		"key configured": {
+			cfg:  map[string]interface{}{"key": "test"},
+			want: "test",
+		},
+		"key must keep case": {
+			cfg:  map[string]interface{}{"key": "Test"},
+			want: "Test",
+		},
+		"key setting": {
+			cfg: map[string]interface{}{
+				"keys": []map[string]interface{}{{"key": "test"}},
+			},
+			want: "test",
+		},
+		"keys setting must keep case": {
+			cfg: map[string]interface{}{
+				"keys": []map[string]interface{}{{"key": "Test"}},
+			},
+			want: "Test",
+		},
+		"use event field": {
+			cfg: map[string]interface{}{"key": "test-%{[field]}"},
+			event: beat.Event{
+				Fields: common.MapStr{"field": "from-event"},
+			},
+			want: "test-from-event",
+		},
+		"use event field must keep case": {
+			cfg: map[string]interface{}{"key": "Test-%{[field]}"},
+			event: beat.Event{
+				Fields: common.MapStr{"field": "From-Event"},
+			},
+			want: "Test-From-Event",
+		},
+	}
+
+	for name, test := range cases {
+		t.Run(name, func(t *testing.T) {
+			selector, err := buildKeySelector(common.MustNewConfigFrom(test.cfg))
+			if err != nil {
+				t.Fatalf("Failed to parse configuration: %v", err)
+			}
+
+			got, err := selector.Select(&test.event)
+			if err != nil {
+				t.Fatalf("Failed to create key name: %v", err)
+			}
+
+			if test.want != got {
+				t.Errorf("Pipeline name missmatch (want: %v, got: %v)", test.want, got)
+			}
+		})
+	}
+}