Allow filters from cdrc to take value from records

cdrc/csv_it_test.go

@@ -430,3 +430,126 @@ func TestCsvIT4KillEngine(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+// Begin tests for cdrc csv with new filters
+var fileContent1_5 = `accid21;*prepaid;itsyscom.com;1002;086517174963;2013-02-03 19:54:00;62;val_extra3;"";val_extra1;10.10.10.10
+accid22;*postpaid;itsyscom.com;1001;+4986517174963;2013-02-03 19:54:00;123;val_extra3;"";val_extra1;11.10.10.10
+accid23;*postpaid;cgrates.org;1002;086517174963;2013-02-03 19:54:00;76;val_extra3;"";val_extra1;12.10.10.10
+accid24;*postpaid;cgrates.org;1001;+4986517174963;2013-02-03 19:54:00;76;val_extra3;"";val_extra1;12.10.10.10`
+
+func TestCsvIT5InitConfig(t *testing.T) {
+	var err error
+	csvCfgPath = path.Join(*dataDir, "conf", "samples", "cdrccsv")
+	if csvCfg, err = config.NewCGRConfigFromFolder(csvCfgPath); err != nil {
+		t.Fatal("Got config error: ", err.Error())
+	}
+}
+
+// InitDb so we can rely on count
+func TestCsvIT5InitCdrDb(t *testing.T) {
+	if err := engine.InitStorDb(csvCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCsvIT5CreateCdrDirs(t *testing.T) {
+	for _, cdrcProfiles := range csvCfg.CdrcProfiles {
+		for _, cdrcInst := range cdrcProfiles {
+			for _, dir := range []string{cdrcInst.CdrInDir, cdrcInst.CdrOutDir} {
+				if err := os.RemoveAll(dir); err != nil {
+					t.Fatal("Error removing folder: ", dir, err)
+				}
+				if err := os.MkdirAll(dir, 0755); err != nil {
+					t.Fatal("Error creating folder: ", dir, err)
+				}
+			}
+		}
+	}
+}
+
+func TestCsvIT5StartEngine(t *testing.T) {
+	if _, err := engine.StopStartEngine(csvCfgPath, *waitRater); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Connect rpc client to rater
+func TestCsvIT5RpcConn(t *testing.T) {
+	var err error
+	cdrcRpc, err = jsonrpc.Dial("tcp", csvCfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
+	if err != nil {
+		t.Fatal("Could not connect to rater: ", err.Error())
+	}
+}
+
+func TestCsvIT5AddFilters(t *testing.T) {
+	filter := &engine.Filter{
+		Tenant: "cgrates.org",
+		ID:     "FLTR_CDRC_ACC",
+		Rules: []*engine.FilterRule{
+			&engine.FilterRule{
+				Type:      "*string",
+				FieldName: "3",
+				Values:    []string{"1002"},
+			},
+		},
+	}
+	var result string
+	if err := cdrcRpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
+		t.Error(err)
+	} else if result != utils.OK {
+		t.Error("Unexpected reply returned", result)
+	}
+	filter2 := &engine.Filter{
+		Tenant: "itsyscom.com",
+		ID:     "FLTR_CDRC_ACC",
+		Rules: []*engine.FilterRule{
+			&engine.FilterRule{
+				Type:      "*string",
+				FieldName: "3",
+				Values:    []string{"1001"},
+			},
+		},
+	}
+	if err := cdrcRpc.Call("ApierV1.SetFilter", filter2, &result); err != nil {
+		t.Error(err)
+	} else if result != utils.OK {
+		t.Error("Unexpected reply returned", result)
+	}
+}
+
+// Scenario out of first .xml config
+func TestCsvIT5HandleCdr2File(t *testing.T) {
+	fileName := "file1.csv"
+	tmpFilePath := path.Join("/tmp", fileName)
+	if err := ioutil.WriteFile(tmpFilePath, []byte(fileContent1_5), 0644); err != nil {
+		t.Fatal(err.Error())
+	}
+	if err := os.Rename(tmpFilePath, path.Join("/tmp/csvwithfilter4/csvit4/in", fileName)); err != nil {
+		t.Fatal("Error moving file to processing directory: ", err)
+	}
+}
+
+func TestCsvIT5ProcessedFiles(t *testing.T) {
+	time.Sleep(time.Duration(2**waitRater) * time.Millisecond)
+	if outContent4, err := ioutil.ReadFile("/tmp/csvwithfilter4/csvit4/out/file1.csv"); err != nil {
+		t.Error(err)
+	} else if fileContent1_5 != string(outContent4) {
+		t.Errorf("Expecting: %q, received: %q", fileContent1_5, string(outContent4))
+	}
+}
+
+func TestCsvIT5AnalyseCDRs(t *testing.T) {
+	var reply []*engine.ExternalCDR
+	if err := cdrcRpc.Call("ApierV2.GetCdrs", utils.RPCCDRsFilter{}, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if len(reply) != 2 {
+		t.Error("Unexpected number of CDRs returned: ", len(reply))
+	}
+}
+
+func TestCsvIT5KillEngine(t *testing.T) {
+	if err := engine.KillEngine(*waitRater); err != nil {
+		t.Error(err)
+	}
+}

cdrc/fwv.go

@@ -115,12 +115,12 @@ func (self *FwvRecordsProcessor) ProcessNextRecord() ([]*engine.CDR, error) {
 	self.processedRecordsNr += 1
 	record := string(buf)
 	for _, cdrcCfg := range self.cdrcCfgs {
+		fwvProvider := newfwvProvider(record)
+		tenant, err := cdrcCfg.Tenant.ParseDataProvider(fwvProvider, utils.NestingSep) // each profile of cdrc can have different tenant
+		if err != nil {
+			return nil, err
+		}
 		if len(cdrcCfg.Filters) != 0 {
-			fwvProvider := newfwvProvider(record)
-			tenant, err := cdrcCfg.Tenant.ParseValue("")
-			if err != nil {
-				return nil, err
-			}
 			if pass, err := self.filterS.Pass(tenant,
 				cdrcCfg.Filters, fwvProvider); err != nil || !pass {
 				continue // Not passes filters, ignore this CDR

data/conf/samples/cdrccsv/cgrates.json

@@ -5,7 +5,9 @@
 //
 // This file contains the default configuration hardcoded into CGRateS.
 // This is what you get when you load CGRateS with an empty configuration file.
-
+"general": {
+	"log_level": 7,
+},
 
 "stor_db": {								// database used to store offline tariff plans and CDRs
 	"db_password": "CGRateS.org",			// password to use when connecting to stordb
@@ -135,6 +137,35 @@
 			{"tag": "RandomVal", "field_id": "RandomVal", "type": "*composed", "value": "*randomValue","filters":["*string:3:1001"]},
 		],
 	},
+	{
+ 		"id": "DifferentTenant",								// identifier of the CDRC runner
+ 		"enabled": true,							// enable CDR client functionality		
+ 		"field_separator": ";",					
+ 		"cdr_in_dir": "/tmp/csvwithfilter4/csvit4/in",	// absolute path towards the directory where the CDRs are stored
+ 		"cdr_out_dir": "/tmp/csvwithfilter4/csvit4/out",	// absolute path towards the directory where processed CDRs will be moved
+ 		"tenant": "~2",						// tenant used in filterS.Pass
+ 		"filters":["FLTR_CDRC_ACC"],		// if tenant is itsyscom.com will check if Account is 1001 and if tenant is cgrates.org will check if Account is 1002
+ 		"cdr_source_id": "diffTenant",			// free form field, tag identifying the source of the CDRs within CDRS database
+		"content_fields":[							// import content_fields template, tag will match internally CDR field, in case of .csv value will be represented by index of the field value
+			{"tag": "TOR", "field_id": "ToR", "type": "*composed", "value": "*voice", "mandatory": true},
+			{"tag": "OriginID", "field_id": "OriginID", "type": "*composed", "value": "~0", "mandatory": true},
+			{"tag": "OriginHost", "field_id": "OriginHost", "type": "*composed", "value": "~10", "mandatory": true},
+			{"tag": "RequestType", "field_id": "RequestType", "type": "*composed", "value": "~1", "mandatory": true},
+			{"tag": "Direction", "field_id": "Direction", "type": "*composed", "value": "*out", "mandatory": true},
+			{"tag": "Tenant", "field_id": "Tenant", "type": "*composed", "value": "~2", "mandatory": true},
+			{"tag": "Category", "field_id": "Category", "type": "*composed", "value": "call", "mandatory": true},
+			{"tag": "Account", "field_id": "Account", "type": "*composed", "value": "~3", "mandatory": true},
+			{"tag": "Subject", "field_id": "Subject", "type": "*composed", "value": "~3", "mandatory": true},
+			{"tag": "Destination", "field_id": "Destination", "type": "*composed", "value": "~4:s/0([1-9]\\d+)/+49${1}/", "mandatory": true},
+			{"tag": "SetupTime", "field_id": "SetupTime", "type": "*composed", "value": "~5", "mandatory": true},
+			{"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*composed", "value": "~5", "mandatory": true},
+			{"tag": "Usage", "field_id": "Usage", "type": "*composed", "value": "~6", "mandatory": true},
+			{"tag": "HDRExtra3", "field_id": "HDRExtra3", "type": "*composed", "value": "~6", "mandatory": true},
+			{"tag": "HDRExtra2", "field_id": "HDRExtra2", "type": "*composed", "value": "~6", "mandatory": true},
+			{"tag": "HDRExtra1", "field_id": "HDRExtra1", "type": "*composed", "value": "~6", "mandatory": true},
+			{"tag": "RandomVal", "field_id": "RandomVal", "type": "*composed", "value": "*randomValue","filters":["*string:3:1001"]},
+		],
+	},
 ],