Add test for Flatstore in ERs

2026-02-11 18:16:24 +05:00 · 2020-01-22 11:06:29 +02:00
parent e7f95a98f3
commit 0d4e989832
10 changed files with 334 additions and 39 deletions
--- a/config/config.go
+++ b/config/config.go
@@ -305,7 +305,7 @@ var posibleLoaderTypes = utils.NewStringSet([]string{utils.MetaAttributes,

 var possibleReaderTypes = utils.NewStringSet([]string{utils.MetaFileCSV,
 	utils.MetaKafkajsonMap, utils.MetaFileXML, utils.MetaSQL, utils.MetaFileFWV,
-	utils.MetaPartialCSV})
+	utils.MetaPartialCSV, utils.MetaOsipsFlatstore, utils.MetaKamFlatstore})

 func (cfg *CGRConfig) LazySanityCheck() {
 	for _, cdrePrfl := range cfg.cdrsCfg.OnlineCDRExports {
--- a/config/configsanity.go
+++ b/config/configsanity.go
@@ -421,7 +421,7 @@ func (cfg *CGRConfig) checkConfigSanity() error {
 			}

 			switch rdr.Type {
-			case utils.MetaFileCSV, utils.MetaPartialCSV:
+			case utils.MetaFileCSV, utils.MetaPartialCSV, utils.MetaOsipsFlatstore, utils.MetaKamFlatstore:
 				for _, dir := range []string{rdr.ProcessedPath, rdr.SourcePath} {
 					if _, err := os.Stat(dir); err != nil && os.IsNotExist(err) {
 						return fmt.Errorf("<%s> Nonexistent folder: %s for reader with ID: %s", utils.ERs, dir, rdr.ID)
--- a/data/conf/samples/ers_internal/cgrates.json
+++ b/data/conf/samples/ers_internal/cgrates.json
@@ -330,10 +330,12 @@
 			"enabled": true,
 			"run_delay": -1,
 			"type": "*opensips_flatstore",
+			"field_separator": "|",
 			"source_path": "/tmp/flatstoreErs/in",
 			"processed_path": "/tmp/flatstoreErs/out",
+			"failed_calls_prefix": "missed_calls",
 			"flags": ["*cdrs"],
-			"partial_record_cache": "1s",
+			"partial_record_cache": "2s",
 			"content_fields":[
 				{"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true},
 				{"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true},
@@ -345,7 +347,7 @@
 				{"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true},
 				{"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
 				{"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
-				{"tag": "Usage", "field_id": "Usage", "type": "*none", "mandatory": true}, // Value for Usage is composed based on record
+				{"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record
 				{"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true},
 				{"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"}
 			],
--- a/data/conf/samples/ers_mongo/cgrates.json
+++ b/data/conf/samples/ers_mongo/cgrates.json
@@ -333,10 +333,12 @@
 			"enabled": true,
 			"run_delay": -1,
 			"type": "*opensips_flatstore",
+			"field_separator": "|",
 			"source_path": "/tmp/flatstoreErs/in",
 			"processed_path": "/tmp/flatstoreErs/out",
+			"failed_calls_prefix": "missed_calls",
 			"flags": ["*cdrs"],
-			"partial_record_cache": "1s",
+			"partial_record_cache": "2s",
 			"content_fields":[
 				{"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true},
 				{"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true},
@@ -348,7 +350,7 @@
 				{"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true},
 				{"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
 				{"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
-				{"tag": "Usage", "field_id": "Usage", "type": "*none", "mandatory": true}, // Value for Usage is composed based on record
+				{"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record
 				{"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true},
 				{"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"}
 			],
--- a/data/conf/samples/ers_mysql/cgrates.json
+++ b/data/conf/samples/ers_mysql/cgrates.json
@@ -324,6 +324,33 @@
 				{"tag": "Usage", "field_id": "Usage", "type": "*variable", "value": "~*req.6:s/^(\\d+)$/${1}s/", "mandatory": true},
 				{"tag": "Partial", "field_id": "Partial", "type": "*constant", "value": "true", "filters":["*string:~*req.10:partial"]}
 			],
+		},
+		{
+			"id": "FlatstoreOsips",
+			"enabled": true,
+			"run_delay": -1,
+			"type": "*opensips_flatstore",
+			"field_separator": "|",
+			"source_path": "/tmp/flatstoreErs/in",
+			"processed_path": "/tmp/flatstoreErs/out",
+			"failed_calls_prefix": "missed_calls",
+			"flags": ["*cdrs"],
+			"partial_record_cache": "2s",
+			"content_fields":[
+				{"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true},
+				{"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true},
+				{"tag": "RequestType", "field_id": "RequestType", "type": "*variable", "value": "~*req.7", "mandatory": true},
+				{"tag": "Tenant", "field_id": "Tenant", "type": "*constant", "value": "cgrates.org", "mandatory": true},
+				{"tag": "Category", "field_id": "Category", "type": "*constant", "value": "call", "mandatory": true},
+				{"tag": "Account", "field_id": "Account", "type": "*variable", "value": "~*req.8", "mandatory": true},
+				{"tag": "Subject", "field_id": "Subject", "type": "*variable", "value": "~*req.8", "mandatory": true},
+				{"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true},
+				{"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
+				{"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
+				{"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record
+				{"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true},
+				{"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"}
+			],
 		}
 	],
 },
--- a/data/conf/samples/ers_postgres/cgrates.json
+++ b/data/conf/samples/ers_postgres/cgrates.json
@@ -327,10 +327,12 @@
 			"enabled": true,
 			"run_delay": -1,
 			"type": "*opensips_flatstore",
+			"field_separator": "|",
 			"source_path": "/tmp/flatstoreErs/in",
 			"processed_path": "/tmp/flatstoreErs/out",
+			"failed_calls_prefix": "missed_calls",
 			"flags": ["*cdrs"],
-			"partial_record_cache": "1s",
+			"partial_record_cache": "2s",
 			"content_fields":[
 				{"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true},
 				{"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true},
@@ -342,7 +344,7 @@
 				{"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true},
 				{"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
 				{"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true},
-				{"tag": "Usage", "field_id": "Usage", "type": "*none", "mandatory": true}, // Value for Usage is composed based on record
+				{"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record
 				{"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true},
 				{"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"}
 			],
--- a/ers/flatstore.go
+++ b/ers/flatstore.go
@@ -47,15 +47,18 @@ func NewFlatstoreER(cfg *config.CGRConfig, cfgIdx int,
 	if strings.HasSuffix(srcPath, utils.Slash) {
 		srcPath = srcPath[:len(srcPath)-1]
 	}
-	return &FlatstoreER{
+	flatER := &FlatstoreER{
 		cgrCfg:    cfg,
 		cfgIdx:    cfgIdx,
 		fltrS:     fltrS,
-		cache:     ltcache.NewCache(ltcache.UnlimitedCaching, cfg.ERsCfg().Readers[cfgIdx].PartialRecordCache, false, nil),
 		rdrDir:    srcPath,
 		rdrEvents: rdrEvents,
 		rdrError:  rdrErr,
-		rdrExit:   rdrExit}, nil
+		rdrExit:   rdrExit,
+	}
+	flatER.cache = ltcache.NewCache(ltcache.UnlimitedCaching, cfg.ERsCfg().Readers[cfgIdx].PartialRecordCache, false, flatER.dumpToFile)
+	return flatER, err
+
 }

 // FlatstoreER implements EventReader interface for Flatstore CDR
@@ -147,27 +150,35 @@ func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) {
 			}
 			return
 		}
-		if strings.HasPrefix(file.Name(), rdr.Config().FailedCallsPrefix) { // Use the first index since they should be the same in all configs
-			record = append(record, "0") // Append duration 0 for failed calls flatstore CDR and do not process it further
-		}
-		pr, err := NewUnpairedRecord(record, rdr.Config().Timezone)
-		if err != nil {
-			fmt.Sprintf("<%s> Converting row : <%s> to unpairedRecord , ignoring due to error: <%s>",
-				utils.ERs, record, err.Error())
-			continue
-		}
-		if val, has := rdr.cache.Get(pr.OriginID); !has {
-			rdr.cache.Set(pr.OriginID, pr, nil)
-			continue
+		if strings.HasPrefix(fName, rdr.Config().FailedCallsPrefix) { // Use the first index since they should be the same in all configs
+			record = append(record, "0") // Append duration 0 for failed calls flatstore CDR
 		} else {
-			pair := val.(*UnpairedRecord)
-			record, err = pairToRecord(pair, pr)
+			pr, err := NewUnpairedRecord(record, rdr.Config().Timezone, fName)
 			if err != nil {
-				fmt.Sprintf("<%s> Merging unpairedRecords : <%s> and <%s> to record , ignoring due to error: <%s>",
-					utils.ERs, utils.ToJSON(pair), utils.ToJSON(pr), err.Error())
+				fmt.Sprintf("<%s> Converting row : <%s> to unpairedRecord , ignoring due to error: <%s>",
+					utils.ERs, record, err.Error())
 				continue
 			}
-			rdr.cache.Remove(pr.OriginID)
+			if val, has := rdr.cache.Get(pr.OriginID); !has {
+				rdr.cache.Set(pr.OriginID, pr, nil)
+				continue
+			} else {
+				pair := val.(*UnpairedRecord)
+				record, err = pairToRecord(pair, pr)
+				if err != nil {
+					fmt.Sprintf("<%s> Merging unpairedRecords : <%s> and <%s> to record , ignoring due to error: <%s>",
+						utils.ERs, utils.ToJSON(pair), utils.ToJSON(pr), err.Error())
+					continue
+				}
+				rdr.cache.Remove(pr.OriginID)
+			}
+		}
+
+		// build Usage from contentFields based on record lenght
+		for i, cntFld := range rdr.Config().ContentFields {
+			if cntFld.FieldId == utils.Usage {
+				rdr.Config().ContentFields[i].Value = config.NewRSRParsersMustCompile("~*req."+strconv.Itoa(len(record)-1), true, utils.INFIELD_SEP) // in case of flatstore, last element will be the duration computed by us
+			}
 		}
 		rowNr++ // increment the rowNr after checking if it's not the end of file
 		agReq := agents.NewAgentRequest(
@@ -181,12 +192,6 @@ func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) {
 			agReq); err != nil || !pass {
 			continue
 		}
-		// build Usage from contentFields based on record lenght
-		for i, cntFld := range rdr.Config().ContentFields {
-			if cntFld.FieldId == utils.Usage {
-				rdr.Config().ContentFields[i].Value = config.NewRSRParsersMustCompile("~*req."+strconv.Itoa(len(record)-1), true, utils.INFIELD_SEP) // in case of flatstore, last element will be the duration computed by us
-			}
-		}
 		navMp, err := agReq.AsNavigableMap(rdr.Config().ContentFields)
 		if err != nil {
 			utils.Logger.Warning(
@@ -214,11 +219,11 @@ func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) {
 	return
 }

-func NewUnpairedRecord(record []string, timezone string) (*UnpairedRecord, error) {
+func NewUnpairedRecord(record []string, timezone string, fileName string) (*UnpairedRecord, error) {
 	if len(record) < 7 {
 		return nil, errors.New("MISSING_IE")
 	}
-	pr := &UnpairedRecord{Method: record[0], OriginID: record[3] + record[1] + record[2], Values: record}
+	pr := &UnpairedRecord{Method: record[0], OriginID: record[3] + record[1] + record[2], Values: record, FileName: fileName}
 	var err error
 	if pr.Timestamp, err = utils.ParseTimeDetectLayout(record[6], timezone); err != nil {
 		return nil, err
@@ -232,6 +237,7 @@ type UnpairedRecord struct {
 	OriginID  string    // Copute here the OriginID
 	Timestamp time.Time // Timestamp of the event, as written by db_flastore module
 	Values    []string  // Can contain original values or updated via UpdateValues
+	FileName  string
 }

 // Pairs INVITE and BYE into final record containing as last element the duration
@@ -271,3 +277,24 @@ func pairToRecord(part1, part2 *UnpairedRecord) ([]string, error) {
 	record = append(record, strconv.FormatFloat(callDur.Seconds(), 'f', -1, 64))
 	return record, nil
 }
+
+func (rdr *FlatstoreER) dumpToFile(itmID string, value interface{}) {
+	unpRcd := value.(*UnpairedRecord)
+
+	dumpFilePath := path.Join(rdr.Config().ProcessedPath, unpRcd.FileName+utils.TmpSuffix)
+	fileOut, err := os.Create(dumpFilePath)
+	if err != nil {
+		utils.Logger.Err(fmt.Sprintf("<%s> Failed creating %s, error: %s",
+			utils.ERs, dumpFilePath, err.Error()))
+		return
+	}
+	csvWriter := csv.NewWriter(fileOut)
+	csvWriter.Comma = rune(rdr.Config().FieldSep[0])
+	if err = csvWriter.Write(unpRcd.Values); err != nil {
+		utils.Logger.Err(fmt.Sprintf("<%s> Failed writing partial record %v to file: %s, error: %s",
+			utils.ERs, unpRcd.Values, dumpFilePath, err.Error()))
+		return
+	}
+
+	csvWriter.Flush()
+}
--- a/ers/flatstore_it_test.go
+++ b/ers/flatstore_it_test.go
@@ -1 +1,235 @@
+// +build integration
+
+/*
+Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
+Copyright (C) ITsysCOM GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>
+*/
+
 package ers
+
+import (
+	"io/ioutil"
+	"net/rpc"
+	"os"
+	"path"
+	"testing"
+	"time"
+
+	v1 "github.com/cgrates/cgrates/apier/v1"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/engine"
+	"github.com/cgrates/cgrates/utils"
+)
+
+var (
+	flatstoreCfgPath string
+	flatstoreCfgDIR  string
+	flatstoreCfg     *config.CGRConfig
+	flatstoreRPC     *rpc.Client
+
+	fullSuccessfull = `INVITE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9e@0:0:0:0:0:0:0:0|200|OK|1436454408|*prepaid|1001|1002||3401:2069362475
+BYE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9e@0:0:0:0:0:0:0:0|200|OK|1436454410|||||3401:2069362475
+INVITE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4cca@0:0:0:0:0:0:0:0|200|OK|1436454647|*postpaid|1002|1001||1877:893549741
+BYE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4cca@0:0:0:0:0:0:0:0|200|OK|1436454651|||||1877:893549741
+INVITE|36e39a5|42d996f9|3a63321dd3b325eec688dc2aefb6ac2d@0:0:0:0:0:0:0:0|200|OK|1436454657|*prepaid|1001|1002||2407:1884881533
+BYE|36e39a5|42d996f9|3a63321dd3b325eec688dc2aefb6ac2d@0:0:0:0:0:0:0:0|200|OK|1436454661|||||2407:1884881533
+INVITE|3111f3c9|49ca4c42|a58ebaae40d08d6757d8424fb09c4c54@0:0:0:0:0:0:0:0|200|OK|1436454690|*prepaid|1001|1002||3099:1909036290
+BYE|3111f3c9|49ca4c42|a58ebaae40d08d6757d8424fb09c4c54@0:0:0:0:0:0:0:0|200|OK|1436454692|||||3099:1909036290`
+
+	fullMissed = `INVITE|ef6c6256|da501581|0bfdd176d1b93e7df3de5c6f4873ee04@0:0:0:0:0:0:0:0|487|Request Terminated|1436454643|*prepaid|1001|1002||1224:339382783
+INVITE|7905e511||81880da80a94bda81b425b09009e055c@0:0:0:0:0:0:0:0|404|Not Found|1436454668|*prepaid|1001|1002||1980:1216490844
+INVITE|324cb497|d4af7023|8deaadf2ae9a17809a391f05af31afb0@0:0:0:0:0:0:0:0|486|Busy here|1436454687|*postpaid|1002|1001||474:130115066`
+
+	part1 = `BYE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4ccb@0:0:0:0:0:0:0:0|200|OK|1436454651|||||1877:893549742`
+
+	part2 = `INVITE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4ccb@0:0:0:0:0:0:0:0|200|OK|1436454647|*postpaid|1002|1003||1877:893549742
+INVITE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9p@0:0:0:0:0:0:0:0|200|OK|1436454408|*prepaid|1001|1002||3401:2069362475`
+
+	flatstoreTests = []func(t *testing.T){
+		testFlatstoreITCreateCdrDirs,
+		testFlatstoreITInitConfig,
+		testFlatstoreITInitCdrDb,
+		testFlatstoreITResetDataDb,
+		testFlatstoreITStartEngine,
+		testFlatstoreITRpcConn,
+		testFlatstoreITLoadTPFromFolder,
+		testFlatstoreITHandleCdr1File,
+		testFlatstoreITAnalyseCDRs,
+		testFlatstoreITCleanupFiles,
+		testFlatstoreITKillEngine,
+	}
+)
+
+func TestFlatstoreFile(t *testing.T) {
+	switch *dbType {
+	case utils.MetaInternal:
+		flatstoreCfgDIR = "ers_internal"
+	case utils.MetaSQL:
+		flatstoreCfgDIR = "ers_mysql"
+	case utils.MetaMongo:
+		flatstoreCfgDIR = "ers_mongo"
+	case utils.MetaPostgres:
+		flatstoreCfgDIR = "ers_postgres"
+	default:
+		t.Fatal("Unknown Database type")
+	}
+	for _, test := range flatstoreTests {
+		t.Run(flatstoreCfgDIR, test)
+	}
+}
+
+func testFlatstoreITCreateCdrDirs(t *testing.T) {
+	for _, dir := range []string{"/tmp/ers/in", "/tmp/ers/out",
+		"/tmp/ers2/in", "/tmp/ers2/out", "/tmp/init_session/in", "/tmp/init_session/out",
+		"/tmp/terminate_session/in", "/tmp/terminate_session/out", "/tmp/cdrs/in",
+		"/tmp/cdrs/out", "/tmp/ers_with_filters/in", "/tmp/ers_with_filters/out",
+		"/tmp/xmlErs/in", "/tmp/xmlErs/out", "/tmp/fwvErs/in", "/tmp/fwvErs/out",
+		"/tmp/partErs1/in", "/tmp/partErs1/out", "/tmp/partErs2/in", "/tmp/partErs2/out",
+		"/tmp/flatstoreErs/in", "/tmp/flatstoreErs/out"} {
+		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 testFlatstoreITInitConfig(t *testing.T) {
+	var err error
+	flatstoreCfgPath = path.Join(*dataDir, "conf", "samples", flatstoreCfgDIR)
+	if flatstoreCfg, err = config.NewCGRConfigFromPath(flatstoreCfgPath); err != nil {
+		t.Fatal("Got config error: ", err.Error())
+	}
+}
+
+// InitDb so we can rely on count
+func testFlatstoreITInitCdrDb(t *testing.T) {
+	if err := engine.InitStorDb(flatstoreCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Remove data in both rating and accounting db
+func testFlatstoreITResetDataDb(t *testing.T) {
+	if err := engine.InitDataDb(flatstoreCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func testFlatstoreITStartEngine(t *testing.T) {
+	if _, err := engine.StopStartEngine(flatstoreCfgPath, *waitRater); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Connect rpc client to rater
+func testFlatstoreITRpcConn(t *testing.T) {
+	var err error
+	flatstoreRPC, err = newRPCClient(flatstoreCfg.ListenCfg()) // We connect over JSON so we can also troubleshoot if needed
+	if err != nil {
+		t.Fatal("Could not connect to rater: ", err.Error())
+	}
+}
+
+func testFlatstoreITLoadTPFromFolder(t *testing.T) {
+	//add a default charger
+	chargerProfile := &v1.ChargerWithCache{
+		ChargerProfile: &engine.ChargerProfile{
+			Tenant:       "cgrates.org",
+			ID:           "Default",
+			RunID:        utils.MetaDefault,
+			AttributeIDs: []string{"*none"},
+			Weight:       20,
+		},
+	}
+	var result string
+	if err := flatstoreRPC.Call(utils.ApierV1SetChargerProfile, chargerProfile, &result); err != nil {
+		t.Error(err)
+	} else if result != utils.OK {
+		t.Error("Unexpected reply returned", result)
+	}
+}
+
+// The default scenario, out of cdrc defined in .cfg file
+func testFlatstoreITHandleCdr1File(t *testing.T) {
+	if err := ioutil.WriteFile(path.Join("/tmp", "acc_1.log"), []byte(fullSuccessfull), 0644); err != nil {
+		t.Fatal(err.Error())
+	}
+	if err := ioutil.WriteFile(path.Join("/tmp", "missed_calls_1.log"), []byte(fullMissed), 0644); err != nil {
+		t.Fatal(err.Error())
+	}
+	if err := ioutil.WriteFile(path.Join("/tmp", "acc_2.log"), []byte(part1), 0644); err != nil {
+		t.Fatal(err.Error())
+	}
+	if err := ioutil.WriteFile(path.Join("/tmp", "acc_3.log"), []byte(part2), 0644); err != nil {
+		t.Fatal(err.Error())
+	}
+	//Rename(oldpath, newpath string)
+	for _, fileName := range []string{"acc_1.log", "missed_calls_1.log", "acc_2.log", "acc_3.log"} {
+		if err := os.Rename(path.Join("/tmp", fileName), path.Join("/tmp/flatstoreErs/in", fileName)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	time.Sleep(3 * time.Second)
+	// check the files to be processed
+	filesInDir, _ := ioutil.ReadDir("/tmp/flatstoreErs/in")
+	if len(filesInDir) != 0 {
+		t.Errorf("Files in cdrcInDir: %+v", filesInDir)
+	}
+	filesOutDir, _ := ioutil.ReadDir("/tmp/flatstoreErs/out")
+	if len(filesOutDir) != 6 {
+		t.Errorf("Unexpected number of files in output directory: %+v", len(filesOutDir))
+	}
+	ePartContent := "INVITE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9p@0:0:0:0:0:0:0:0|200|OK|1436454408|*prepaid|1001|1002||3401:2069362475\n"
+	if partContent, err := ioutil.ReadFile(path.Join("/tmp/flatstoreErs/out", "acc_3.log.tmp")); err != nil {
+		t.Error(err)
+	} else if (ePartContent) != (string(partContent)) {
+		t.Errorf("Expecting:\n%s\nReceived:\n%s", ePartContent, string(partContent))
+	}
+}
+
+func testFlatstoreITAnalyseCDRs(t *testing.T) {
+	var reply []*engine.ExternalCDR
+	if err := flatstoreRPC.Call(utils.ApierV2GetCDRs, utils.RPCCDRsFilter{}, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if len(reply) != 8 {
+		t.Error("Unexpected number of CDRs returned: ", len(reply))
+	}
+	if err := flatstoreRPC.Call(utils.ApierV2GetCDRs, utils.RPCCDRsFilter{MinUsage: "1"}, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if len(reply) != 5 {
+		t.Error("Unexpected number of CDRs returned: ", len(reply))
+	}
+}
+
+func testFlatstoreITCleanupFiles(t *testing.T) {
+	for _, dir := range []string{"/tmp/ers",
+		"/tmp/ers2", "/tmp/init_session", "/tmp/terminate_session",
+		"/tmp/cdrs", "/tmp/ers_with_filters", "/tmp/xmlErs", "/tmp/fwvErs",
+		"/tmp/partErs1", "/tmp/partErs2", "tmp/flatstoreErs"} {
+		if err := os.RemoveAll(dir); err != nil {
+			t.Fatal("Error removing folder: ", dir, err)
+		}
+	}
+}
+
+func testFlatstoreITKillEngine(t *testing.T) {
+	if err := engine.KillEngine(*waitRater); err != nil {
+		t.Error(err)
+	}
+}
--- a/ers/partial_csv.go
+++ b/ers/partial_csv.go
@@ -284,8 +284,8 @@ func (rdr *PartialCSVFileER) dumpToFile(itmID string, value interface{}) {
 				utils.ERs, cdr.CGRID, err.Error()))
 		return
 	}
-	dumpFilePath := path.Join(rdr.Config().ProcessedPath, fmt.Sprintf("%s.%s.%d",
-		cdr.OriginID, PartialRecordsSuffix, time.Now().Unix()))
+	dumpFilePath := path.Join(rdr.Config().ProcessedPath, fmt.Sprintf("%s%s.%d",
+		cdr.OriginID, utils.TmpSuffix, time.Now().Unix()))
 	fileOut, err := os.Create(dumpFilePath)
 	if err != nil {
 		utils.Logger.Err(fmt.Sprintf("<%s> Failed creating %s, error: %s",
@@ -293,7 +293,7 @@ func (rdr *PartialCSVFileER) dumpToFile(itmID string, value interface{}) {
 		return
 	}
 	csvWriter := csv.NewWriter(fileOut)
-	csvWriter.Comma = utils.CSV_SEP
+	csvWriter.Comma = rune(rdr.Config().FieldSep[0])
 	if err = csvWriter.Write(record); err != nil {
 		utils.Logger.Err(fmt.Sprintf("<%s> Failed writing partial record %v to file: %s, error: %s",
 			utils.ERs, record, dumpFilePath, err.Error()))
--- a/utils/consts.go
+++ b/utils/consts.go
@@ -619,6 +619,7 @@ const (
 	MinCost                     = "MinCost"
 	MaxCost                     = "MaxCost"
 	MetaLoaders                 = "*loaders"
+	TmpSuffix                   = ".tmp"
 )

 // Migrator Action