fix(conf): fix config overwritten by environment issue

Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>

internal/conf/connect_selector.go

@@ -5,8 +5,6 @@ import (
 	"strings"
 )
 
-const CONNECTION_CONF = "connections/connection.yaml"
-
 type ConSelector struct {
 	ConnSelectorStr string
 	Type            string // mqtt edgex
@@ -19,7 +17,7 @@ func (c *ConSelector) Init() error {
 		return fmt.Errorf("not a valid connection selector : %s", c.ConnSelectorStr)
 	}
 	c.Type = strings.ToLower(conTypeSel[0])
-	c.CfgKey = strings.ToLower(conTypeSel[1])
+	c.CfgKey = conTypeSel[1]
 	return nil
 }
 
@@ -34,7 +32,9 @@ func (c *ConSelector) ReadCfgFromYaml() (props map[string]interface{}, err error
 	if len(cfg) == 0 {
 		return nil, fmt.Errorf("fail to parse yaml for connection Type %s", c.Type)
 	} else {
-		if cons, found := cfg[c.CfgKey]; found {
+		if cons, found := cfg[strings.ToLower(c.CfgKey)]; found {
+			props = cons
+		} else if cons, found := cfg[c.CfgKey]; found {
 			props = cons
 		} else {
 			return nil, fmt.Errorf("not found connection Type and Selector:  %s.%s", c.Type, c.CfgKey)

internal/conf/connect_selector_test.go

@@ -0,0 +1,81 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package conf
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestConSelector_ReadCfgFromYaml(t *testing.T) {
+	type fields struct {
+		ConnSelectorStr string
+		Type            string
+		CfgKey          string
+	}
+	tests := []struct {
+		name      string
+		fields    fields
+		wantProps map[string]interface{}
+		wantErr   bool
+	}{
+		{
+			name: "mqtt localConnection",
+			fields: fields{
+				ConnSelectorStr: "mqtt.localconnection",
+				Type:            "mqtt",
+				CfgKey:          "localConnection",
+			},
+			wantProps: map[string]interface{}{
+				"username": "ekuiper",
+				"password": "password",
+				"server":   "tcp://127.0.0.1:1883",
+			},
+			wantErr: false,
+		},
+		{
+			name: "edgex redisMsgBus",
+			fields: fields{
+				ConnSelectorStr: "edgex.redismsgbus",
+				Type:            "edgex",
+				CfgKey:          "redisMsgBus",
+			},
+			wantProps: map[string]interface{}{
+				"protocol": "redis",
+				"port":     6379,
+				"server":   "127.0.0.1",
+				"type":     "redis",
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &ConSelector{
+				ConnSelectorStr: tt.fields.ConnSelectorStr,
+				Type:            tt.fields.Type,
+				CfgKey:          tt.fields.CfgKey,
+			}
+			gotProps, err := c.ReadCfgFromYaml()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("ReadCfgFromYaml() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(gotProps, tt.wantProps) {
+				t.Errorf("ReadCfgFromYaml() gotProps = %v, want %v", gotProps, tt.wantProps)
+			}
+		})
+	}
+}

internal/topo/node/conf/source.go

@@ -38,10 +38,18 @@ func GetSourceConf(sourceType string, options *ast.Options) map[string]interface
 		} else {
 			props = def
 		}
+		//config keys in etc folder will transform to lowercase
+		//while those in data will not
 		if c, ok := cfg[strings.ToLower(confkey)]; ok {
 			for k, v := range c {
 				props[k] = v
 			}
+		} else if c, ok := cfg[confkey]; ok {
+			for k, v := range c {
+				props[k] = v
+			}
+		} else {
+			conf.Log.Warnf("fail to find config key %s for source %s", confkey, sourceType)
 		}
 	}
 	f := options.FORMAT

internal/topo/node/conf/source_test.go

@@ -0,0 +1,71 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package conf
+
+import (
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"reflect"
+	"testing"
+)
+
+func TestGetSourceConf(t *testing.T) {
+	type args struct {
+		sourceType string
+		options    *ast.Options
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string]interface{}
+	}{
+		{
+			name: "default",
+			args: args{
+				sourceType: "mqtt",
+				options: &ast.Options{
+					CONF_KEY: "",
+				},
+			},
+			want: map[string]interface{}{
+				"qos":    1,
+				"server": "tcp://127.0.0.1:1883",
+				"format": "json",
+				"key":    "",
+			},
+		},
+		{
+			name: "demo_conf",
+			args: args{
+				sourceType: "mqtt",
+				options: &ast.Options{
+					CONF_KEY: "Demo_conf",
+				},
+			},
+			want: map[string]interface{}{
+				"qos":    0,
+				"server": "tcp://10.211.55.6:1883",
+				"format": "json",
+				"key":    "",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := GetSourceConf(tt.args.sourceType, tt.args.options); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("GetSourceConf() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}