Add test for *json type for ERs

2026-02-11 18:16:24 +05:00 · 2020-03-31 17:18:04 +03:00
parent ed66641f18
commit eaa3469aa2
16 changed files with 654 additions and 152 deletions
--- a/apier/v2/cdrs_it_test.go
+++ b/apier/v2/cdrs_it_test.go
@@ -66,6 +66,10 @@ var (

 		testv2CDRsGetCDRsDest,

+		testV2CDRsInitDataDb,
+		testV2CDRsInitCdrDb,
+		testV2CDRsRerate,
+
 		testV2CDRsKillEngine,
 	}
 )
@@ -706,6 +710,7 @@ func testV2CDRsGetCdrsWithRattingPlan(t *testing.T) {
 		}
 	}
 }
+
 func testV2CDRsSetThreshold(t *testing.T) {
 	var reply string
 	if err := cdrsRpc.Call(utils.APIerSv2SetActions, &utils.AttrSetActions{
@@ -836,6 +841,122 @@ func testv2CDRsGetCDRsDest(t *testing.T) {
 	}
 }

+func testV2CDRsRerate(t *testing.T) {
+	var reply string
+	//add a charger
+	chargerProfile := &v1.ChargerWithCache{
+		ChargerProfile: &engine.ChargerProfile{
+			Tenant: "cgrates.org",
+			ID:     "Default",
+			ActivationInterval: &utils.ActivationInterval{
+				ActivationTime: time.Date(2014, 7, 14, 14, 35, 0, 0, time.UTC),
+			},
+			RunID:        utils.MetaDefault,
+			AttributeIDs: []string{"*none"},
+			Weight:       20,
+		},
+		Cache: utils.StringPointer(utils.MetaReload),
+	}
+	if err := cdrsRpc.Call(utils.APIerSv1SetChargerProfile, chargerProfile, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply returned", reply)
+	}
+
+	attrSetAcnt := AttrSetAccount{
+		Tenant:  "cgrates.org",
+		Account: "voiceAccount",
+	}
+	if err := cdrsRpc.Call(utils.APIerSv2SetAccount, attrSetAcnt, &reply); err != nil {
+		t.Fatal(err)
+	}
+	attrs := &utils.AttrSetBalance{
+		Tenant:      "cgrates.org",
+		Account:     "voiceAccount",
+		BalanceType: utils.VOICE,
+		Value:       600000000000,
+		Balance: map[string]interface{}{
+			utils.ID:        utils.MetaDefault,
+			"RatingSubject": "*zero1m",
+			utils.Weight:    10.0,
+		},
+	}
+	if err := cdrsRpc.Call(utils.APIerSv2SetBalance, attrs, &reply); err != nil {
+		t.Fatal(err)
+	}
+
+	var acnt *engine.Account
+	if err := cdrsRpc.Call(utils.APIerSv2GetAccount,
+		&utils.AttrGetAccount{Tenant: "cgrates.org", Account: "voiceAccount"}, &acnt); err != nil {
+		t.Error(err)
+	} else if len(acnt.BalanceMap) != 1 || acnt.BalanceMap[utils.VOICE][0].Value != 600000000000 {
+		t.Errorf("Unexpected balance received: %+v", acnt.BalanceMap[utils.VOICE][0])
+	}
+
+	args := &engine.ArgV1ProcessEvent{
+		Flags: []string{utils.MetaRerate},
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+			Event: map[string]interface{}{
+				utils.OriginID:    "testV2CDRsRerate",
+				utils.OriginHost:  "192.168.1.1",
+				utils.Source:      "testV2CDRsRerate",
+				utils.RequestType: utils.META_PSEUDOPREPAID,
+				utils.Account:     "voiceAccount",
+				utils.Subject:     "ANY2CNT",
+				utils.Destination: "+4986517174963",
+				utils.AnswerTime:  time.Date(2018, 8, 24, 16, 00, 26, 0, time.UTC),
+				utils.Usage:       2 * time.Minute,
+				"field_extr1":     "val_extr1",
+				"fieldextr2":      "valextr2",
+			},
+		},
+	}
+
+	var rplProcEv []*utils.EventWithFlags
+	if err := cdrsRpc.Call(utils.CDRsV2ProcessEvent, args, &rplProcEv); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	}
+
+	if err := cdrsRpc.Call(utils.APIerSv2GetAccount,
+		&utils.AttrGetAccount{Tenant: "cgrates.org", Account: "voiceAccount"}, &acnt); err != nil {
+		t.Error(err)
+	} else if len(acnt.BalanceMap) != 1 || acnt.BalanceMap[utils.VOICE][0].Value != 480000000000 {
+		t.Errorf("Unexpected balance received: %+v", acnt.BalanceMap[utils.VOICE][0])
+	}
+
+	args2 := &engine.ArgV1ProcessEvent{
+		Flags: []string{utils.MetaRerate},
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+			Event: map[string]interface{}{
+				utils.OriginID:    "testV2CDRsRerate",
+				utils.OriginHost:  "192.168.1.1",
+				utils.Source:      "testV2CDRsRerate",
+				utils.RequestType: utils.META_PSEUDOPREPAID,
+				utils.Account:     "voiceAccount",
+				utils.Subject:     "ANY2CNT",
+				utils.Destination: "+4986517174963",
+				utils.AnswerTime:  time.Date(2018, 8, 24, 16, 00, 26, 0, time.UTC),
+				utils.Usage:       1 * time.Minute,
+				"field_extr1":     "val_extr1",
+				"fieldextr2":      "valextr2",
+			},
+		},
+	}
+
+	if err := cdrsRpc.Call(utils.CDRsV2ProcessEvent, args2, &rplProcEv); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	}
+
+	if err := cdrsRpc.Call(utils.APIerSv2GetAccount,
+		&utils.AttrGetAccount{Tenant: "cgrates.org", Account: "voiceAccount"}, &acnt); err != nil {
+		t.Error(err)
+	} else if len(acnt.BalanceMap) != 1 || acnt.BalanceMap[utils.VOICE][0].Value != 540000000000 {
+		t.Errorf("Unexpected balance received: %+v", acnt.BalanceMap[utils.VOICE][0])
+	}
+}
+
 func testV2CDRsKillEngine(t *testing.T) {
 	if err := engine.KillEngine(*waitRater); err != nil {
 		t.Error(err)
--- a/config/config.go
+++ b/config/config.go
@@ -299,7 +299,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.MetaFlatstore})
+	utils.MetaPartialCSV, utils.MetaFlatstore, utils.MetaJSON})

 func (cfg *CGRConfig) LazySanityCheck() {
 	for _, cdrePrfl := range cfg.cdrsCfg.OnlineCDRExports {
--- a/config/configsanity.go
+++ b/config/configsanity.go
@@ -417,7 +417,7 @@ func (cfg *CGRConfig) checkConfigSanity() error {
 				if rdr.RunDelay > 0 {
 					return fmt.Errorf("<%s> the RunDelay field can not be bigger than zero for reader with ID: %s", utils.ERs, rdr.ID)
 				}
-			case utils.MetaFileXML, utils.MetaFileFWV:
+			case utils.MetaFileXML, utils.MetaFileFWV, utils.MetaJSON:
 				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
@@ -349,6 +349,27 @@
 				{"tag": "DisconnectCause", "path": "*cgreq.DisconnectCause", "type": "*variable", "value": "~*req.4; ;~*req.5", "mandatory": true},
 				{"tag": "DialogId", "path": "*cgreq.DialogId", "type": "*variable", "value": "~*req.11"}
 			],
+		},
+		{
+			"id": "JSONReader",
+			"run_delay":  "-1",
+			"type": "*json",
+			"source_path": "/tmp/ErsJSON/in",
+			"flags": ["*cdrs"],
+			"processed_path": "/tmp/ErsJSON/out",
+			"fields":[
+				{"tag": "ToR", "path": "*cgreq.ToR", "type": "*constant", "value": "*voice", "mandatory": true},
+				{"tag": "OriginID", "path": "*cgreq.OriginID", "type": "*variable", "value": "~*req.OriginID", "mandatory": true},
+				{"tag": "RequestType", "path": "*cgreq.RequestType", "type": "*variable", "value": "~*req.RequestType", "mandatory": true},
+				{"tag": "Tenant", "path": "*cgreq.Tenant", "type": "*variable", "value": "~*req.Tenant", "mandatory": true},
+				{"tag": "Category", "path": "*cgreq.Category", "type": "*constant", "value": "call", "mandatory": true},
+				{"tag": "Account", "path": "*cgreq.Account", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Subject", "path": "*cgreq.Subject", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Destination", "path": "*cgreq.Destination", "type": "*variable", "value": "~*req.Destination:s/0([1-9]\\d+)/+49${1}/", "mandatory": true},
+				{"tag": "SetupTime", "path": "*cgreq.SetupTime", "type": "*variable", "value": "~*req.SetupTime", "mandatory": true},
+				{"tag": "AnswerTime", "path": "*cgreq.AnswerTime", "type": "*variable", "value": "~*req.AnswerTime", "mandatory": true},
+				{"tag": "Usage", "path": "*cgreq.Usage", "type": "*variable", "value": "~*req.Usage", "mandatory": true},
+			],
 		}
 	],
 },
--- a/data/conf/samples/ers_mongo/cgrates.json
+++ b/data/conf/samples/ers_mongo/cgrates.json
@@ -350,6 +350,27 @@
 				{"tag": "DisconnectCause", "path": "*cgreq.DisconnectCause", "type": "*variable", "value": "~*req.4; ;~*req.5", "mandatory": true},
 				{"tag": "DialogId", "path": "*cgreq.DialogId", "type": "*variable", "value": "~*req.11"}
 			],
+		},
+		{
+			"id": "JSONReader",
+			"run_delay":  "-1",
+			"type": "*json",
+			"source_path": "/tmp/ErsJSON/in",
+			"flags": ["*cdrs"],
+			"processed_path": "/tmp/ErsJSON/out",
+			"fields":[
+				{"tag": "ToR", "path": "*cgreq.ToR", "type": "*constant", "value": "*voice", "mandatory": true},
+				{"tag": "OriginID", "path": "*cgreq.OriginID", "type": "*variable", "value": "~*req.OriginID", "mandatory": true},
+				{"tag": "RequestType", "path": "*cgreq.RequestType", "type": "*variable", "value": "~*req.RequestType", "mandatory": true},
+				{"tag": "Tenant", "path": "*cgreq.Tenant", "type": "*variable", "value": "~*req.Tenant", "mandatory": true},
+				{"tag": "Category", "path": "*cgreq.Category", "type": "*constant", "value": "call", "mandatory": true},
+				{"tag": "Account", "path": "*cgreq.Account", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Subject", "path": "*cgreq.Subject", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Destination", "path": "*cgreq.Destination", "type": "*variable", "value": "~*req.Destination:s/0([1-9]\\d+)/+49${1}/", "mandatory": true},
+				{"tag": "SetupTime", "path": "*cgreq.SetupTime", "type": "*variable", "value": "~*req.SetupTime", "mandatory": true},
+				{"tag": "AnswerTime", "path": "*cgreq.AnswerTime", "type": "*variable", "value": "~*req.AnswerTime", "mandatory": true},
+				{"tag": "Usage", "path": "*cgreq.Usage", "type": "*variable", "value": "~*req.Usage", "mandatory": true},
+			],
 		}
 	]
 },
--- a/data/conf/samples/ers_mysql/cgrates.json
+++ b/data/conf/samples/ers_mysql/cgrates.json
@@ -347,6 +347,27 @@
 				{"tag": "DisconnectCause", "path": "*cgreq.DisconnectCause", "type": "*variable", "value": "~*req.4; ;~*req.5", "mandatory": true},
 				{"tag": "DialogId", "path": "*cgreq.DialogId", "type": "*variable", "value": "~*req.11"}
 			],
+		},
+		{
+			"id": "JSONReader",
+			"run_delay":  "-1",
+			"type": "*json",
+			"source_path": "/tmp/ErsJSON/in",
+			"flags": ["*cdrs"],
+			"processed_path": "/tmp/ErsJSON/out",
+			"fields":[
+				{"tag": "ToR", "path": "*cgreq.ToR", "type": "*constant", "value": "*voice", "mandatory": true},
+				{"tag": "OriginID", "path": "*cgreq.OriginID", "type": "*variable", "value": "~*req.OriginID", "mandatory": true},
+				{"tag": "RequestType", "path": "*cgreq.RequestType", "type": "*variable", "value": "~*req.RequestType", "mandatory": true},
+				{"tag": "Tenant", "path": "*cgreq.Tenant", "type": "*variable", "value": "~*req.Tenant", "mandatory": true},
+				{"tag": "Category", "path": "*cgreq.Category", "type": "*constant", "value": "call", "mandatory": true},
+				{"tag": "Account", "path": "*cgreq.Account", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Subject", "path": "*cgreq.Subject", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Destination", "path": "*cgreq.Destination", "type": "*variable", "value": "~*req.Destination:s/0([1-9]\\d+)/+49${1}/", "mandatory": true},
+				{"tag": "SetupTime", "path": "*cgreq.SetupTime", "type": "*variable", "value": "~*req.SetupTime", "mandatory": true},
+				{"tag": "AnswerTime", "path": "*cgreq.AnswerTime", "type": "*variable", "value": "~*req.AnswerTime", "mandatory": true},
+				{"tag": "Usage", "path": "*cgreq.Usage", "type": "*variable", "value": "~*req.Usage", "mandatory": true},
+			],
 		}
 	]
 },
--- a/data/conf/samples/ers_postgres/cgrates.json
+++ b/data/conf/samples/ers_postgres/cgrates.json
@@ -344,6 +344,27 @@
 				{"tag": "DisconnectCause", "path": "*cgreq.DisconnectCause", "type": "*variable", "value": "~*req.4; ;~*req.5", "mandatory": true},
 				{"tag": "DialogId", "path": "*cgreq.DialogId", "type": "*variable", "value": "~*req.11"}
 			],
+		},
+		{
+			"id": "JSONReader",
+			"run_delay":  "-1",
+			"type": "*json",
+			"source_path": "/tmp/ErsJSON/in",
+			"flags": ["*cdrs"],
+			"processed_path": "/tmp/ErsJSON/out",
+			"fields":[
+				{"tag": "ToR", "path": "*cgreq.ToR", "type": "*constant", "value": "*voice", "mandatory": true},
+				{"tag": "OriginID", "path": "*cgreq.OriginID", "type": "*variable", "value": "~*req.OriginID", "mandatory": true},
+				{"tag": "RequestType", "path": "*cgreq.RequestType", "type": "*variable", "value": "~*req.RequestType", "mandatory": true},
+				{"tag": "Tenant", "path": "*cgreq.Tenant", "type": "*variable", "value": "~*req.Tenant", "mandatory": true},
+				{"tag": "Category", "path": "*cgreq.Category", "type": "*constant", "value": "call", "mandatory": true},
+				{"tag": "Account", "path": "*cgreq.Account", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Subject", "path": "*cgreq.Subject", "type": "*variable", "value": "~*req.Account", "mandatory": true},
+				{"tag": "Destination", "path": "*cgreq.Destination", "type": "*variable", "value": "~*req.Destination:s/0([1-9]\\d+)/+49${1}/", "mandatory": true},
+				{"tag": "SetupTime", "path": "*cgreq.SetupTime", "type": "*variable", "value": "~*req.SetupTime", "mandatory": true},
+				{"tag": "AnswerTime", "path": "*cgreq.AnswerTime", "type": "*variable", "value": "~*req.AnswerTime", "mandatory": true},
+				{"tag": "Usage", "path": "*cgreq.Usage", "type": "*variable", "value": "~*req.Usage", "mandatory": true},
+			],
 		}
 	]
 },
--- a/ers/filecsv_it_test.go
+++ b/ers/filecsv_it_test.go
@@ -56,7 +56,7 @@ accid22;*postpaid;itsyscom.com;1002;+4986517174963;2013-02-03 19:54:00;123;val_e
 accid23;*rated;cgrates.org;1001;086517174963;2013-02-03 19:54:00;26;val_extra3;"";val_extra1`

 	csvTests = []func(t *testing.T){
-		testCsvITCreateCdrDirs,
+		testCreateDirs,
 		testCsvITInitConfig,
 		testCsvITInitCdrDb,
 		testCsvITResetDataDb,
@@ -73,7 +73,7 @@ accid23;*rated;cgrates.org;1001;086517174963;2013-02-03 19:54:00;26;val_extra3;"
 		testCsvITProcessFilteredCDR,
 		testCsvITAnalyzeFilteredCDR,
 		testCsvITProcessedFiles,
-		testCsvITCleanupFiles,
+		testCleanupFiles,
 		testCsvITKillEngine,
 	}
 )
@@ -120,23 +120,6 @@ func testCsvITResetDataDb(t *testing.T) {
 	}
 }

-func testCsvITCreateCdrDirs(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 testCsvITStartEngine(t *testing.T) {
 	if _, err := engine.StopStartEngine(csvCfgPath, *waitRater); err != nil {
 		t.Fatal(err)
@@ -347,17 +330,6 @@ func testCsvITProcessedFiles(t *testing.T) {
 	}
 }

-func testCsvITCleanupFiles(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 testCsvITKillEngine(t *testing.T) {
 	if err := engine.KillEngine(*waitRater); err != nil {
 		t.Error(err)
--- a/ers/filefwv_it_test.go
+++ b/ers/filefwv_it_test.go
@@ -41,7 +41,7 @@ var (
 	fwvRPC     *rpc.Client

 	fwvTests = []func(t *testing.T){
-		testFWVITCreateCdrDirs,
+		testCreateDirs,
 		testFWVITInitConfig,
 		testFWVITInitCdrDb,
 		testFWVITResetDataDb,
@@ -50,7 +50,7 @@ var (
 		testFWVITLoadTPFromFolder,
 		testFWVITHandleCdr1File,
 		testFWVITAnalyseCDRs,
-		testFWVITCleanupFiles,
+		testCleanupFiles,
 		testFWVITKillEngine,
 	}
 )
@@ -74,23 +74,6 @@ func TestFWVReadFile(t *testing.T) {
 	}
 }

-func testFWVITCreateCdrDirs(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 testFWVITInitConfig(t *testing.T) {
 	var err error
 	fwvCfgPath = path.Join(*dataDir, "conf", "samples", fwvCfgDIR)
@@ -211,17 +194,6 @@ func testFWVITAnalyseCDRs(t *testing.T) {
 	}
 }

-func testFWVITCleanupFiles(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 testFWVITKillEngine(t *testing.T) {
 	if err := engine.KillEngine(*waitRater); err != nil {
 		t.Error(err)
--- a/ers/filejson.go
+++ b/ers/filejson.go
@@ -0,0 +1,178 @@
+/*
+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 (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/cgrates/cgrates/agents"
+	"github.com/cgrates/cgrates/utils"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/engine"
+)
+
+func NewJSONFileER(cfg *config.CGRConfig, cfgIdx int,
+	rdrEvents chan *erEvent, rdrErr chan error,
+	fltrS *engine.FilterS, rdrExit chan struct{}) (er EventReader, err error) {
+	srcPath := cfg.ERsCfg().Readers[cfgIdx].SourcePath
+	if strings.HasSuffix(srcPath, utils.Slash) {
+		srcPath = srcPath[:len(srcPath)-1]
+	}
+	jsonEr := &JSONFileER{
+		cgrCfg:    cfg,
+		cfgIdx:    cfgIdx,
+		fltrS:     fltrS,
+		rdrDir:    srcPath,
+		rdrEvents: rdrEvents,
+		rdrError:  rdrErr,
+		rdrExit:   rdrExit,
+		conReqs:   make(chan struct{}, cfg.ERsCfg().Readers[cfgIdx].ConcurrentReqs)}
+	var processFile struct{}
+	for i := 0; i < cfg.ERsCfg().Readers[cfgIdx].ConcurrentReqs; i++ {
+		jsonEr.conReqs <- processFile // Empty initiate so we do not need to wait later when we pop
+	}
+	return jsonEr, nil
+}
+
+// JSONFileER implements EventReader interface for .json files
+type JSONFileER struct {
+	sync.RWMutex
+	cgrCfg    *config.CGRConfig
+	cfgIdx    int // index of config instance within ERsCfg.Readers
+	fltrS     *engine.FilterS
+	rdrDir    string
+	rdrEvents chan *erEvent // channel to dispatch the events created to
+	rdrError  chan error
+	rdrExit   chan struct{}
+	conReqs   chan struct{} // limit number of opened files
+}
+
+func (rdr *JSONFileER) Config() *config.EventReaderCfg {
+	return rdr.cgrCfg.ERsCfg().Readers[rdr.cfgIdx]
+}
+
+func (rdr *JSONFileER) Serve() (err error) {
+	switch rdr.Config().RunDelay {
+	case time.Duration(0): // 0 disables the automatic read, maybe done per API
+		return
+	case time.Duration(-1):
+		return watchDir(rdr.rdrDir, rdr.processFile,
+			utils.ERs, rdr.rdrExit)
+	default:
+		go func() {
+			for {
+				// Not automated, process and sleep approach
+				select {
+				case <-rdr.rdrExit:
+					utils.Logger.Info(
+						fmt.Sprintf("<%s> stop monitoring path <%s>",
+							utils.ERs, rdr.rdrDir))
+					return
+				default:
+				}
+				filesInDir, _ := ioutil.ReadDir(rdr.rdrDir)
+				for _, file := range filesInDir {
+					if !strings.HasSuffix(file.Name(), utils.JSNSuffix) { // hardcoded file extension for json event reader
+						continue // used in order to filter the files from directory
+					}
+					go func(fileName string) {
+						if err := rdr.processFile(rdr.rdrDir, fileName); err != nil {
+							utils.Logger.Warning(
+								fmt.Sprintf("<%s> processing file %s, error: %s",
+									utils.ERs, fileName, err.Error()))
+						}
+					}(file.Name())
+				}
+				time.Sleep(rdr.Config().RunDelay)
+			}
+		}()
+	}
+	return
+}
+
+// processFile is called for each file in a directory and dispatches erEvents from it
+func (rdr *JSONFileER) processFile(fPath, fName string) (err error) {
+	if cap(rdr.conReqs) != 0 { // 0 goes for no limit
+		processFile := <-rdr.conReqs // Queue here for maxOpenFiles
+		defer func() { rdr.conReqs <- processFile }()
+	}
+	absPath := path.Join(fPath, fName)
+	utils.Logger.Info(
+		fmt.Sprintf("<%s> parsing <%s>", utils.ERs, absPath))
+	var file *os.File
+	if file, err = os.Open(absPath); err != nil {
+		return
+	}
+	defer file.Close()
+	timeStart := time.Now()
+	var byteValue []byte
+	if byteValue, err = ioutil.ReadAll(file); err != nil {
+		return
+	}
+
+	var data map[string]interface{}
+	if err = json.Unmarshal(byteValue, &data); err != nil {
+		return
+	}
+
+	evsPosted := 0
+	reqVars := map[string]interface{}{utils.FileName: fName}
+
+	agReq := agents.NewAgentRequest(
+		config.NewNavigableMap(data), reqVars, nil, nil, rdr.Config().Tenant,
+		rdr.cgrCfg.GeneralCfg().DefaultTenant,
+		utils.FirstNonEmpty(rdr.Config().Timezone,
+			rdr.cgrCfg.GeneralCfg().DefaultTimezone),
+		rdr.fltrS, nil, nil) // create an AgentRequest
+	if pass, err := rdr.fltrS.Pass(agReq.Tenant, rdr.Config().Filters,
+		agReq); err != nil || !pass {
+		return err
+	}
+	if err = agReq.SetFields(rdr.Config().Fields); err != nil {
+		utils.Logger.Warning(
+			fmt.Sprintf("<%s> reading file: <%s>  ignoring due to error: <%s>",
+				utils.ERs, absPath, err.Error()))
+		return
+	}
+	rdr.rdrEvents <- &erEvent{cgrEvent: agReq.CGRRequest.AsCGREvent(
+		agReq.Tenant, utils.NestingSep),
+		rdrCfg: rdr.Config()}
+	evsPosted++
+
+	if rdr.Config().ProcessedPath != "" {
+		// Finished with file, move it to processed folder
+		outPath := path.Join(rdr.Config().ProcessedPath, fName)
+		if err = os.Rename(absPath, outPath); err != nil {
+			return
+		}
+	}
+
+	utils.Logger.Info(
+		fmt.Sprintf("%s finished processing file <%s>. Events posted: %d, run duration: %s",
+			utils.ERs, absPath, evsPosted, time.Now().Sub(timeStart)))
+	return
+}
--- a/ers/filejson_it_test.go
+++ b/ers/filejson_it_test.go
@@ -0,0 +1,226 @@
+// +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"
+	v2 "github.com/cgrates/cgrates/apier/v2"
+
+	"github.com/cgrates/cgrates/engine"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/utils"
+)
+
+var (
+	jsonCfgPath string
+	jsonCfgDIR  string
+	jsonCfg     *config.CGRConfig
+	jsonRPC     *rpc.Client
+
+	fileContent = `
+{
+	"Tenant": "cgrates.org",
+	"Account": "voiceAccount",
+	"AnswerTime": "2018-08-24T16:00:26Z",
+	"SetupTime": "2018-08-24T16:00:26Z",
+	"Destination": "+4986517174963",
+	"OriginHost": "192.168.1.1",
+	"OriginID": "testJsonCDR",
+	"RequestType": "*pseudoprepaid",
+	"Source": "jsonFile",
+	"Usage": 120000000000
+}`
+	jsonTests = []func(t *testing.T){
+		testCreateDirs,
+		testJSONInitConfig,
+		testJSONInitCdrDb,
+		testJSONResetDataDb,
+		testJSONStartEngine,
+		testJSONRpcConn,
+		testJSONAddData,
+		testJSONHandleFile,
+		testJSONVerify,
+		testCleanupFiles,
+		testJSONKillEngine,
+	}
+)
+
+func TestJSONReadFile(t *testing.T) {
+	switch *dbType {
+	case utils.MetaInternal:
+		jsonCfgDIR = "ers_internal"
+	case utils.MetaMySQL:
+		jsonCfgDIR = "ers_mysql"
+	case utils.MetaMongo:
+		jsonCfgDIR = "ers_mongo"
+	case utils.MetaPostgres:
+		jsonCfgDIR = "ers_postgres"
+	default:
+		t.Fatal("Unknown Database type")
+	}
+
+	for _, test := range jsonTests {
+		t.Run(jsonCfgDIR, test)
+	}
+}
+
+func testJSONInitConfig(t *testing.T) {
+	var err error
+	jsonCfgPath = path.Join(*dataDir, "conf", "samples", jsonCfgDIR)
+	if jsonCfg, err = config.NewCGRConfigFromPath(jsonCfgPath); err != nil {
+		t.Fatal("Got config error: ", err.Error())
+	}
+}
+
+// InitDb so we can rely on count
+func testJSONInitCdrDb(t *testing.T) {
+	if err := engine.InitStorDb(jsonCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Remove data in both rating and accounting db
+func testJSONResetDataDb(t *testing.T) {
+	if err := engine.InitDataDb(jsonCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func testJSONStartEngine(t *testing.T) {
+	if _, err := engine.StopStartEngine(jsonCfgPath, *waitRater); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Connect rpc client to rater
+func testJSONRpcConn(t *testing.T) {
+	var err error
+	jsonRPC, err = newRPCClient(jsonCfg.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 testJSONAddData(t *testing.T) {
+	var reply string
+	//add a charger
+	chargerProfile := &v1.ChargerWithCache{
+		ChargerProfile: &engine.ChargerProfile{
+			Tenant: "cgrates.org",
+			ID:     "Default",
+			ActivationInterval: &utils.ActivationInterval{
+				ActivationTime: time.Date(2014, 7, 14, 14, 35, 0, 0, time.UTC),
+			},
+			RunID:        utils.MetaDefault,
+			AttributeIDs: []string{"*none"},
+			Weight:       20,
+		},
+		Cache: utils.StringPointer(utils.MetaReload),
+	}
+	if err := jsonRPC.Call(utils.APIerSv1SetChargerProfile, chargerProfile, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply returned", reply)
+	}
+
+	attrSetAcnt := v2.AttrSetAccount{
+		Tenant:  "cgrates.org",
+		Account: "voiceAccount",
+	}
+	if err := jsonRPC.Call(utils.APIerSv2SetAccount, attrSetAcnt, &reply); err != nil {
+		t.Fatal(err)
+	}
+	attrs := &utils.AttrSetBalance{
+		Tenant:      "cgrates.org",
+		Account:     "voiceAccount",
+		BalanceType: utils.VOICE,
+		Value:       600000000000,
+		Balance: map[string]interface{}{
+			utils.ID:        utils.MetaDefault,
+			"RatingSubject": "*zero1m",
+			utils.Weight:    10.0,
+		},
+	}
+	if err := jsonRPC.Call(utils.APIerSv2SetBalance, attrs, &reply); err != nil {
+		t.Fatal(err)
+	}
+
+	var acnt *engine.Account
+	if err := jsonRPC.Call(utils.APIerSv2GetAccount,
+		&utils.AttrGetAccount{Tenant: "cgrates.org", Account: "voiceAccount"}, &acnt); err != nil {
+		t.Error(err)
+	} else if len(acnt.BalanceMap) != 1 || acnt.BalanceMap[utils.VOICE][0].Value != 600000000000 {
+		t.Errorf("Unexpected balance received: %+v", acnt.BalanceMap[utils.VOICE][0])
+	}
+}
+
+// The default scenario, out of cdrc defined in .cfg file
+func testJSONHandleFile(t *testing.T) {
+	fileName := "file1.json"
+	tmpFilePath := path.Join("/tmp", fileName)
+	if err := ioutil.WriteFile(tmpFilePath, []byte(fileContent), 0644); err != nil {
+		t.Fatal(err.Error())
+	}
+	if err := os.Rename(tmpFilePath, path.Join("/tmp/ErsJSON/in", fileName)); err != nil {
+		t.Fatal("Error moving file to processing directory: ", err)
+	}
+	time.Sleep(100 * time.Millisecond)
+}
+
+func testJSONVerify(t *testing.T) {
+	var cdrs []*engine.CDR
+	args := &utils.RPCCDRsFilterWithArgDispatcher{
+		RPCCDRsFilter: &utils.RPCCDRsFilter{
+			OriginIDs: []string{"testJsonCDR"},
+		},
+	}
+	if err := jsonRPC.Call(utils.CDRsV1GetCDRs, args, &cdrs); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if len(cdrs) != 1 {
+		t.Error("Unexpected number of CDRs returned: ", len(cdrs))
+	} else {
+		if cdrs[0].Usage != 2*time.Minute {
+			t.Errorf("Unexpected usage for CDR: %d", cdrs[0].Usage)
+		}
+	}
+
+	var acnt *engine.Account
+	if err := jsonRPC.Call(utils.APIerSv2GetAccount,
+		&utils.AttrGetAccount{Tenant: "cgrates.org", Account: "voiceAccount"}, &acnt); err != nil {
+		t.Error(err)
+	} else if len(acnt.BalanceMap) != 1 || acnt.BalanceMap[utils.VOICE][0].Value != 480000000000 {
+		t.Errorf("Unexpected balance received: %+v", acnt.BalanceMap[utils.VOICE][0])
+	}
+}
+
+func testJSONKillEngine(t *testing.T) {
+	if err := engine.KillEngine(*waitRater); err != nil {
+		t.Error(err)
+	}
+}
--- a/ers/filexml_it_test.go
+++ b/ers/filexml_it_test.go
@@ -40,7 +40,7 @@ var (
 	xmlRPC     *rpc.Client

 	xmlTests = []func(t *testing.T){
-		testXMLITCreateCdrDirs,
+		testCreateDirs,
 		testXMLITInitConfig,
 		testXMLITInitCdrDb,
 		testXMLITResetDataDb,
@@ -49,7 +49,7 @@ var (
 		testXMLITLoadTPFromFolder,
 		testXMLITHandleCdr1File,
 		testXmlITAnalyseCDRs,
-		testXMLITCleanupFiles,
+		testCleanupFiles,
 		testXMLITKillEngine,
 	}
 )
@@ -72,23 +72,6 @@ func TestXMLReadFile(t *testing.T) {
 	}
 }

-func testXMLITCreateCdrDirs(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 testXMLITInitConfig(t *testing.T) {
 	var err error
 	xmlCfgPath = path.Join(*dataDir, "conf", "samples", xmlCfgDIR)
@@ -297,17 +280,6 @@ func testXmlITAnalyseCDRs(t *testing.T) {
 	}
 }

-func testXMLITCleanupFiles(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 testXMLITKillEngine(t *testing.T) {
 	if err := engine.KillEngine(*waitRater); err != nil {
 		t.Error(err)
--- a/ers/flatstore_it_test.go
+++ b/ers/flatstore_it_test.go
@@ -60,7 +60,7 @@ INVITE|324cb497|d4af7023|8deaadf2ae9a17809a391f05af31afb0@0:0:0:0:0:0:0:0|486|Bu
 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,
+		testCreateDirs,
 		testFlatstoreITInitConfig,
 		testFlatstoreITInitCdrDb,
 		testFlatstoreITResetDataDb,
@@ -69,7 +69,7 @@ INVITE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9p@0:0:0:0:0:0:0:0|200|OK
 		testFlatstoreITLoadTPFromFolder,
 		testFlatstoreITHandleCdr1File,
 		testFlatstoreITAnalyseCDRs,
-		testFlatstoreITCleanupFiles,
+		testCleanupFiles,
 		testFlatstoreITKillEngine,
 	}
 )
@@ -92,23 +92,6 @@ func TestFlatstoreFile(t *testing.T) {
 	}
 }

-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)
@@ -217,17 +200,6 @@ func testFlatstoreITAnalyseCDRs(t *testing.T) {
 	}
 }

-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/lib_test.go
+++ b/ers/lib_test.go
@@ -23,6 +23,8 @@ import (
 	"flag"
 	"net/rpc"
 	"net/rpc/jsonrpc"
+	"os"
+	"testing"

 	"github.com/cgrates/cgrates/config"
 	"github.com/cgrates/cgrates/utils"
@@ -45,3 +47,32 @@ func newRPCClient(cfg *config.ListenCfg) (c *rpc.Client, err error) {
 		return nil, errors.New("UNSUPPORTED_RPC")
 	}
 }
+
+func testCreateDirs(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", "/tmp/ErsJSON/in", "/tmp/ErsJSON/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 testCleanupFiles(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",
+		"/tmp/ErsJSON"} {
+		if err := os.RemoveAll(dir); err != nil {
+			t.Fatal("Error removing folder: ", dir, err)
+		}
+	}
+}
--- a/ers/partial_csv_it_test.go
+++ b/ers/partial_csv_it_test.go
@@ -44,7 +44,7 @@ var (
 	partRPC     *rpc.Client

 	partTests = []func(t *testing.T){
-		testPartITCreateCdrDirs,
+		testCreateDirs,
 		testPartITInitConfig,
 		testPartITInitCdrDb,
 		testPartITResetDataDb,
@@ -56,7 +56,7 @@ var (
 		testPartITHandleCdr3File,
 		testPartITVerifyFiles,
 		testPartITAnalyseCDRs,
-		testPartITCleanupFiles,
+		testCleanupFiles,
 		testPartITKillEngine,
 	}

@@ -90,23 +90,6 @@ func TestPartReadFile(t *testing.T) {
 	}
 }

-func testPartITCreateCdrDirs(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 testPartITInitConfig(t *testing.T) {
 	var err error
 	partCfgPath = path.Join(*dataDir, "conf", "samples", partCfgDIR)
@@ -239,17 +222,6 @@ func testPartITAnalyseCDRs(t *testing.T) {
 	}
 }

-func testPartITCleanupFiles(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 testPartITKillEngine(t *testing.T) {
 	if err := engine.KillEngine(*waitRater); err != nil {
 		t.Error(err)
--- a/ers/reader.go
+++ b/ers/reader.go
@@ -52,6 +52,8 @@ func NewEventReader(cfg *config.CGRConfig, cfgIdx int,
 		return NewSQLEventReader(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit)
 	case utils.MetaFlatstore:
 		return NewFlatstoreER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit)
+	case utils.MetaJSON:
+		return NewJSONFileER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit)
 	}
 	return
 }