Add test for replication DataManager + populate Timing when using SetBalance API

apier/v1/accounts.go

@@ -571,6 +571,24 @@ func (self *ApierV1) SetBalance(attr *utils.AttrSetBalance, reply *string) error
 	if attr.TimingIds != nil {
 		a.Balance.TimingIDs = utils.StringMapPointer(utils.ParseStringMap(*attr.TimingIds))
 	}
+
+	if attr.TimingIds != nil && *attr.TimingIds != "" {
+		timingIds := strings.Split(*attr.TimingIds, utils.INFIELD_SEP)
+		for _, timingID := range timingIds {
+			timing, err := self.DataManager.GetTiming(timingID, false, utils.NonTransactional)
+			if err != nil {
+				return err
+			}
+			a.Balance.Timings = append(a.Balance.Timings, &engine.RITiming{
+				Years:     timing.Years,
+				Months:    timing.Months,
+				MonthDays: timing.MonthDays,
+				WeekDays:  timing.WeekDays,
+				StartTime: timing.StartTime,
+				EndTime:   timing.EndTime,
+			})
+		}
+	}
 	publishAction := &engine.Action{
 		ActionType: utils.MetaPublishBalance,
 	}

apier/v1/dm_remote_it_test.go

@@ -63,6 +63,7 @@ var sTestsInternalRemoteIT = []func(t *testing.T){
 	testInternalRemoteITGetAction,
 	testInternalRemoteITGetActionPlan,
 	testInternalRemoteITGetAccountActionPlan,
+	testInternalReplicationSetThreshold,
 	testInternalRemoteITKillEngine,
 }
 
@@ -571,6 +572,72 @@ func testInternalRemoteITGetAccountActionPlan(t *testing.T) {
 	}
 }
 
+func testInternalReplicationSetThreshold(t *testing.T) {
+	var reply *engine.ThresholdProfile
+	var result string
+	if err := internalRPC.Call("ApierV1.GetThresholdProfile",
+		&utils.TenantID{Tenant: "cgrates.org", ID: "THD_Replication"}, &reply); err == nil ||
+		err.Error() != utils.ErrNotFound.Error() {
+		t.Error(err)
+	}
+	tPrfl := &ThresholdWithCache{
+		ThresholdProfile: &engine.ThresholdProfile{
+			Tenant:    "cgrates.org",
+			ID:        "THD_Replication",
+			FilterIDs: []string{"*string:~Account:1001", "*string:~CustomField:CustomValue"},
+			ActivationInterval: &utils.ActivationInterval{
+				ActivationTime: time.Date(2014, 7, 14, 14, 35, 0, 0, time.UTC),
+			},
+			MaxHits:   -1,
+			MinSleep:  time.Duration(5 * time.Minute),
+			Blocker:   false,
+			Weight:    20.0,
+			ActionIDs: []string{"ACT_1"},
+			Async:     true,
+		},
+	}
+	if err := internalRPC.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
+		t.Error(err)
+	} else if result != utils.OK {
+		t.Error("Unexpected reply returned", result)
+	}
+
+	if err := internalRPC.Call("ApierV1.GetThresholdProfile",
+		&utils.TenantID{Tenant: "cgrates.org", ID: "THD_Replication"}, &reply); err != nil {
+		t.Error(err)
+	} else if !reflect.DeepEqual(tPrfl.ThresholdProfile, reply) {
+		t.Errorf("Expecting: %+v, received: %+v", tPrfl.ThresholdProfile, reply)
+	}
+
+	// verify threshold profile in replication dataDB
+	if rcv, err := rmtDM.GetThresholdProfile("cgrates.org", "THD_Replication",
+		false, false, utils.NonTransactional); err != nil {
+		t.Error(err)
+	} else if !reflect.DeepEqual(tPrfl.ThresholdProfile, rcv) {
+		t.Errorf("Expecting: %+v, received: %+v", tPrfl.ThresholdProfile, rcv)
+	}
+	//
+	eIdxes := map[string]utils.StringMap{
+		"*string:~Account:1001": {
+			"THD_ACNT_1001":   true,
+			"THD_Replication": true,
+		},
+		"*string:~Account:1002": {
+			"THD_ACNT_1002": true,
+		},
+		"*string:~CustomField:CustomValue": {
+			"THD_Replication": true,
+		},
+	}
+	if rcvIdx, err := rmtDM.GetFilterIndexes(
+		utils.PrefixToIndexCache[utils.ThresholdProfilePrefix], tPrfl.Tenant,
+		utils.EmptyString, nil); err != nil {
+		t.Error(err)
+	} else if !reflect.DeepEqual(eIdxes, rcvIdx) {
+		t.Errorf("Expecting %+v, received: %+v", eIdxes, rcvIdx)
+	}
+}
+
 func testInternalRemoteITKillEngine(t *testing.T) {
 	if err := engine.KillEngine(100); err != nil {
 		t.Error(err)

data/conf/samples/remote_mongo/cgrates.json

@@ -25,6 +25,13 @@
 			"db_name": "10",
 			"db_port": 27017
 		}
+	],
+	"replication_conns": [
+		{
+			"db_type": "mongo",
+			"db_name": "10",
+			"db_port": 27017
+		}
 	]
 },
 

data/conf/samples/remote_redis/cgrates.json

@@ -25,6 +25,13 @@
 			"db_port": 6379,
 			"db_name": "10",
 		}
+	],
+	"replication_conns": [
+		{
+			"db_type": "*redis",
+			"db_port": 6379,
+			"db_name": "10",
+		}
 	]
 },
 

engine/balance_filter.go

@@ -152,6 +152,12 @@ func (bf *BalanceFilter) LoadFromBalance(b *Balance) *BalanceFilter {
 	if !b.TimingIDs.IsEmpty() {
 		bf.TimingIDs = &b.TimingIDs
 	}
+	if len(b.Timings) != 0 {
+		bf.Timings = make([]*RITiming, len(b.Timings))
+		for i, timing := range b.Timings {
+			bf.Timings[i] = timing
+		}
+	}
 	if len(b.Factor) != 0 {
 		bf.Factor = &b.Factor
 	}
@@ -320,6 +326,12 @@ func (bf *BalanceFilter) ModifyBalance(b *Balance) {
 	if bf.TimingIDs != nil {
 		b.TimingIDs = *bf.TimingIDs
 	}
+	if bf.Timings != nil && len(bf.Timings) != 0 {
+		b.Timings = make([]*RITiming, len(bf.Timings))
+		for i, timing := range bf.Timings {
+			b.Timings[i] = timing
+		}
+	}
 	if bf.Weight != nil {
 		b.Weight = *bf.Weight
 	}