Updated integration tests

config/config_it_test.go

@@ -412,32 +412,48 @@ func testCGRConfigReloadERs(t *testing.T) {
 		SessionSConns: []string{utils.MetaLocalHost},
 		Readers: []*EventReaderCfg{
 			{
-				ID:              utils.MetaDefault,
-				Type:            utils.MetaNone,
-				RunDelay:        0,
-				ConcurrentReqs:  1024,
-				SourcePath:      "/var/spool/cgrates/ers/in",
-				ProcessedPath:   "/var/spool/cgrates/ers/out",
-				Filters:         []string{},
-				Flags:           flagsDefault,
-				Fields:          content,
-				CacheDumpFields: []*FCTemplate{},
-				Opts:            make(map[string]interface{}),
+				ID:                  utils.MetaDefault,
+				Type:                utils.MetaNone,
+				RunDelay:            0,
+				ConcurrentReqs:      1024,
+				SourcePath:          "/var/spool/cgrates/ers/in",
+				ProcessedPath:       "/var/spool/cgrates/ers/out",
+				Filters:             []string{},
+				Flags:               flagsDefault,
+				Fields:              content,
+				CacheDumpFields:     []*FCTemplate{},
+				PartialCommitFields: []*FCTemplate{},
+				Opts: map[string]interface{}{
+					"csvFieldSeparator":   ",",
+					"csvHeaderDefineChar": ":",
+					"csvRowLength":        0.,
+					"partialOrderField":   "~*req.AnswerTime",
+					"xmlRootPath":         "",
+				},
 			},
 			{
-				ID:              "file_reader1",
-				Type:            utils.MetaFileCSV,
-				RunDelay:        -1,
-				ConcurrentReqs:  1024,
-				SourcePath:      "/tmp/ers/in",
-				ProcessedPath:   "/tmp/ers/out",
-				Filters:         []string{},
-				Flags:           flags,
-				Fields:          content,
-				CacheDumpFields: []*FCTemplate{},
-				Opts:            make(map[string]interface{}),
+				ID:                  "file_reader1",
+				Type:                utils.MetaFileCSV,
+				RunDelay:            -1,
+				ConcurrentReqs:      1024,
+				SourcePath:          "/tmp/ers/in",
+				ProcessedPath:       "/tmp/ers/out",
+				Filters:             []string{},
+				Flags:               flags,
+				Fields:              content,
+				CacheDumpFields:     []*FCTemplate{},
+				PartialCommitFields: []*FCTemplate{},
+				Opts: map[string]interface{}{
+					"csvFieldSeparator":   ",",
+					"csvHeaderDefineChar": ":",
+					"csvRowLength":        0.,
+					"partialOrderField":   "~*req.AnswerTime",
+					"xmlRootPath":         "",
+				},
 			},
 		},
+		PartialCacheTTL:    time.Second,
+		PartialCacheAction: utils.MetaNone,
 	}
 	if !reflect.DeepEqual(expAttr, cfg.ERsCfg()) {
 		t.Errorf("Expected %s,\n received: %s ", utils.ToJSON(expAttr), utils.ToJSON(cfg.ERsCfg()))
@@ -596,51 +612,54 @@ func testCgrCfgV1ReloadConfigSection(t *testing.T) {
 		},
 	}
 	expected := map[string]interface{}{
-		"enabled": true,
+		"enabled":              true,
+		"partial_cache_action": "*none",
+		"partial_cache_ttl":    "1s",
+		"partial_path":         "",
 		"readers": []interface{}{
 			map[string]interface{}{
-				"partial_cache_expiry_action": "",
-				"partial_record_cache":        "0",
-				"cache_dump_fields":           []interface{}{},
-				"concurrent_requests":         1024,
-				"fields":                      content,
-				"field_separator":             ",",
-				"header_define_character":     ":",
-				"filters":                     []string{},
-				"flags":                       []string{},
-				"failed_calls_prefix":         "",
-				"id":                          "*default",
-				"processed_path":              "/var/spool/cgrates/ers/out",
-				"row_length":                  0,
-				"run_delay":                   "0",
-				"source_path":                 "/var/spool/cgrates/ers/in",
-				"tenant":                      "",
-				"timezone":                    "",
-				"type":                        utils.MetaNone,
-				"xml_root_path":               []interface{}{utils.EmptyString},
-				"opts":                        make(map[string]interface{}),
+				"id":                  utils.MetaDefault,
+				"cache_dump_fields":   []interface{}{},
+				"concurrent_requests": 1024,
+				"fields":              content,
+				"filters":             []string{},
+				"flags":               []string{},
+				"run_delay":           "0",
+				"source_path":         "/var/spool/cgrates/ers/in",
+				"processed_path":      "/var/spool/cgrates/ers/out",
+				"tenant":              "",
+				"timezone":            "",
+				"type":                utils.MetaNone,
+				"opts": map[string]interface{}{
+					"csvFieldSeparator":   ",",
+					"csvHeaderDefineChar": ":",
+					"csvRowLength":        0.,
+					"partialOrderField":   "~*req.AnswerTime",
+					"xmlRootPath":         "",
+				},
+				"partial_commit_fields": []interface{}{},
 			},
 			map[string]interface{}{
-				"cache_dump_fields":           []interface{}{},
-				"concurrent_requests":         1024,
-				"field_separator":             ",",
-				"header_define_character":     ":",
-				"filters":                     []string{},
-				"flags":                       []string{"*dryrun"},
-				"failed_calls_prefix":         "",
-				"partial_cache_expiry_action": "",
-				"partial_record_cache":        "0",
-				"id":                          "file_reader1",
-				"processed_path":              "/tmp/ers/out",
-				"row_length":                  0,
-				"run_delay":                   "-1",
-				"source_path":                 "/tmp/ers/in",
-				"tenant":                      "",
-				"timezone":                    "",
-				"type":                        "*file_csv",
-				"xml_root_path":               []interface{}{utils.EmptyString},
-				"fields":                      content,
-				"opts":                        make(map[string]interface{}),
+				"cache_dump_fields":   []interface{}{},
+				"concurrent_requests": 1024,
+				"filters":             []string{},
+				"flags":               []string{"*dryrun"},
+				"id":                  "file_reader1",
+				"processed_path":      "/tmp/ers/out",
+				"run_delay":           "-1",
+				"source_path":         "/tmp/ers/in",
+				"tenant":              "",
+				"timezone":            "",
+				"type":                "*file_csv",
+				"fields":              content,
+				"opts": map[string]interface{}{
+					"csvFieldSeparator":   ",",
+					"csvHeaderDefineChar": ":",
+					"csvRowLength":        0.,
+					"partialOrderField":   "~*req.AnswerTime",
+					"xmlRootPath":         "",
+				},
+				"partial_commit_fields": []interface{}{},
 			},
 		},
 		"sessions_conns": []string{

dispatchers/dispatchers.go

@@ -112,36 +112,37 @@ func (dS *DispatcherService) dispatcherProfileForEvent(tnt string, ev *utils.CGR
 		utils.MetaReq:  ev.Event,
 		utils.MetaOpts: ev.APIOpts,
 	}
-	prflIDs, err := engine.MatchingItemIDsForEvent(context.TODO(), evNm,
+	var prflIDs utils.StringSet
+	if prflIDs, err = engine.MatchingItemIDsForEvent(context.TODO(), evNm,
 		dS.cfg.DispatcherSCfg().StringIndexedFields,
 		dS.cfg.DispatcherSCfg().PrefixIndexedFields,
 		dS.cfg.DispatcherSCfg().SuffixIndexedFields,
 		dS.dm, utils.CacheDispatcherFilterIndexes, idxKeyPrfx,
 		dS.cfg.DispatcherSCfg().IndexedSelects,
 		dS.cfg.DispatcherSCfg().NestedFields,
-	)
-	if err != nil &&
+	); err != nil &&
 		err != utils.ErrNotFound {
-		return nil, err
+		return
 	}
 	if err == utils.ErrNotFound ||
 		dS.cfg.DispatcherSCfg().AnySubsystem {
-		dPrflAnyIDs, err := engine.MatchingItemIDsForEvent(context.TODO(), evNm,
+		var dPrflAnyIDs utils.StringSet
+		if dPrflAnyIDs, err = engine.MatchingItemIDsForEvent(context.TODO(), evNm,
 			dS.cfg.DispatcherSCfg().StringIndexedFields,
 			dS.cfg.DispatcherSCfg().PrefixIndexedFields,
 			dS.cfg.DispatcherSCfg().SuffixIndexedFields,
 			dS.dm, utils.CacheDispatcherFilterIndexes, anyIdxPrfx,
 			dS.cfg.DispatcherSCfg().IndexedSelects,
 			dS.cfg.DispatcherSCfg().NestedFields,
-		)
-		if prflIDs.Size() == 0 {
+		); prflIDs.Size() == 0 {
 			if err != nil { // return the error if no dispatcher matched the needed subsystem
-				return nil, err
+				return
 			}
 			prflIDs = dPrflAnyIDs
 		} else if err == nil && dPrflAnyIDs.Size() != 0 {
 			prflIDs = utils.JoinStringSet(prflIDs, dPrflAnyIDs)
 		}
+		err = nil // make sure we ignore the error from *any subsystem matching
 	}
 	for prflID := range prflIDs {
 		prfl, err := dS.dm.GetDispatcherProfile(tnt, prflID, true, true, utils.NonTransactional)
@@ -151,7 +152,8 @@ func (dS *DispatcherService) dispatcherProfileForEvent(tnt string, ev *utils.CGR
 			}
 			continue
 		}
-		if !(len(prfl.Subsystems) == 1 && prfl.Subsystems[0] == utils.MetaAny) &&
+
+		if (len(prfl.Subsystems) != 1 || prfl.Subsystems[0] != utils.MetaAny) &&
 			!utils.IsSliceMember(prfl.Subsystems, subsys) {
 			continue
 		}
@@ -166,7 +168,7 @@ func (dS *DispatcherService) dispatcherProfileForEvent(tnt string, ev *utils.CGR
 		}
 	}
 	if dPrlf == nil {
-		return nil, utils.ErrNotFound
+		err = utils.ErrNotFound
 	}
 	return
 }

loaders/loader.go

@@ -197,13 +197,9 @@ func (ldr *Loader) processFiles(ctx *context.Context, loaderType, caching, loadO
 	// based on load option will store or remove the content
 	switch loadOption {
 	case utils.MetaStore:
-		if err = ldr.processContent(ctx, loaderType, caching); err != nil {
-			return
-		}
+		return ldr.processContent(ctx, loaderType, caching)
 	case utils.MetaRemove:
-		if err = ldr.removeContent(ctx, loaderType, caching); err != nil {
-			return
-		}
+		return ldr.removeContent(ctx, loaderType, caching)
 	}
 	return
 }

loaders/loaders_it_test.go

@@ -58,6 +58,7 @@ func TestITLoaders(t *testing.T) {
 }
 
 func testV1LoadResource(t *testing.T) {
+	utils.Logger.SetLogLevel(7)
 	flPath := "/tmp/testV1LoadResource"
 	if err := os.MkdirAll(flPath, 0777); err != nil {
 		t.Error(err)
@@ -66,10 +67,21 @@ func testV1LoadResource(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	file.Write([]byte(`
-#Tenant[0],ID[1]
+	if _, err := file.Write([]byte(`#Tenant[0],ID[1]
 cgrates.org,NewRes1
-`))
+`)); err != nil {
+		t.Fatal(err)
+	}
+	if err := file.Sync(); err != nil {
+		t.Fatal(err)
+	}
+	if err := file.Close(); err != nil {
+		t.Fatal(err)
+	}
+	file, err = os.Create(path.Join(flPath, "res.lck"))
+	if err != nil {
+		t.Error(err)
+	}
 	file.Close()
 
 	data := engine.NewInternalDB(nil, nil, true)
@@ -81,7 +93,7 @@ cgrates.org,NewRes1
 		FieldSeparator: utils.FieldsSep,
 		TpInDir:        flPath,
 		TpOutDir:       "/tmp",
-		LockFileName:   utils.ResourcesCsv,
+		LockFileName:   "res.lck",
 		Data:           nil,
 	}
 	ldrs := NewLoaderService(dm, cfgLdr, "UTC", nil, nil)
@@ -100,10 +112,9 @@ cgrates.org,NewRes1
 		},
 	}
 
-	resCsv := `
-#Tenant[0],ID[1]
-cgrates.org,NewRes1
-`
+	resCsv := `#Tenant[0],ID[1]
+	cgrates.org,NewRes1
+	`
 	rdr := io.NopCloser(strings.NewReader(resCsv))
 	csvRdr := csv.NewReader(rdr)
 	csvRdr.Comment = '#'
@@ -133,8 +144,10 @@ cgrates.org,NewRes1
 	}
 
 	expRes := &engine.ResourceProfile{
-		Tenant: "cgrates.org",
-		ID:     "NewRes1",
+		Tenant:       "cgrates.org",
+		ID:           "NewRes1",
+		FilterIDs:    make([]string, 0),
+		ThresholdIDs: make([]string, 0),
 	}
 
 	if rcv, err := ldrs.ldrs["testV1LoadResource"].dm.GetResourceProfile(expRes.Tenant, expRes.ID,
@@ -144,7 +157,7 @@ cgrates.org,NewRes1
 		t.Errorf("Expected %+v \n, received %+v", utils.ToJSON(expRes), utils.ToJSON(rcv))
 	}
 
-	if err := os.Remove(flPath); err != nil {
+	if err := os.RemoveAll(flPath); err != nil {
 		t.Error(err)
 	}
 }