Updated integration tests

apier/v2/cdrs_it_test.go

@@ -461,7 +461,7 @@ func testV2CDRsDifferentTenants(t *testing.T) {
 	//add a charger
 	chargerProfile := &v1.ChargerWithCache{
 		ChargerProfile: &engine.ChargerProfile{
-			Tenant: "cgrates.com",
+			Tenant: "CustomTenant",
 			ID:     "CustomCharger",
 			ActivationInterval: &utils.ActivationInterval{
 				ActivationTime: time.Date(2014, 7, 14, 14, 35, 0, 0, time.UTC),
@@ -480,7 +480,7 @@ func testV2CDRsDifferentTenants(t *testing.T) {
 	}
 	var reply2 *engine.ChargerProfile
 	if err := cdrsRpc.Call(utils.ApierV1GetChargerProfile,
-		&utils.TenantID{Tenant: "cgrates.com", ID: "CustomCharger"}, &reply2); err != nil {
+		&utils.TenantID{Tenant: "CustomTenant", ID: "CustomCharger"}, &reply2); err != nil {
 		t.Error(err)
 	} else if !reflect.DeepEqual(chargerProfile.ChargerProfile, reply2) {
 		t.Errorf("Expecting : %+v, received: %+v", chargerProfile.ChargerProfile, reply2)

data/conf/samples/ers/cgrates.json

@@ -5,7 +5,8 @@
 
 
 "general": {
-    "log_level": 7,
+	"log_level": 7,
+	"reply_timeout": "10s",									// consider connection down for replies taking longer than this value
 },
 
 
@@ -183,7 +184,7 @@
 			"id": "file_reader_with_filters",
 			"run_delay": -1,
 			"type": "*file_csv",
-			"flags": ["*dryrun"],
+			// "flags": ["*dryrun"],
 			"field_separator": ";",
 			"source_path": "/tmp/ers_with_filters/in",
 			"processed_path": "/tmp/ers_with_filters/out",

data/conf/samples/rpccaching/cgrates.json

@@ -2,6 +2,7 @@
 
 "general": {
 	"log_level": 7,
+	"reply_timeout": "10s",									// consider connection down for replies taking longer than this value
 },
 
 "listen": {

data/tariffplans/testtp/Actions.csv

@@ -9,8 +9,8 @@ TOPUP_DATA_r,*topup,,,,*monetary,,DATA_DEST,,,*unlimited,,5000000,10,false,false
 TOPUP_DATA_r,*topup,,,,*data,,DATA_DEST,datar,,*unlimited,,50000000000,10,false,false,10
 TOPUP_VOICE,*topup,,,,*voice,,GERMANY_MOBILE,,,*unlimited,,50000,10,false,false,10
 TOPUP_NEG,*topup,,,,*voice,,GERMANY;!GERMANY_MOBILE,*zero1m,,*unlimited,,100,10,false,false,10
-RPC,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""gob"",""Method"":""ApierV2.SetAccount"",""Attempts"":1,""Async"" :false,""Params"":{""Account"":""rpc"",""Tenant"":""cgrates.org""}}",,,,,,,,,,,,,,
+RPC,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""*gob"",""Method"":""ApierV2.SetAccount"",""Attempts"":1,""Async"" :false,""Params"":{""Account"":""rpc"",""Tenant"":""cgrates.org""}}",,,,,,,,,,,,,,
 DID,*debit,,,,*monetary,,*any,,,*unlimited,*any,"{""Method"":""*incremental"",""Params"":{""Units"":1, ""Interval"":""month"",""Increment"":""day""}}",10.0,,,10.0
 DID,*cdrlog,"{""action"":""^DID"",""prev_balance"":""BalanceValue""}",,,*monetary,,*any,,,*unlimited,,,10.0,,,10.0
-RPC_DEST,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""gob"",""Method"":""ApierV2.SetDestination"",""Attempts"":1,""Async"" :false,""Params"":{""Id"":""<<.Account.GetID>>"",""Prefixes"":[""1"",""2"",""3""]}}",,,,,,,,,,,,,,
-RPC_CDRSTATS,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""gob"",""Method"":""CDRStatsV1.AddQueue"",""Attempts"":1,""Async"" :false,""Params"":{""Id"":""qtest""}}",,,,,,,,,,,,,,
+RPC_DEST,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""*gob"",""Method"":""ApierV2.SetDestination"",""Attempts"":1,""Async"" :false,""Params"":{""Id"":""<<.Account.GetID>>"",""Prefixes"":[""1"",""2"",""3""]}}",,,,,,,,,,,,,,
+RPC_CDRSTATS,*cgr_rpc,"{""Address"": ""localhost:2013"",""Transport"":""*gob"",""Method"":""CDRStatsV1.AddQueue"",""Attempts"":1,""Async"" :false,""Params"":{""Id"":""qtest""}}",,,,,,,,,,,,,,

dispatchers/sessions_it_test.go

@@ -309,14 +309,14 @@ func testDspGetSessions(t *testing.T) {
 	if err := dispEngine.RPC.Call(utils.SessionSv1GetActiveSessionsCount,
 		&filtr, &reply); err != nil {
 		t.Fatal(err)
-	} else if reply != 2 {
-		t.Errorf("Expected 2 active sessions recived %v", reply)
+	} else if reply != 3 {
+		t.Errorf("Expected 3 active sessions recived %v", reply)
 	}
 	var rply []*sessions.ExternalSession
 	if err := dispEngine.RPC.Call(utils.SessionSv1GetActiveSessions,
 		&filtr, &rply); err != nil {
 		t.Fatal(err)
-	} else if len(rply) != 2 {
+	} else if len(rply) != 3 {
 		t.Errorf("Unexpected number of sessions returned %v :%s", len(rply), utils.ToJSON(rply))
 	}
 
@@ -719,8 +719,8 @@ func testDspSessionReplicate(t *testing.T) {
 	if err := allEngine2.RPC.Call(utils.SessionSv1GetPassiveSessionsCount,
 		new(utils.SessionFilter), &repl); err != nil {
 		t.Fatal(err)
-	} else if repl != 2 {
-		t.Errorf("Expected 1 sessions recived %v", repl)
+	} else if repl != 3 {
+		t.Errorf("Expected 3 sessions recived %v", repl)
 	}
 }
 
@@ -745,15 +745,15 @@ func testDspSessionPassive(t *testing.T) {
 	if err := dispEngine.RPC.Call(utils.SessionSv1GetActiveSessionsCount,
 		filtr, &repl); err != nil {
 		t.Fatal(err)
-	} else if repl != 2 {
-		t.Errorf("Expected 1 active sessions recived %v", repl)
+	} else if repl != 3 {
+		t.Errorf("Expected 3 active sessions recived %v", repl)
 	}
 
 	var rply []*sessions.ExternalSession
 	if err := dispEngine.RPC.Call(utils.SessionSv1GetActiveSessions,
 		&filtr, &rply); err != nil {
 		t.Fatal(err)
-	} else if len(rply) != 2 {
+	} else if len(rply) != 3 {
 		t.Errorf("Unexpected number of sessions returned %v :%s", len(rply), utils.ToJSON(rply))
 	}
 
@@ -847,15 +847,15 @@ func testDspSessionForceDisconect(t *testing.T) {
 	if err := dispEngine.RPC.Call(utils.SessionSv1GetActiveSessionsCount,
 		filtr, &repl); err != nil {
 		t.Fatal(err)
-	} else if repl != 2 {
-		t.Errorf("Expected 1 active sessions recived %v", repl)
+	} else if repl != 3 {
+		t.Errorf("Expected 3 active sessions recived %v", repl)
 	}
 
 	var rply []*sessions.ExternalSession
 	if err := dispEngine.RPC.Call(utils.SessionSv1GetActiveSessions,
 		&filtr, &rply); err != nil {
 		t.Fatal(err)
-	} else if len(rply) != 2 {
+	} else if len(rply) != 3 {
 		t.Errorf("Unexpected number of sessions returned %v :%s", len(rply), utils.ToJSON(rply))
 	}
 

engine/cdrs.go

@@ -209,6 +209,9 @@ var reqTypes = utils.NewStringSet([]string{utils.META_PSEUDOPREPAID, utils.META_
 
 // getCostFromRater will retrieve the cost from RALs
 func (cdrS *CDRServer) getCostFromRater(cdr *CDRWithArgDispatcher) (*CallCost, error) {
+	if cdrS.rals == nil {
+		return nil, utils.NewErrNotConnected(utils.RALService)
+	}
 	cc := new(CallCost)
 	var err error
 	timeStart := cdr.AnswerTime
@@ -449,6 +452,15 @@ func (cdrS *CDRServer) exportCDRs(cdrs []*CDR) (err error) {
 // processEvent processes a CGREvent based on arguments
 func (cdrS *CDRServer) processEvent(ev *utils.CGREventWithArgDispatcher,
 	chrgS, attrS, ralS, store, reRate, export, thdS, stS bool) (err error) {
+	if attrS {
+		if err = cdrS.attrSProcessEvent(ev); err != nil {
+			utils.Logger.Warning(
+				fmt.Sprintf("<%s> error: <%s> processing event %+v with %s",
+					utils.CDRs, err.Error(), utils.ToJSON(ev), utils.AttributeS))
+			err = utils.ErrPartiallyExecuted
+			return
+		}
+	}
 	var cgrEvs []*utils.CGREventWithArgDispatcher
 	if chrgS {
 		if cgrEvs, err = cdrS.chrgrSProcessEvent(ev); err != nil {
@@ -461,17 +473,6 @@ func (cdrS *CDRServer) processEvent(ev *utils.CGREventWithArgDispatcher,
 	} else { // ChargerS not requested, charge the original event
 		cgrEvs = []*utils.CGREventWithArgDispatcher{ev}
 	}
-	if attrS {
-		for _, cgrEv := range cgrEvs {
-			if err = cdrS.attrSProcessEvent(cgrEv); err != nil {
-				utils.Logger.Warning(
-					fmt.Sprintf("<%s> error: <%s> processing event %+v with %s",
-						utils.CDRs, err.Error(), cgrEv, utils.AttributeS))
-				err = utils.ErrPartiallyExecuted
-				return
-			}
-		}
-	}
 	// Check if the unique ID was not already processed
 	for _, cgrEv := range cgrEvs {
 		me := MapEvent(cgrEv.CGREvent.Event)
@@ -575,7 +576,7 @@ func (cdrS *CDRServer) processEvent(ev *utils.CGREventWithArgDispatcher,
 			utils.Logger.Warning(
 				fmt.Sprintf("<%s> error: <%s> exporting CDRs %+v",
 					utils.CDRs, err.Error(), cdrs))
-
+			partiallyExecuted = true
 		}
 	}
 	if thdS {
@@ -584,6 +585,7 @@ func (cdrS *CDRServer) processEvent(ev *utils.CGREventWithArgDispatcher,
 				utils.Logger.Warning(
 					fmt.Sprintf("<%s> error: <%s> processing event %+v with %s",
 						utils.CDRs, err.Error(), cgrEv, utils.ThresholdS))
+				partiallyExecuted = true
 			}
 		}
 	}
@@ -593,6 +595,7 @@ func (cdrS *CDRServer) processEvent(ev *utils.CGREventWithArgDispatcher,
 				utils.Logger.Warning(
 					fmt.Sprintf("<%s> error: <%s> processing event %+v with %s",
 						utils.CDRs, err.Error(), cgrEv, utils.StatS))
+				partiallyExecuted = true
 			}
 		}
 	}
@@ -714,13 +717,14 @@ func (cdrS *CDRServer) V1ProcessCDR(cdr *CDRWithArgDispatcher, reply *string) (e
 	return
 }
 
+// ArgV1ProcessEvent is the CGREvent with proccesing Flags
 type ArgV1ProcessEvent struct {
 	Flags []string
 	utils.CGREvent
 	*utils.ArgDispatcher
 }
 
-// V1ProcessCDR will process the CDR out of CGREvent
+// V1ProcessEvent will process the CGREvent
 func (cdrS *CDRServer) V1ProcessEvent(arg *ArgV1ProcessEvent, reply *string) (err error) {
 	if arg.CGREvent.ID == "" {
 		arg.CGREvent.ID = utils.GenUUID()
@@ -801,7 +805,7 @@ func (cdrS *CDRServer) V1ProcessEvent(arg *ArgV1ProcessEvent, reply *string) (er
 	return nil
 }
 
-// V1StoreSMCost handles storing of the cost into session_costs table
+// V1StoreSessionCost handles storing of the cost into session_costs table
 func (cdrS *CDRServer) V1StoreSessionCost(attr *AttrCDRSStoreSMCost, reply *string) (err error) {
 	if attr.Cost.CGRID == "" {
 		return utils.NewCGRError(utils.CDRSCtx,
@@ -895,6 +899,7 @@ func (cdrS *CDRServer) V2StoreSessionCost(args *ArgsV2CDRSStoreSMCost, reply *st
 
 }
 
+// ArgRateCDRs a cdr with extra flags
 type ArgRateCDRs struct {
 	Flags []string
 	utils.RPCCDRsFilter
@@ -952,7 +957,7 @@ func (cdrS *CDRServer) V1RateCDRs(arg *ArgRateCDRs, reply *string) (err error) {
 	return nil
 }
 
-// Used to process external CDRs
+// V1ProcessExternalCDR is used to process external CDRs
 func (cdrS *CDRServer) V1ProcessExternalCDR(eCDR *ExternalCDRWithArgDispatcher, reply *string) error {
 	cdr, err := NewCDRFromExternalCDR(eCDR.ExternalCDR,
 		cdrS.cgrCfg.GeneralCfg().DefaultTimezone)
@@ -972,11 +977,11 @@ func (cdrS *CDRServer) V1GetCDRs(args utils.RPCCDRsFilterWithArgDispatcher, cdrs
 		}
 		return err
 	}
-	if qryCDRs, _, err := cdrS.cdrDb.GetCDRs(cdrsFltr, false); err != nil {
+	qryCDRs, _, err := cdrS.cdrDb.GetCDRs(cdrsFltr, false)
+	if err != nil {
 		return utils.NewErrServerError(err)
-	} else {
-		*cdrs = qryCDRs
 	}
+	*cdrs = qryCDRs
 	return nil
 }
 
@@ -990,11 +995,11 @@ func (cdrS *CDRServer) V1CountCDRs(args *utils.RPCCDRsFilterWithArgDispatcher, c
 		return err
 	}
 	cdrsFltr.Count = true
-	if _, qryCnt, err := cdrS.cdrDb.GetCDRs(cdrsFltr, false); err != nil {
+	_, qryCnt, err := cdrS.cdrDb.GetCDRs(cdrsFltr, false)
+	if err != nil {
 		return utils.NewErrServerError(err)
-	} else {
-		*cnt = qryCnt
 	}
+	*cnt = qryCnt
 	return nil
 }
 

ers/filecsv_it_test.go

@@ -204,7 +204,7 @@ func testCsvITCheckSession(t *testing.T) {
 	aSessions := make([]*sessions.ExternalSession, 0)
 	if err := csvRPC.Call(utils.SessionSv1GetActiveSessions, &utils.SessionFilter{}, &aSessions); err != nil {
 		t.Error(err)
-	} else if len(aSessions) != 2 {
+	} else if len(aSessions) != 3 {
 		t.Errorf("wrong active sessions: %s \n , and len(aSessions) %+v", utils.ToJSON(aSessions), len(aSessions))
 	}
 }

general_tests/cdrs_onlexp_it_test.go

@@ -52,6 +52,7 @@ var (
 		testCDRsOnExpStartSlaveEngine,
 		testCDRsOnExpAMQPQueuesCreation,
 		testCDRsOnExpInitMasterRPC,
+		testCDRsOnExpLoadDefaultCharger,
 		testCDRsOnExpDisableOnlineExport,
 		testCDRsOnExpHttpCdrReplication,
 		testCDRsOnExpAMQPReplication,
@@ -85,6 +86,12 @@ func testCDRsOnExpInitConfig(t *testing.T) {
 
 // InitDb so we can rely on count
 func testCDRsOnExpInitCdrDb(t *testing.T) {
+	if err := engine.InitDataDb(cdrsMasterCfg); err != nil {
+		t.Fatal(err)
+	}
+	if err := engine.InitDataDb(cdrsSlaveCfg); err != nil {
+		t.Fatal(err)
+	}
 	if err := engine.InitStorDb(cdrsMasterCfg); err != nil {
 		t.Fatal(err)
 	}
@@ -149,6 +156,23 @@ func testCDRsOnExpInitMasterRPC(t *testing.T) {
 	}
 }
 
+func testCDRsOnExpLoadDefaultCharger(t *testing.T) {
+	// //add a default charger
+	chargerProfile := &engine.ChargerProfile{
+		Tenant:       "cgrates.org",
+		ID:           "Default",
+		RunID:        utils.MetaDefault,
+		AttributeIDs: []string{"*none"},
+		Weight:       20,
+	}
+	var result string
+	if err := cdrsMasterRpc.Call(utils.ApierV1SetChargerProfile, chargerProfile, &result); err != nil {
+		t.Error(err)
+	} else if result != utils.OK {
+		t.Error("Unexpected reply returned", result)
+	}
+}
+
 // Disable ExportCDR
 func testCDRsOnExpDisableOnlineExport(t *testing.T) {
 	// stop RabbitMQ server so we can test reconnects
@@ -198,20 +222,6 @@ func testCDRsOnExpDisableOnlineExport(t *testing.T) {
 }
 
 func testCDRsOnExpHttpCdrReplication(t *testing.T) {
-	//add a default charger
-	chargerProfile := &engine.ChargerProfile{
-		Tenant:       "cgrates.org",
-		ID:           "Default",
-		RunID:        utils.MetaDefault,
-		AttributeIDs: []string{"*none"},
-		Weight:       20,
-	}
-	var result string
-	if err := cdrsMasterRpc.Call(utils.ApierV1SetChargerProfile, chargerProfile, &result); err != nil {
-		t.Error(err)
-	} else if result != utils.OK {
-		t.Error("Unexpected reply returned", result)
-	}
 	testCdr1 := &engine.CDR{
 		CGRID:       utils.Sha1("httpjsonrpc1", time.Date(2013, 12, 7, 8, 42, 24, 0, time.UTC).String()),
 		ToR:         utils.VOICE,

general_tests/rpccaching_it_test.go

@@ -54,7 +54,7 @@ var sTestsRPCMethods = []func(t *testing.T){
 	testRPCMethodsTerminateSession,
 	testRPCMethodsProcessCDR,
 	testRPCMethodsProcessEvent,
-	//reset the storDB and dataDB
+	// reset the storDB and dataDB
 	testRPCMethodsInitDataDb,
 	testRPCMethodsResetStorDb,
 	testRPCMethodsCdrsProcessCDR,
@@ -693,6 +693,7 @@ func testRPCMethodsCdrsProcessCDR(t *testing.T) {
 		Event: map[string]interface{}{
 			utils.Tenant:      "cgrates.org",
 			utils.ToR:         utils.VOICE,
+			utils.OriginHost:  "host",
 			utils.OriginID:    "testRPCMethodsCdrsProcessCDR",
 			utils.RequestType: utils.META_PREPAID,
 			utils.Account:     "1001",
@@ -738,7 +739,7 @@ func testRPCMethodsCdrsProcessCDR(t *testing.T) {
 	//give time to CGRateS to delete the response from cache
 	time.Sleep(1*time.Second + 500*time.Millisecond)
 	//change originID so CGRID be different
-	args.Event[utils.OriginID] = "testRPCMethodsProcessCDR3"
+	args.Event[utils.OriginID] = "testRPCMethodsProcessCDR4"
 	if err := rpcRpc.Call(utils.CDRsV1ProcessEvent, args, &reply); err != nil {
 		t.Error("Unexpected error: ", err.Error())
 	} else if reply != utils.OK {