Finished process event tests

data/conf/samples/cdrsv1processevent/cgrates.json

@@ -1,8 +1,11 @@
 {
 // CGRateS Configuration file
 //
-// Used in apier/v1/cdrs_it_test
+// Used in general_tests
 
+"general": {
+	"log_level": 7,
+},
 
 "data_db":{
 	"db_type": "*internal"
@@ -22,11 +25,26 @@
 	"attributes_conns": ["*internal"],
 },
 
+"cdre":{
+	"aws_test_file": {
+		"export_format": "*amqpv1_json_map",
+		"export_path": "amqps://guest:guest@localhost:25672/?queue_id=cgrates_cdrswrong",
+		"content_fields": [
+			{"tag": "CGRID", "type": "*composed", "value": "~CGRID", "field_id": "CGRID"},
+		],
+	},
+},
+
 "stats": {									
 	"enabled": true,
 	"store_interval": "-1",
 },
 
+"thresholds": {
+	"enabled": true,
+	"store_interval": "-1",
+},
+
 "rals": {
 	"enabled": true
 },
@@ -43,6 +61,8 @@
 	"attributes_conns": ["*internal"],
 	"chargers_conns": ["*internal"],
 	"stats_conns": ["*internal"],
+	"thresholds_conns": ["*internal"],
+	"online_cdr_exports": ["aws_test_file"],
 },
 
 

data/conf/samples/cdrsv1processeventmongo/cgrates.json

@@ -25,11 +25,26 @@
 	"attributes_conns": ["*internal"],
 },
 
+"cdre":{
+	"aws_test_file": {
+		"export_format": "*amqpv1_json_map",
+		"export_path": "amqps://guest:guest@localhost:25672/?queue_id=cgrates_cdrswrong",
+		"content_fields": [
+			{"tag": "CGRID", "type": "*composed", "value": "~CGRID", "field_id": "CGRID"},
+		],
+	},
+},
+
 "stats": {									
 	"enabled": true,
 	"store_interval": "-1",
 },
 
+"thresholds": {
+	"enabled": true,
+	"store_interval": "-1",
+},
+
 "rals": {
 	"enabled": true
 },
@@ -46,6 +61,8 @@
 	"attributes_conns": ["*internal"],
 	"chargers_conns": ["*internal"],
 	"stats_conns": ["*internal"],
+	"thresholds_conns": ["*internal"],
+	"online_cdr_exports": ["aws_test_file"],
 },
 
 

data/conf/samples/cdrsv1processeventmysql/cgrates.json

@@ -23,11 +23,26 @@
 	"attributes_conns": ["*internal"],
 },
 
+"cdre":{
+	"aws_test_file": {
+		"export_format": "*amqpv1_json_map",
+		"export_path": "amqps://guest:guest@localhost:25672/?queue_id=cgrates_cdrswrong",
+		"content_fields": [
+			{"tag": "CGRID", "type": "*composed", "value": "~CGRID", "field_id": "CGRID"},
+		],
+	},
+},
+
 "stats": {									
 	"enabled": true,
 	"store_interval": "-1",
 },
 
+"thresholds": {
+	"enabled": true,
+	"store_interval": "-1",
+},
+
 "rals": {
 	"enabled": true
 },
@@ -44,6 +59,8 @@
 	"attributes_conns": ["*internal"],
 	"chargers_conns": ["*internal"],
 	"stats_conns": ["*internal"],
+	"thresholds_conns": ["*internal"],
+	"online_cdr_exports": ["aws_test_file"],
 },
 
 

general_tests/cdrs_processevent_it_test.go

@@ -20,13 +20,18 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package general_tests
 
 import (
+	"fmt"
+	"io/ioutil"
 	"net/rpc"
+	"os"
 	"path"
 	"reflect"
 	"sort"
+	"strings"
 	"testing"
 	"time"
 
+	v2 "github.com/cgrates/cgrates/apier/v2"
 	"github.com/cgrates/cgrates/config"
 	"github.com/cgrates/cgrates/engine"
 	"github.com/cgrates/cgrates/utils"
@@ -44,10 +49,14 @@ var sTestsCDRsIT_ProcessEvent = []func(t *testing.T){
 	testV1CDRsStartEngine,
 	testV1CDRsRpcConn,
 	testV1CDRsLoadTariffPlanFromFolder,
+	testV1CDRsProcessEventExport,
 	testV1CDRsProcessEventAttrS,
 	testV1CDRsProcessEventChrgS,
 	testV1CDRsProcessEventRalS,
 	testV1CDRsProcessEventSts,
+	testV1CDRsProcessEventStore,
+	testV1CDRsProcessEventThreshold,
+	testV1CDRsProcessEventExportCheck,
 	testV1CDRsKillEngine,
 }
 
@@ -59,14 +68,14 @@ func TestCDRsITPEInternal(t *testing.T) {
 }
 
 func TestCDRsITPEMongo(t *testing.T) {
-	pecdrsConfDIR = "cdrsv1mongo"
+	pecdrsConfDIR = "cdrsv1processeventmongo"
 	for _, stest := range sTestsCDRsIT_ProcessEvent {
 		t.Run(pecdrsConfDIR, stest)
 	}
 }
 
 func TestCDRsITPEMySql(t *testing.T) {
-	pecdrsConfDIR = "cdrsv1mysql"
+	pecdrsConfDIR = "cdrsv1processeventmysql"
 	for _, stest := range sTestsCDRsIT_ProcessEvent {
 		t.Run(pecdrsConfDIR, stest)
 	}
@@ -144,7 +153,7 @@ func testV1CDRsProcessEventAttrS(t *testing.T) {
 		t.Errorf("Expecting: %v, received: %v", expectedVoice, rply)
 	}
 	argsEv := &engine.ArgV1ProcessEvent{
-		Flags: []string{utils.MetaAttributes, utils.MetaStore, "*chargers:false"},
+		Flags: []string{utils.MetaAttributes, utils.MetaStore, "*chargers:false", "*export:false"},
 		CGREvent: utils.CGREvent{
 			Tenant: "cgrates.org",
 			ID:     "test1",
@@ -202,7 +211,9 @@ func testV1CDRsProcessEventAttrS(t *testing.T) {
 		t.Errorf("Expecting: 1, received: %+v", len(cdrs))
 	} else if !reflect.DeepEqual(argsEv.Event["Account"], cdrs[0].Account) {
 		t.Errorf("Expecting: %+v, received: %+v", argsEv.Event["Account"], cdrs[0].Account)
-	} else if !reflect.DeepEqual(argsEv.Event["AnswerTime"], cdrs[0].AnswerTime) {
+	}
+	cdrs[0].AnswerTime = cdrs[0].AnswerTime.UTC()
+	if !reflect.DeepEqual(argsEv.Event["AnswerTime"], cdrs[0].AnswerTime) {
 		t.Errorf("Expecting: %+v, received: %+v", argsEv.Event["AnswerTime"], cdrs[0].AnswerTime)
 	} else if !reflect.DeepEqual(argsEv.Event["Destination"], cdrs[0].Destination) {
 		t.Errorf("Expecting: %+v, received: %+v", argsEv.Event["Destination"], cdrs[0].Destination)
@@ -224,7 +235,7 @@ func testV1CDRsProcessEventAttrS(t *testing.T) {
 
 func testV1CDRsProcessEventChrgS(t *testing.T) {
 	argsEv := &engine.ArgV1ProcessEvent{
-		Flags: []string{utils.MetaChargers, "*attributes:false"},
+		Flags: []string{utils.MetaChargers, "*attributes:false", "*export:false"},
 		CGREvent: utils.CGREvent{
 			Tenant: "cgrates.org",
 			ID:     "test2",
@@ -266,7 +277,7 @@ func testV1CDRsProcessEventChrgS(t *testing.T) {
 
 func testV1CDRsProcessEventRalS(t *testing.T) {
 	argsEv := &engine.ArgV1ProcessEvent{
-		Flags: []string{utils.MetaRALs, "*attributes:false", "*chargers:false"},
+		Flags: []string{utils.MetaRALs, "*attributes:false", "*chargers:false", "*export:false"},
 		CGREvent: utils.CGREvent{
 			Tenant: "cgrates.org",
 			ID:     "test3",
@@ -300,7 +311,7 @@ func testV1CDRsProcessEventRalS(t *testing.T) {
 
 func testV1CDRsProcessEventSts(t *testing.T) {
 	argsEv := &engine.ArgV1ProcessEvent{
-		Flags: []string{utils.MetaStatS, "*rals:false", "*attributes:false", "*chargers:false"},
+		Flags: []string{utils.MetaStatS, "*rals:false", "*attributes:false", "*chargers:false", "*export:false"},
 		CGREvent: utils.CGREvent{
 			Tenant: "cgrates.org",
 			ID:     "test4",
@@ -361,6 +372,8 @@ func testV1CDRsProcessEventSts(t *testing.T) {
 		t.Errorf("Expecting: 1, received: %+v", len(cdrs))
 	}
 	cdrs[0].OrderID = 0
+	cdrs[0].SetupTime = cdrs[0].SetupTime.UTC()
+	cdrs[0].AnswerTime = cdrs[0].AnswerTime.UTC()
 	if !reflect.DeepEqual(eOut[0], cdrs[0]) {
 		t.Errorf("\nExpected: %+v,\nreceived: %+v", utils.ToJSON(eOut[0]), utils.ToJSON(cdrs[0]))
 	}
@@ -380,6 +393,208 @@ func testV1CDRsProcessEventSts(t *testing.T) {
 	}
 }
 
+func testV1CDRsProcessEventStore(t *testing.T) {
+	argsEv := &engine.ArgV1ProcessEvent{
+		Flags: []string{"*store:false", "*attributes:false", "*chargers:false", "*export:false"},
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+			ID:     "test5",
+			Event: map[string]interface{}{
+				utils.RunID:       "testv1",
+				utils.OriginID:    "test5_processEvent",
+				utils.OriginHost:  "OriginHost5",
+				utils.RequestType: utils.META_PSEUDOPREPAID,
+				utils.Account:     "1001",
+				utils.Destination: "+4986517174963",
+				utils.AnswerTime:  time.Date(2019, 11, 27, 12, 21, 26, 0, time.UTC),
+				utils.Usage:       2 * time.Minute,
+			},
+		},
+	}
+	var reply string
+	if err := pecdrsRpc.Call(utils.CDRsV1ProcessEvent, argsEv, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply received: ", reply)
+	}
+	var cdrs []*engine.CDR
+	if err := pecdrsRpc.Call(utils.CDRsV1GetCDRs,
+		utils.RPCCDRsFilter{OriginHosts: []string{"OriginHost5"}}, &cdrs); err == nil ||
+		err.Error() != "SERVER_ERROR: NOT_FOUND" {
+		t.Fatal("Unexpected error: ", err.Error())
+	} else if len(cdrs) != 0 {
+		t.Errorf("Expecting: 0, received: %+v", len(cdrs))
+	}
+}
+
+func testV1CDRsProcessEventThreshold(t *testing.T) {
+	var reply string
+	if err := pecdrsRpc.Call(utils.ApierV2SetActions, &utils.AttrSetActions{
+		ActionsId: "ACT_LOG",
+		Actions: []*utils.TPAction{
+			&utils.TPAction{
+				Identifier: utils.LOG},
+			&utils.TPAction{
+				Identifier: utils.TOPUP_RESET, BalanceType: utils.VOICE,
+				Units: "10", ExpiryTime: "*unlimited",
+				DestinationIds: "*any", BalanceWeight: "10", Weight: 10},
+		},
+	}, &reply); err != nil && err.Error() != utils.ErrExists.Error() {
+		t.Error(err)
+	} else if reply != utils.OK {
+		t.Errorf("Calling ApierV2.SetActions received: %s", reply)
+	}
+	tPrfl := engine.ThresholdWithCache{
+		ThresholdProfile: &engine.ThresholdProfile{
+			Tenant: "cgrates.org",
+			ID:     "THD_Test",
+			FilterIDs: []string{
+				"*lt:~*req.CostDetails.AccountSummary.BalanceSummaries[0].Value:10",
+				"*string:~*req.Account:1005", // only for indexes
+			},
+			MaxHits:   -1,
+			Weight:    30,
+			ActionIDs: []string{"ACT_LOG"},
+			Async:     true,
+		},
+	}
+	if err := pecdrsRpc.Call(utils.ApierV1SetThresholdProfile, tPrfl, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply returned", reply)
+	}
+	attrSetAcnt := v2.AttrSetAccount{
+		Tenant:  "cgrates.org",
+		Account: "1005",
+		ExtraOptions: map[string]bool{
+			utils.AllowNegative: true,
+		},
+	}
+	if err := pecdrsRpc.Call(utils.ApierV2SetAccount, attrSetAcnt, &reply); err != nil {
+		t.Fatal(err)
+	}
+	attrs := &utils.AttrSetBalance{
+		Tenant:      "cgrates.org",
+		Account:     "1005",
+		BalanceType: utils.MONETARY,
+		Balance: map[string]interface{}{
+			utils.ID:     utils.MetaDefault,
+			utils.Value:  1,
+			utils.Weight: 10.0,
+		},
+	}
+	if err := pecdrsRpc.Call(utils.ApierV2SetBalance, attrs, &reply); err != nil {
+		t.Fatal(err)
+	}
+	args := &engine.ArgV1ProcessEvent{
+		Flags: []string{utils.MetaThresholds, utils.MetaRALs, utils.ConcatenatedKey(utils.MetaChargers, "false"), "*export:false"},
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+			Event: map[string]interface{}{
+				utils.OriginID:    "testV2CDRsProcessCDRWithThreshold",
+				utils.OriginHost:  "OriginHost6",
+				utils.Source:      "testV2CDRsProcessCDRWithThreshold",
+				utils.RequestType: utils.META_PREPAID,
+				utils.Category:    "call",
+				utils.Account:     "1005",
+				utils.Subject:     "ANY2CNT",
+				utils.Destination: "+4986517174963",
+				utils.AnswerTime:  time.Date(2018, 8, 24, 16, 00, 26, 0, time.UTC),
+				utils.Usage:       100 * time.Minute,
+				"field_extr1":     "val_extr1",
+				"fieldextr2":      "valextr2",
+			},
+		},
+	}
+	if err := pecdrsRpc.Call(utils.CDRsV1ProcessEvent, args, &reply); err != nil {
+		t.Error("Unexpected error: ", err)
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply received: ", reply)
+	}
+
+	var cdrs []*engine.CDR
+	if err := pecdrsRpc.Call(utils.CDRsV1GetCDRs, utils.RPCCDRsFilter{OriginHosts: []string{"OriginHost6"}}, &cdrs); err != nil {
+		t.Error("Unexpected error: ", err)
+	} else if len(cdrs) != 1 {
+		t.Errorf("Expecting: 1, received: %+v", len(cdrs))
+	}
+	var td engine.Threshold
+	if err := pecdrsRpc.Call(utils.ThresholdSv1GetThreshold,
+		&utils.TenantIDWithArgDispatcher{TenantID: &utils.TenantID{Tenant: "cgrates.org", ID: "THD_Test"}}, &td); err != nil {
+		t.Error(err)
+	} else if td.Hits != 1 {
+		t.Errorf("Expecting threshold to be hit once received: %v", td.Hits)
+	}
+	var acnt *engine.Account
+	acntAttrs := &utils.AttrGetAccount{
+		Tenant:  "cgrates.org",
+		Account: "1005"}
+	expectedVoice := 10.0
+	if err := pecdrsRpc.Call(utils.ApierV2GetAccount, acntAttrs, &acnt); err != nil {
+		t.Error(err)
+	} else if rply := acnt.BalanceMap[utils.VOICE].GetTotalValue(); rply != expectedVoice {
+		t.Errorf("Expecting: %v, received: %v", expectedVoice, rply)
+	}
+}
+func testV1CDRsProcessEventExport(t *testing.T) {
+	var reply string
+	args := &engine.ArgV1ProcessEvent{
+		Flags: []string{utils.MetaExport, "*store:false", "*attributes:false", "*chargers:false", "*stats:false", "*thresholds:false"},
+		CGREvent: utils.CGREvent{
+			Tenant: "cgrates.org",
+			ID:     "test7",
+			Event: map[string]interface{}{
+				utils.RunID:       "testv1",
+				utils.OriginID:    "test7_processEvent",
+				utils.OriginHost:  "OriginHost7",
+				utils.RequestType: utils.META_PSEUDOPREPAID,
+				utils.Account:     "1001",
+				utils.Destination: "+4986517174963",
+				utils.AnswerTime:  time.Date(2019, 11, 27, 12, 21, 26, 0, time.UTC),
+				utils.Usage:       2 * time.Minute,
+			},
+		},
+	}
+	if err := pecdrsRpc.Call(utils.CDRsV1ProcessEvent, args, &reply); err != nil {
+		t.Error("Unexpected error: ", err)
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply received: ", reply)
+	}
+	// var cdrs []*engine.CDR
+	// if err := pecdrsRpc.Call(utils.CDRsV1GetCDRs, utils.RPCCDRsFilter{OriginHosts: []string{"OriginHost7"}}, &cdrs); err != nil {
+	// 	t.Error("Unexpected error: ", err)
+	// } else if len(cdrs) != 1 {
+	// 	t.Errorf("Expecting: 1, received: %+v", len(cdrs))
+	// }
+}
+func testV1CDRsProcessEventExportCheck(t *testing.T) {
+	failoverContent := []byte(fmt.Sprintf(`{"CGRID":"%s"}`, utils.Sha1("test7_processEvent", "OriginHost7")))
+	filesInDir, _ := ioutil.ReadDir(pecdrsCfg.GeneralCfg().FailedPostsDir)
+	if len(filesInDir) == 0 {
+		t.Fatalf("No files in directory: %s", pecdrsCfg.GeneralCfg().FailedPostsDir)
+	}
+	var foundFile bool
+	var fileName string
+	for _, file := range filesInDir { // First file in directory is the one we need, harder to find it's name out of config
+		fileName = file.Name()
+		filePath := path.Join(pecdrsCfg.GeneralCfg().FailedPostsDir, fileName)
+		if strings.HasPrefix(fileName, "cdr|*amqpv1_json_map") {
+			foundFile = true
+			if readBytes, err := ioutil.ReadFile(filePath); err != nil {
+				t.Error(err)
+			} else if !reflect.DeepEqual(failoverContent, readBytes) { // Checking just the prefix should do since some content is dynamic
+				t.Errorf("Expecting: %v , received: %v", string(failoverContent), string(readBytes))
+			}
+			if err := os.Remove(filePath); err != nil {
+				t.Error("Failed removing file: ", filePath)
+			}
+		}
+	}
+
+	if !foundFile {
+		t.Fatal("Could not find the file in folder")
+	}
+}
 func testV1CDRsKillEngine(t *testing.T) {
 	if err := engine.KillEngine(*waitRater); err != nil {
 		t.Error(err)