Adding local tests for Load*Filtered methods in datadb

apier/v1/apier.go

@@ -273,20 +273,15 @@ func (self *ApierV1) AddAccount(attr AttrAddAccount, reply *string) error {
 	return nil
 }
 
-type AttrSetAccountActions struct {
-	TPid             string
-	AccountActionsId string
-}
-
 // Process dependencies and load a specific AccountActions profile from storDb into dataDb.
-func (self *ApierV1) SetAccountActions(attrs AttrSetAccountActions, reply *string) error {
-	if missing := utils.MissingStructFields(&attrs, []string{"TPid", "AccountActionsId"}); len(missing) != 0 {
+func (self *ApierV1) SetAccountActions(attrs utils.TPAccountActions, reply *string) error {
+	if missing := utils.MissingStructFields(&attrs, []string{"TPid", "LoadId", "Tenant", "Account", "Direction"}); len(missing) != 0 {
 		return fmt.Errorf("%s:%v", utils.ERR_MANDATORY_IE_MISSING, missing)
 	}
 	dbReader := engine.NewDbReader(self.StorDb, self.DataDb, attrs.TPid)
 
-	if _, err := engine.AccLock.Guard(attrs.AccountActionsId, func() (float64, error) {
-		if err := dbReader.LoadAccountActionsByTag(attrs.AccountActionsId); err != nil {
+	if _, err := engine.AccLock.Guard(attrs.KeyId(), func() (float64, error) {
+		if err := dbReader.LoadAccountActionsFiltered(&attrs); err != nil {
 			return 0, err
 		}
 		return 0, nil

console/set_accountactions.go

@@ -20,7 +20,7 @@ package console
 
 import (
 	"fmt"
-	"github.com/cgrates/cgrates/apier/v1"
+	"github.com/cgrates/cgrates/utils"
 )
 
 func init() {
@@ -30,19 +30,19 @@ func init() {
 // Commander implementation
 type CmdSetAccountActions struct {
 	rpcMethod string
-	rpcParams *apier.AttrSetAccountActions
+	rpcParams *utils.TPAccountActions
 	rpcResult string
 }
 
 // name should be exec's name
 func (self *CmdSetAccountActions) Usage(name string) string {
-	return fmt.Sprintf("\n\tUsage: cgr-console [cfg_opts...{-h}] set_accountactions <tpid> <accountactionsid>")
+	return fmt.Sprintf("\n\tUsage: cgr-console [cfg_opts...{-h}] set_accountactions <tpid> <loadid> <tenant> <account>")
 }
 
 // set param defaults
 func (self *CmdSetAccountActions) defaults() error {
 	self.rpcMethod = "ApierV1.SetAccountActions"
-	self.rpcParams = &apier.AttrSetAccountActions{}
+	self.rpcParams = new(utils.TPAccountActions)
 	return nil
 }
 
@@ -53,8 +53,7 @@ func (self *CmdSetAccountActions) FromArgs(args []string) error {
 	}
 	// Args look OK, set defaults before going further
 	self.defaults()
-	self.rpcParams.TPid = args[2]
-	self.rpcParams.AccountActionsId = args[3]
+	self.rpcParams = &utils.TPAccountActions{TPid: args[2], LoadId: args[3], Tenant: args[4], Account: args[5], Direction:"*out"}
 	return nil
 }
 

data/tariffplans/prepaid1centpsec/DestinationRates.csv

@@ -1,2 +1,3 @@
 #Tag,DestinationsTag,RatesTag
-DR_FREESWITCH_USERS,FS_USERS,RT_FS_USERS
+DR_RETAIL,GERMANY,RT_1CENT
+DR_RETAIL,GERMANY_MOBILE,RT_1CENT

data/tariffplans/prepaid1centpsec/Destinations.csv

@@ -3,4 +3,3 @@ GERMANY,+49
 GERMANY_MOBILE,+4915
 GERMANY_MOBILE,+4916
 GERMANY_MOBILE,+4917
-FS_USERS,10

data/tariffplans/prepaid1centpsec/Rates.csv

@@ -1,2 +1,2 @@
 #Tag,ConnectFee,Rate,RateUnit,RateIncrement,GroupIntervalStart,RoundingMethod,RoundingDecimals
-RT_FS_USERS,0,0,60s,60s,0s,*up,0
+RT_1CENT,0,1,60s,60s,0s,*up,2

data/tariffplans/prepaid1centpsec/RatingPlans.csv

@@ -1,2 +1,2 @@
 #Tag,DestinationRatesTag,TimingTag,Weight
-RETAIL1,DR_FREESWITCH_USERS,ALWAYS,10
+RP_RETAIL,DR_RETAIL,ALWAYS,10

data/tariffplans/prepaid1centpsec/RatingProfiles.csv

@@ -1,2 +1,2 @@
 #Tenant,TOR,Direction,Subject,ActivationTime,RatingPlanTag,FallbackSubject
-cgrates.org,call,*out,*any,2012-01-01T00:00:00Z,RETAIL1,
+cgrates.org,call,*out,*any,2012-01-01T00:00:00Z,RP_RETAIL,

engine/loader_csv.go

@@ -530,7 +530,6 @@ func (csvr *CSVReader) LoadAccountActions() (err error) {
 			ActionTriggers: aTriggers,
 		}
 		csvr.accountActions = append(csvr.accountActions, ub)
-
 		aTimings, exists := csvr.actionsTimings[record[3]]
 		if !exists {
 			log.Printf("Could not get action timing for tag %v", record[3])

engine/loader_db.go

@@ -297,8 +297,10 @@ func (dbr *DbReader) LoadRatingPlanByTag(tag string) error {
 func (dbr *DbReader) LoadRatingProfileFiltered(qriedRpf *utils.TPRatingProfile) error {
 	var resultRatingProfile *RatingProfile
 	mpTpRpfs, err := dbr.storDb.GetTpRatingProfiles(qriedRpf) //map[string]*utils.TPRatingProfile
-	if err != nil || len(mpTpRpfs) == 0 {
+	if err != nil {
 		return fmt.Errorf("No RateProfile for filter %v, error: %s", qriedRpf, err.Error())
+	} else if len(mpTpRpfs) == 0 {
+		return fmt.Errorf("No RateProfile for filter %v", qriedRpf)
 	} else if len(mpTpRpfs) > 1 {
 		return fmt.Errorf("More than one rating profile returned in LoadRatingProfileFiltered") // Should never reach here
 	}
@@ -418,16 +420,16 @@ func (dbr *DbReader) LoadAccountActions() (err error) {
 	return nil
 }
 
-func (dbr *DbReader) LoadAccountActionsByTag(tag string) error {
-	accountActions, err := dbr.storDb.GetTpAccountActions(&utils.TPAccountActions{TPid: dbr.tpid, LoadId: tag})
+func (dbr *DbReader) LoadAccountActionsFiltered(qriedAA *utils.TPAccountActions) error {
+	accountActions, err := dbr.storDb.GetTpAccountActions(qriedAA)
 	if err != nil {
 		return err
 	} else if len(accountActions) == 0 {
-		return fmt.Errorf("No AccountActions with id <%s>", tag)
+		return fmt.Errorf("No AccountActions for queried %v", qriedAA)
 	} else if len(accountActions) > 1 {
-		return fmt.Errorf("StorDb configuration error for AccountActions <%s>", tag)
+		return fmt.Errorf("StorDb configuration error for AccountActions <%v>", qriedAA)
 	}
-	accountAction := accountActions[tag]
+	accountAction := accountActions[qriedAA.KeyId()]
 	id := fmt.Sprintf("%s:%s:%s", accountAction.Direction, accountAction.Tenant, accountAction.Account)
 
 	var actionsIds []string // collects action ids
@@ -458,13 +460,13 @@ func (dbr *DbReader) LoadAccountActionsByTag(tag string) error {
 			} else if len(actions) == 0 {
 				return fmt.Errorf("No Action with id <%s>", at.ActionsId)
 			}
-			timingsMap, err := dbr.storDb.GetTpTimings(dbr.tpid, accountAction.ActionTimingsId)
+			timingsMap, err := dbr.storDb.GetTpTimings(dbr.tpid, at.TimingId)
 			if err != nil {
 				return err
 			} else if len(timingsMap) == 0 {
-				return fmt.Errorf("No Timing with id <%s>", accountAction.ActionTimingsId)
+				return fmt.Errorf("No Timing with id <%s>", at.TimingId)
 			}
-			t := timingsMap[accountAction.ActionTimingsId]
+			t := timingsMap[at.TimingId]
 			actTmg := &ActionTiming{
 				Id:     utils.GenUUID(),
 				Tag:    accountAction.ActionTimingsId,

engine/loader_local_test.go

@@ -40,7 +40,7 @@ README:
 */
 
 // Globals used
-var dataDbCsv, dataDbStor DataStorage // Test data coming from csv files getting inthere
+var dataDbCsv, dataDbStor, dataDbApier DataStorage // Each dataDb will have it's own sources to collect data
 var storDb LoadStorage
 var cfg *config.CGRConfig
 
@@ -63,6 +63,14 @@ func TestConnDataDbs(t *testing.T) {
 	if dataDbStor, err = ConfigureDataStorage(cfg.DataDBType, cfg.DataDBHost, cfg.DataDBPort, "14", cfg.DataDBUser, cfg.DataDBPass, cfg.DBDataEncoding); err != nil {
 		t.Fatal("Error on dataDb connection: ", err.Error())
 	}
+	if dataDbApier, err = ConfigureDataStorage(cfg.DataDBType, cfg.DataDBHost, cfg.DataDBPort, "15", cfg.DataDBUser, cfg.DataDBPass, cfg.DBDataEncoding); err != nil {
+		t.Fatal("Error on dataDb connection: ", err.Error())
+	}
+	for _,db := range []DataStorage{dataDbCsv, dataDbStor, dataDbApier} {
+		if err = db.Flush(); err != nil {
+			t.Fatal("Error when flushing datadb")
+		}
+	}
 }
 
 // Create/reset storage tariff plan tables, used as database connectin establishment also
@@ -201,6 +209,51 @@ func TestLoadFromStorDb(t *testing.T) {
 	}
 }
 
+func TestLoadIndividualProfiles(t *testing.T) {
+	if !*testLocal {
+		return
+	}
+	loader := NewDbReader(storDb, dataDbApier, TEST_SQL)
+	// Load ratingPlans. This will also set destination keys
+	if ratingPlans, err := storDb.GetTpRatingPlans(TEST_SQL, ""); err != nil {
+		t.Fatal("Could not retrieve rating plans")
+	} else {
+		for tag := range ratingPlans {
+			if err := loader.LoadRatingPlanByTag(tag); err != nil {
+				t.Fatalf("Could not load ratingPlan for tag: %s, error: %s", tag, err.Error())
+			}
+		}
+	}
+	// Load rating profiles
+	loadId := utils.CSV_LOAD+"_"+TEST_SQL
+	if ratingProfiles, err := storDb.GetTpRatingProfiles(&utils.TPRatingProfile{TPid:TEST_SQL, LoadId: loadId}); err != nil {
+		t.Fatal("Could not retrieve rating profiles, error: ", err.Error())
+	} else if len(ratingProfiles) == 0 {
+		t.Fatal("Could not retrieve rating profiles")
+	} else {
+		for rpId := range ratingProfiles {
+			rp, _ := utils.NewTPRatingProfileFromKeyId(TEST_SQL, loadId, rpId)
+			if err := loader.LoadRatingProfileFiltered(rp); err != nil {
+				t.Fatalf("Could not load ratingProfile with id: %s, error: %s", rpId, err.Error())
+			}
+		}
+	}
+	// Load account actions
+	if aas, err := storDb.GetTpAccountActions(&utils.TPAccountActions{TPid:TEST_SQL, LoadId: loadId}); err != nil {
+		t.Fatal("Could not retrieve account action profiles, error: ", err.Error())
+	} else if len(aas) == 0 {
+		t.Error("No account actions")
+	} else {
+		for aaId := range aas {
+			aa, _ := utils.NewTPAccountActionsFromKeyId(TEST_SQL, loadId, aaId)
+			if err := loader.LoadAccountActionsFiltered(aa); err != nil {
+				t.Fatalf("Could not load account actions with id: %s, error: %s", aaId, err.Error())
+			}
+		}
+	}
+}
+	
+
 // Compares previously loaded data from csv and stor to be identical, redis specific tests
 func TestMatchLoadCsvWithStor(t *testing.T) {
 	if !*testLocal {
@@ -210,24 +263,26 @@ func TestMatchLoadCsvWithStor(t *testing.T) {
 	if !redisDb {
 		return // We only support these tests for redis
 	}
-	rsStor := dataDbCsv.(*RedisStorage)
+	rsStor := dataDbStor.(*RedisStorage)
+	rsApier := dataDbApier.(*RedisStorage)
 	keysCsv, err := rsCsv.db.Keys("*")
 	if err != nil {
 		t.Fatal("Failed querying redis keys for csv data")
 	}
 	for _, key := range keysCsv {
-		valCsv, err := rsCsv.db.Get(key)
-		if err != nil {
-			t.Errorf("Error when querying dataDbCsv for key: %s - %s ", key, err.Error())
-			continue
-		}
-		valStor, err := rsStor.db.Get(key)
-		if err != nil {
-			t.Errorf("Error when querying dataDbStor for key: %s - %s", key, err.Error())
-			continue
-		}
-		if valCsv != valStor {
-			t.Errorf("Missmatched data for key: %s\n\t, dataDbCsv: %s \n\t dataDbStor: %s\n", key, valCsv, valStor)
+		refVal := ""
+		for idx, rs := range []*RedisStorage{rsCsv, rsStor, rsApier} {
+			qVal, err := rs.db.Get(key)
+			if err != nil {
+				t.Fatal("Could not retrieve key %s, error: %s", key, err.Error())
+			}
+			if idx == 0 { // Only compare at second iteration, first one is to set reference value
+				refVal = qVal
+				continue
+			}
+			if len(refVal) != len(qVal) {
+				t.Errorf("Missmatched data for key: %s\n\t, reference val: %s \n\t retrieved value: %s\n on iteration: %d", key, refVal, qVal, idx)
+			}
 		}
 	}
 }

engine/tpimporter_csv.go

@@ -264,7 +264,7 @@ func (self *TPCSVImporter) importRatingProfiles(fn string) error {
 			}
 			continue
 		}
-		loadId := "TPCSV" //Autogenerate rating profile id
+		loadId := utils.CSV_LOAD //Autogenerate rating profile id
 		if self.ImportId != "" {
 			loadId += "_" + self.ImportId
 		}
@@ -461,7 +461,7 @@ func (self *TPCSVImporter) importAccountActions(fn string) error {
 			continue
 		}
 		tenant, account, direction, actionTimingsTag, actionTriggersTag := record[0], record[1], record[2], record[3], record[4]
-		loadId := "TPCSV" //Autogenerate account actions profile id
+		loadId := utils.CSV_LOAD //Autogenerate account actions profile id
 		if self.ImportId != "" {
 			loadId += "_" + self.ImportId
 		}

utils/apitpdata.go

@@ -138,6 +138,18 @@ func (self *TPRatingPlanBinding) Timing() *TPTiming {
 	return self.timing
 }
 
+
+// Used to rebuild a TPRatingProfile (empty RatingPlanActivations) out of it's key in nosqldb
+func NewTPRatingProfileFromKeyId( tpid, loadId, keyId string ) (*TPRatingProfile, error) {
+	// *out:cgrates.org:call:*any
+	s := strings.Split(keyId, ":")
+	// [*out cgrates.org call *any]
+	if len(s) != 4 {
+		return nil, fmt.Errorf("Cannot parse key %s into RatingProfile", keyId)
+	}
+	return &TPRatingProfile{TPid: tpid, LoadId: loadId, Tenant:s[1], TOR:s[2], Direction:s[0], Subject:s[3]}, nil
+}
+
 type TPRatingProfile struct {
 	TPid                  string                // Tariff plan id
 	LoadId                string                // Gives ability to load specific RatingProfile based on load identifier, hence being able to keep history also in stordb
@@ -230,6 +242,17 @@ type TPActionTrigger struct {
 	Weight         float64 // weight
 }
 
+// Used to rebuild a TPAccountActions (empty ActionTimingsId and ActionTriggersId) out of it's key in nosqldb
+func NewTPAccountActionsFromKeyId( tpid, loadId, keyId string ) (*TPAccountActions, error) {
+	// *out:cgrates.org:1001
+	s := strings.Split(keyId, ":")
+	// [*out cgrates.org 1001]
+	if len(s) != 3 {
+		return nil, fmt.Errorf("Cannot parse key %s into AccountActions", keyId)
+	}
+	return &TPAccountActions{TPid: tpid, LoadId: loadId, Tenant:s[1], Account: s[2], Direction:s[0]}, nil
+}
+
 type TPAccountActions struct {
 	TPid             string // Tariff plan id
 	LoadId           string // LoadId, used to group actions on a load

utils/consts.go

@@ -61,4 +61,6 @@ const (
 	FALLBACK_SEP             = ';'
 	JSON                     = "json"
 	MSGPACK                  = "msgpack"
+	CSV_LOAD                 = "CSVLOAD"
+
 )