Files
cgrates/apier/v1/dm_remote_it_test.go

986 lines
32 KiB
Go

//go:build integration
// +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 v1
import (
"net/rpc"
"path"
"reflect"
"sort"
"testing"
"time"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
var (
internalCfgPath string
internalCfgDirPath string
internalCfg *config.CGRConfig
internalRPC *rpc.Client
engineOneCfgPath string
engineOneCfgDirPath string
engineOneCfg *config.CGRConfig
engineOneRPC *rpc.Client
engineTwoCfgPath string
engineTwoCfgDirPath string
engineTwoCfg *config.CGRConfig
engineTwoRPC *rpc.Client
sTestsInternalRemoteIT = []func(t *testing.T){
testInternalRemoteITInitCfg,
testInternalRemoteITDataFlush,
testInternalRemoteITStartEngine,
testInternalRemoteITRPCConn,
testInternalRemoteLoadDataInEngineTwo,
testInternalRemoteITGetAccount,
testInternalRemoteITGetAttribute,
testInternalRemoteITGetThreshold,
testInternalRemoteITGetThresholdProfile,
testInternalRemoteITGetResource,
testInternalRemoteITGetResourceProfile,
testInternalRemoteITGetStatQueueProfile,
testInternalRemoteITGetSupplier,
testInternalRemoteITGetFilter,
testInternalRemoteITGetRatingPlan,
testInternalRemoteITGetRatingProfile,
testInternalRemoteITGetAction,
testInternalRemoteITGetActionPlan,
testInternalRemoteITGetAccountActionPlan,
testInternalRemoteITGetDestination,
testInternalRemoteITGetReverseDestination,
testInternalReplicationSetThreshold,
testInternalMatchThreshold,
testInternalAccountBalanceOperations,
testInternalSetAccount,
testInternalReplicateStats,
testInternalRemoteITKillEngine,
}
)
func TestInternalRemoteIT(t *testing.T) {
internalCfgDirPath = "internal"
switch *utils.DBType {
case utils.MetaInternal:
t.SkipNow()
case utils.MetaMySQL:
engineOneCfgDirPath = "engine1_redis"
engineTwoCfgDirPath = "engine2_redis"
case utils.MetaMongo:
engineOneCfgDirPath = "engine1_mongo"
engineTwoCfgDirPath = "engine2_mongo"
case utils.MetaPostgres:
t.SkipNow()
default:
t.Fatal("Unknown Database type")
}
if *utils.Encoding == utils.MetaGOB {
internalCfgDirPath += "_gob"
}
for _, stest := range sTestsInternalRemoteIT {
t.Run(*utils.DBType, stest)
}
}
func testInternalRemoteITInitCfg(t *testing.T) {
var err error
internalCfgPath = path.Join(*utils.DataDir, "conf", "samples", "remote_replication", internalCfgDirPath)
internalCfg, err = config.NewCGRConfigFromPath(internalCfgPath)
if err != nil {
t.Error(err)
}
internalCfg.DataFolderPath = *utils.DataDir // Share DataFolderPath through config towards StoreDb for Flush()
config.SetCgrConfig(internalCfg)
// prepare config for engine1
engineOneCfgPath = path.Join(*utils.DataDir, "conf", "samples",
"remote_replication", engineOneCfgDirPath)
engineOneCfg, err = config.NewCGRConfigFromPath(engineOneCfgPath)
if err != nil {
t.Error(err)
}
engineOneCfg.DataFolderPath = *utils.DataDir // Share DataFolderPath through config towards StoreDb for Flush()
// prepare config for engine2
engineTwoCfgPath = path.Join(*utils.DataDir, "conf", "samples",
"remote_replication", engineTwoCfgDirPath)
engineTwoCfg, err = config.NewCGRConfigFromPath(engineTwoCfgPath)
if err != nil {
t.Error(err)
}
engineTwoCfg.DataFolderPath = *utils.DataDir // Share DataFolderPath through config towards StoreDb for Flush()
}
func testInternalRemoteITDataFlush(t *testing.T) {
if err := engine.InitDataDb(engineOneCfg); err != nil {
t.Fatal(err)
}
time.Sleep(100 * time.Millisecond)
if err := engine.InitDataDb(engineTwoCfg); err != nil {
t.Fatal(err)
}
time.Sleep(100 * time.Millisecond)
}
func testInternalRemoteITStartEngine(t *testing.T) {
if _, err := engine.StartEngine(engineOneCfgPath, 500); err != nil {
t.Fatal(err)
}
if _, err := engine.StartEngine(engineTwoCfgPath, 500); err != nil {
t.Fatal(err)
}
if _, err := engine.StartEngine(internalCfgPath, 500); err != nil {
t.Fatal(err)
}
}
func testInternalRemoteITRPCConn(t *testing.T) {
var err error
engineOneRPC, err = newRPCClient(engineOneCfg.ListenCfg())
if err != nil {
t.Fatal(err)
}
time.Sleep(200 * time.Millisecond)
engineTwoRPC, err = newRPCClient(engineTwoCfg.ListenCfg())
if err != nil {
t.Fatal(err)
}
time.Sleep(200 * time.Millisecond)
internalRPC, err = newRPCClient(internalCfg.ListenCfg())
if err != nil {
t.Fatal(err)
}
time.Sleep(200 * time.Millisecond)
}
func testInternalRemoteLoadDataInEngineTwo(t *testing.T) {
var reply string
attrs := &utils.AttrLoadTpFromFolder{FolderPath: path.Join(*utils.DataDir, "tariffplans", "tutorial")}
if err := engineTwoRPC.Call(utils.APIerSv1LoadTariffPlanFromFolder, attrs, &reply); err != nil {
t.Error(err)
}
time.Sleep(500 * time.Millisecond)
}
func testInternalRemoteITGetAccount(t *testing.T) {
var acnt *engine.Account
expAcc := &engine.Account{
ID: "cgrates.org:1001",
BalanceMap: map[string]engine.Balances{
utils.MONETARY: []*engine.Balance{
{
ID: "testAccount",
Value: 10,
Weight: 10,
},
},
},
}
attrs := &utils.AttrGetAccount{
Tenant: "cgrates.org",
Account: "1001",
}
// make sure account exist in engine2
if err := engineTwoRPC.Call(utils.APIerSv2GetAccount, attrs, &acnt); err != nil {
t.Error(err)
} else if acnt.ID != expAcc.ID {
t.Errorf("expecting: %+v, received: %+v", expAcc.ID, acnt.ID)
} else if len(acnt.BalanceMap) != 1 {
t.Errorf("unexpected number of balances received: %+v", utils.ToJSON(acnt))
}
// check the account in internal
if err := internalRPC.Call(utils.APIerSv2GetAccount, attrs, &acnt); err != nil {
t.Error(err)
} else if acnt.ID != expAcc.ID {
t.Errorf("expecting: %+v, received: %+v", expAcc.ID, acnt.ID)
} else if len(acnt.BalanceMap) != 1 {
t.Errorf("unexpected number of balances received: %+v", utils.ToJSON(acnt))
}
attrs = &utils.AttrGetAccount{
Tenant: "cgrates.org",
Account: "nonexistAccount",
}
if err := internalRPC.Call(utils.APIerSv2GetAccount, attrs, &acnt); err == nil ||
err.Error() != utils.ErrNotFound.Error() {
t.Errorf("expecting: %+v, received: %+v", utils.ErrNotFound, err)
}
}
func testInternalRemoteITGetAttribute(t *testing.T) {
alsPrf = &AttributeWithCache{
AttributeProfile: &engine.AttributeProfile{
Tenant: "cgrates.org",
ID: "ATTR_1001_SIMPLEAUTH",
Contexts: []string{"simpleauth"},
FilterIDs: []string{"*string:~*req.Account:1001"},
Attributes: []*engine.Attribute{
{
Path: utils.MetaReq + utils.NestingSep + "Password",
FilterIDs: []string{},
Type: utils.META_CONSTANT,
Value: config.NewRSRParsersMustCompile("CGRateS.org", true, utils.INFIELD_SEP),
},
},
Weight: 20,
},
}
alsPrf.Compile()
var reply *engine.AttributeProfile
if err := internalRPC.Call(utils.APIerSv1GetAttributeProfile,
utils.TenantIDWithArgDispatcher{TenantID: &utils.TenantID{Tenant: "cgrates.org", ID: "ATTR_1001_SIMPLEAUTH"}}, &reply); err != nil {
t.Fatal(err)
}
if *utils.Encoding == utils.MetaGOB { // in gob emtpty slice is encoded as nil
alsPrf.AttributeProfile.Attributes[0].FilterIDs = nil
}
reply.Compile()
if !reflect.DeepEqual(alsPrf.AttributeProfile, reply) {
t.Errorf("Expecting : %+v, received: %+v", utils.ToJSON(alsPrf.AttributeProfile), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetThreshold(t *testing.T) {
var td engine.Threshold
eTd := engine.Threshold{Tenant: "cgrates.org", ID: "THD_ACNT_1001"}
if err := internalRPC.Call(utils.ThresholdSv1GetThreshold,
&utils.TenantIDWithArgDispatcher{TenantID: &utils.TenantID{Tenant: "cgrates.org", ID: "THD_ACNT_1001"}}, &td); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(eTd, td) {
t.Errorf("expecting: %+v, received: %+v", eTd, td)
}
}
func testInternalRemoteITGetThresholdProfile(t *testing.T) {
var reply *engine.ThresholdProfile
tPrfl = &engine.ThresholdWithCache{
ThresholdProfile: &engine.ThresholdProfile{
Tenant: "cgrates.org",
ID: "THD_ACNT_1001",
FilterIDs: []string{"FLTR_ACNT_1001"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 29, 15, 0, 0, 0, time.UTC),
},
MaxHits: 1,
MinHits: 1,
MinSleep: time.Duration(1 * time.Second),
Weight: 10.0,
ActionIDs: []string{"ACT_LOG_WARNING"},
Async: true,
},
}
if err := internalRPC.Call(utils.APIerSv1GetThresholdProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "THD_ACNT_1001"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tPrfl.ThresholdProfile, reply) {
t.Errorf("Expecting: %+v, received: %+v", utils.ToJSON(tPrfl.ThresholdProfile), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetResource(t *testing.T) {
var reply *engine.Resource
expectedResources := &engine.Resource{
Tenant: "cgrates.org",
ID: "ResGroup1",
Usages: map[string]*engine.ResourceUsage{},
}
if err := internalRPC.Call(utils.ResourceSv1GetResource,
&utils.TenantID{Tenant: "cgrates.org", ID: "ResGroup1"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(reply, expectedResources) {
t.Errorf("Expecting: %+v, received: %+v", utils.ToJSON(expectedResources), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetResourceProfile(t *testing.T) {
rlsPrf := &ResourceWithCache{
ResourceProfile: &engine.ResourceProfile{
Tenant: "cgrates.org",
ID: "ResGroup1",
FilterIDs: []string{"FLTR_RES"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 29, 15, 0, 0, 0, time.UTC),
},
UsageTTL: time.Duration(-1),
Limit: 7,
AllocationMessage: "",
Stored: true,
Weight: 10,
ThresholdIDs: []string{utils.META_NONE},
},
}
var reply *engine.ResourceProfile
if err := internalRPC.Call(utils.APIerSv1GetResourceProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: rlsPrf.ID}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(reply, rlsPrf.ResourceProfile) {
t.Errorf("Expecting: %+v, received: %+v", utils.ToJSON(rlsPrf.ResourceProfile), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetStatQueueProfile(t *testing.T) {
expStq := &engine.StatQueueProfile{
Tenant: "cgrates.org",
ID: "Stats2",
FilterIDs: []string{"FLTR_ACNT_1001_1002"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 29, 15, 0, 0, 0, time.UTC),
},
QueueLength: 100,
TTL: -1,
MinItems: 0,
Metrics: []*engine.MetricWithFilters{
{
MetricID: utils.MetaTCC,
},
{
MetricID: utils.MetaTCD,
},
},
Stored: false,
Blocker: true,
Weight: 30,
ThresholdIDs: []string{utils.META_NONE},
}
//reverse metric order
expStq2 := &engine.StatQueueProfile{
Tenant: "cgrates.org",
ID: "Stats2",
FilterIDs: []string{"FLTR_ACNT_1001_1002"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 29, 15, 0, 0, 0, time.UTC),
},
QueueLength: 100,
TTL: -1,
MinItems: 0,
Metrics: []*engine.MetricWithFilters{
{
MetricID: utils.MetaTCD,
},
{
MetricID: utils.MetaTCC,
},
},
Stored: false,
Blocker: true,
Weight: 30,
ThresholdIDs: []string{utils.META_NONE},
}
var reply *engine.StatQueueProfile
if err := internalRPC.Call(utils.APIerSv1GetStatQueueProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "Stats2"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(expStq, reply) && !reflect.DeepEqual(reply, expStq2) {
t.Errorf("Expecting: %+v or %+v, received: %+v", utils.ToJSON(expStq),
utils.ToJSON(expStq2), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetSupplier(t *testing.T) {
var reply *engine.SupplierProfile
splPrf := &engine.SupplierProfile{
Tenant: "cgrates.org",
ID: "SPL_ACNT_1001",
FilterIDs: []string{"FLTR_ACNT_1001"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2017, 11, 27, 0, 0, 0, 0, time.UTC),
},
Sorting: utils.MetaWeight,
SortingParameters: []string{},
Suppliers: []*engine.Supplier{
{
ID: "supplier1",
Weight: 10,
},
{
ID: "supplier2",
Weight: 20,
},
},
Weight: 20,
}
// supplier in reverse order
splPrf2 := &engine.SupplierProfile{
Tenant: "cgrates.org",
ID: "SPL_ACNT_1001",
FilterIDs: []string{"FLTR_ACNT_1001"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2017, 11, 27, 0, 0, 0, 0, time.UTC),
},
Sorting: utils.MetaWeight,
SortingParameters: []string{},
Suppliers: []*engine.Supplier{
{
ID: "supplier2",
Weight: 20,
},
{
ID: "supplier1",
Weight: 10,
},
},
Weight: 20,
}
if *utils.Encoding == utils.MetaGOB { // in gob emtpty slice is encoded as nil
splPrf.SortingParameters = nil
splPrf2.SortingParameters = nil
}
if err := internalRPC.Call(utils.APIerSv1GetSupplierProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "SPL_ACNT_1001"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(splPrf, reply) && !reflect.DeepEqual(splPrf2, reply) {
t.Errorf("Expecting: %+v, \n received: %+v", utils.ToJSON(splPrf), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetFilter(t *testing.T) {
expFltr := &engine.Filter{
Tenant: "cgrates.org",
ID: "FLTR_ACNT_1001",
Rules: []*engine.FilterRule{
{
Type: utils.MetaString,
Element: utils.DynamicDataPrefix + utils.MetaReq + utils.NestingSep + utils.Account,
Values: []string{"1001"},
},
},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 29, 15, 0, 0, 0, time.UTC),
},
}
var reply *engine.Filter
if err := internalRPC.Call(utils.APIerSv1GetFilter,
&utils.TenantID{Tenant: "cgrates.org", ID: "FLTR_ACNT_1001"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(expFltr, reply) {
t.Errorf("Expecting : %+v, received: %+v", utils.ToJSON(expFltr), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetRatingPlan(t *testing.T) {
var reply engine.RatingPlan
if err := internalRPC.Call(utils.APIerSv1GetRatingPlan, "RP_1001", &reply); err != nil {
t.Error(err.Error())
} else if reply.Id != "RP_1001" {
t.Errorf("Expected: %+v, received: %+v", "RP_1001", reply.Id)
}
}
func testInternalRemoteITGetRatingProfile(t *testing.T) {
var rpl engine.RatingProfile
attrGetRatingPlan := &utils.AttrGetRatingProfile{
Tenant: "cgrates.org", Category: "call", Subject: "1001"}
actTime, err := utils.ParseTimeDetectLayout("2014-01-14T00:00:00Z", "")
if err != nil {
t.Error(err)
}
expected := engine.RatingProfile{
Id: "*out:cgrates.org:call:1001",
RatingPlanActivations: engine.RatingPlanActivations{
{
ActivationTime: actTime,
RatingPlanId: "RP_1001",
},
},
}
if err := internalRPC.Call(utils.APIerSv1GetRatingProfile, attrGetRatingPlan, &rpl); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(expected, rpl) {
t.Errorf("Expected: %+v, received: %+v", utils.ToJSON(expected), utils.ToJSON(rpl))
}
}
func testInternalRemoteITGetAction(t *testing.T) {
expectActs := []*utils.TPAction{
{Identifier: utils.TOPUP_RESET, BalanceId: "test", BalanceType: utils.MONETARY,
Units: "10", BalanceWeight: "10", BalanceBlocker: "false",
BalanceDisabled: "false", ExpiryTime: utils.UNLIMITED, Weight: 10.0}}
var reply []*utils.TPAction
if err := internalRPC.Call(utils.APIerSv1GetActions, "ACT_TOPUP_RST_10", &reply); err != nil {
t.Error("Got error on APIerSv1.GetActions: ", err.Error())
} else if !reflect.DeepEqual(expectActs, reply) {
t.Errorf("Expected: %v,\n received: %v", utils.ToJSON(expectActs), utils.ToJSON(reply))
}
}
func testInternalRemoteITGetActionPlan(t *testing.T) {
var aps []*engine.ActionPlan
accIDsStrMp := utils.StringMap{
"cgrates.org:1001": true,
"cgrates.org:1002": true,
"cgrates.org:1003": true,
}
if err := internalRPC.Call(utils.APIerSv1GetActionPlan,
AttrGetActionPlan{ID: "AP_PACKAGE_10"}, &aps); err != nil {
t.Error(err)
} else if len(aps) != 1 {
t.Errorf("Expected: %v,\n received: %v", 1, len(aps))
} else if aps[0].Id != "AP_PACKAGE_10" {
t.Errorf("Expected: %v,\n received: %v", "AP_PACKAGE_10", aps[0].Id)
} else if !reflect.DeepEqual(aps[0].AccountIDs, accIDsStrMp) {
t.Errorf("Expected: %v,\n received: %v", accIDsStrMp, aps[0].AccountIDs)
}
if err := internalRPC.Call(utils.APIerSv1GetActionPlan,
AttrGetActionPlan{ID: utils.EmptyString}, &aps); err != nil {
t.Error(err)
} else if len(aps) != 1 {
t.Errorf("Expected: %v,\n received: %v", 1, len(aps))
} else if aps[0].Id != "AP_PACKAGE_10" {
t.Errorf("Expected: %v,\n received: %v", "AP_PACKAGE_10", aps[0].Id)
} else if !reflect.DeepEqual(aps[0].AccountIDs, accIDsStrMp) {
t.Errorf("Expected: %v,\n received: %v", accIDsStrMp, aps[0].AccountIDs)
}
}
func testInternalRemoteITGetAccountActionPlan(t *testing.T) {
var aap []*AccountActionTiming
if err := internalRPC.Call(utils.APIerSv1GetAccountActionPlan,
utils.TenantAccount{
Tenant: "cgrates.org",
Account: "1001",
}, &aap); err != nil {
t.Error(err)
} else if len(aap) != 1 {
t.Errorf("Expected: %v,\n received: %v", 1, len(aap))
} else if aap[0].ActionPlanId != "AP_PACKAGE_10" {
t.Errorf("Expected: %v,\n received: %v", "AP_PACKAGE_10", aap[0].ActionPlanId)
} else if aap[0].ActionsId != "ACT_TOPUP_RST_10" {
t.Errorf("Expected: %v,\n received: %v", "ACT_TOPUP_RST_10", aap[0].ActionsId)
}
}
func testInternalRemoteITGetDestination(t *testing.T) {
var dst *engine.Destination
eDst := &engine.Destination{
Id: "DST_1002",
Prefixes: []string{"1002"},
}
if err := internalRPC.Call(utils.APIerSv1GetDestination, "DST_1002", &dst); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(eDst, dst) {
t.Errorf("Expected: %v,\n received: %v", eDst, dst)
}
}
func testInternalRemoteITGetReverseDestination(t *testing.T) {
var ids []string
eIDs := []string{"DST_1002"}
if err := internalRPC.Call(utils.APIerSv1GetReverseDestination, "1002", &ids); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(eIDs, ids) {
t.Errorf("Expected: %v,\n received: %v", eIDs, ids)
}
}
func testInternalReplicationSetThreshold(t *testing.T) {
var reply *engine.ThresholdProfile
var result string
if err := internalRPC.Call(utils.APIerSv1GetThresholdProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "THD_Replication"}, &reply); err == nil ||
err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
//verify indexes on engine2 before adding new threshold profile
var indexes []string
expectedIDX := []string{"*string:~*req.Account:1001:THD_ACNT_1001",
"*string:~*req.Account:1002:THD_ACNT_1002"}
if err := engineTwoRPC.Call(utils.APIerSv1GetFilterIndexes, &AttrGetFilterIndexes{
ItemType: utils.MetaThresholds, Tenant: "cgrates.org", FilterType: utils.MetaString},
&indexes); err != nil {
t.Error(err)
}
sort.Strings(indexes)
if !reflect.DeepEqual(expectedIDX, indexes) {
t.Errorf("Expecting: %+v, received: %+v",
expectedIDX, utils.ToJSON(indexes))
}
//verify indexes on internal before adding new threshold profile
if err := internalRPC.Call(utils.APIerSv1GetFilterIndexes, &AttrGetFilterIndexes{
ItemType: utils.MetaThresholds, Tenant: "cgrates.org", FilterType: utils.MetaString},
&indexes); err != nil {
t.Error(err)
}
sort.Strings(indexes)
if !reflect.DeepEqual(expectedIDX, indexes) {
t.Errorf("Expecting: %+v, received: %+v",
expectedIDX, utils.ToJSON(indexes))
}
tPrfl := &engine.ThresholdWithCache{
ThresholdProfile: &engine.ThresholdProfile{
Tenant: "cgrates.org",
ID: "THD_Replication",
FilterIDs: []string{"*string:~*req.Account:1001", "*string:~*req.CustomField:CustomValue"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 35, 0, 0, time.UTC),
},
MaxHits: -1,
MinSleep: time.Duration(5 * time.Minute),
Blocker: false,
Weight: 20.0,
ActionIDs: []string{"ACT_LOG_WARNING"},
Async: true,
},
}
if err := internalRPC.Call(utils.APIerSv1SetThresholdProfile, tPrfl, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
time.Sleep(50 * time.Millisecond)
if err := internalRPC.Call(utils.APIerSv1GetThresholdProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "THD_Replication"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tPrfl.ThresholdProfile, reply) {
t.Errorf("Expecting: %+v, received: %+v", tPrfl.ThresholdProfile, reply)
}
expectedIDX = []string{
"*string:~*req.Account:1001:THD_ACNT_1001",
"*string:~*req.Account:1001:THD_Replication",
"*string:~*req.Account:1002:THD_ACNT_1002",
"*string:~*req.CustomField:CustomValue:THD_Replication",
}
// verify index on internal
sort.Strings(expectedIDX)
if err := internalRPC.Call(utils.APIerSv1GetFilterIndexes, &AttrGetFilterIndexes{
ItemType: utils.MetaThresholds, Tenant: "cgrates.org", FilterType: utils.MetaString},
&indexes); err != nil {
t.Error(err)
}
sort.Strings(indexes)
if !reflect.DeepEqual(expectedIDX, indexes) {
t.Errorf("Expecting: %+v, received: %+v",
expectedIDX, utils.ToJSON(indexes))
}
// verify data on engine1
if err := engineOneRPC.Call(utils.APIerSv1GetThresholdProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "THD_Replication"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tPrfl.ThresholdProfile, reply) {
t.Errorf("Expecting: %+v, received: %+v", tPrfl.ThresholdProfile, reply)
}
expectedIDX2 := []string{
"*string:~*req.Account:1001:THD_ACNT_1001",
"*string:~*req.Account:1001:THD_Replication",
"*string:~*req.CustomField:CustomValue:THD_Replication",
}
// verify indexes on engine1 (should be the same as internal)
if err := engineOneRPC.Call(utils.APIerSv1GetFilterIndexes, &AttrGetFilterIndexes{
ItemType: utils.MetaThresholds, Tenant: "cgrates.org", FilterType: utils.MetaString},
&indexes); err != nil {
t.Error(err)
}
sort.Strings(indexes)
if !reflect.DeepEqual(expectedIDX2, indexes) {
t.Errorf("Expecting: %+v, received: %+v",
expectedIDX2, utils.ToJSON(indexes))
}
// verify data on engine2
if err := engineTwoRPC.Call(utils.APIerSv1GetThresholdProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "THD_Replication"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tPrfl.ThresholdProfile, reply) {
t.Errorf("Expecting: %+v, received: %+v", tPrfl.ThresholdProfile, reply)
}
expectedIDX = []string{"*string:~*req.Account:1001:THD_ACNT_1001",
"*string:~*req.Account:1001:THD_Replication",
"*string:~*req.Account:1002:THD_ACNT_1002",
"*string:~*req.CustomField:CustomValue:THD_Replication",
}
// check if indexes was created correctly on engine2
if err := engineTwoRPC.Call(utils.APIerSv1GetFilterIndexes, &AttrGetFilterIndexes{
ItemType: utils.MetaThresholds, Tenant: "cgrates.org", FilterType: utils.MetaString},
&indexes); err != nil {
t.Error(err)
}
sort.Strings(indexes)
if !reflect.DeepEqual(expectedIDX, indexes) {
t.Errorf("Expecting: %+v,\n received: %+v",
expectedIDX, utils.ToJSON(indexes))
}
}
func testInternalMatchThreshold(t *testing.T) {
ev := &engine.ArgsProcessEvent{
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "event2",
Event: map[string]any{
utils.Account: "1002",
},
},
}
var ids []string
eIDs := []string{"THD_ACNT_1002"}
if err := internalRPC.Call(utils.ThresholdSv1ProcessEvent, ev, &ids); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(ids, eIDs) {
t.Errorf("Expecting ids: %s, received: %s", eIDs, ids)
}
ev = &engine.ArgsProcessEvent{
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "event2",
Event: map[string]any{
utils.Account: "1001",
},
},
}
eIDs = []string{"THD_ACNT_1001"}
if err := internalRPC.Call(utils.ThresholdSv1ProcessEvent, ev, &ids); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(ids, eIDs) {
t.Errorf("Expecting ids: %s, received: %s", eIDs, ids)
}
ev2 := &engine.ArgsProcessEvent{
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "event3",
Event: map[string]any{
utils.Account: "1001",
},
},
}
if err := internalRPC.Call(utils.ThresholdSv1ProcessEvent, ev2, &ids); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
}
func testInternalAccountBalanceOperations(t *testing.T) {
var reply string
attrs := &utils.AttrSetBalance{
Tenant: "cgrates.org",
Account: "testAccount1",
BalanceType: utils.MONETARY,
Value: 17.4,
Balance: map[string]any{
utils.ID: "testAccSetBalance",
},
}
if err := internalRPC.Call(utils.APIerSv1SetBalance, attrs, &reply); err != nil {
t.Error(err)
}
time.Sleep(100 * time.Millisecond)
var acnt *engine.Account
attrAcc := &utils.AttrGetAccount{
Tenant: "cgrates.org",
Account: "testAccount1",
}
// verify account on engineOne
if err := engineOneRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 17.4 {
t.Errorf("Expecting: %+v, received: %+v",
17.4, val)
}
// verify account on engineTwo
if err := engineTwoRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 17.4 {
t.Errorf("Expecting: %+v, received: %+v",
17.4, val)
}
// debit balance on internal and the account should be replicated to other engines
if err := internalRPC.Call(utils.APIerSv1DebitBalance, &AttrAddBalance{
Tenant: "cgrates.org",
Account: "testAccount1",
BalanceType: utils.MONETARY,
Value: 3.62,
}, &reply); err != nil {
t.Error(err)
} else if reply != utils.OK {
t.Errorf("Received: %s", reply)
}
time.Sleep(50 * time.Millisecond)
// verify debited account on engineOne
if err := engineOneRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 13.78 {
t.Errorf("Expecting: %+v, received: %+v",
13.78, val)
}
// verify debited account on engineTwo
if err := engineTwoRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 13.78 {
t.Errorf("Expecting: %+v, received: %+v",
13.78, val)
}
addBal := &AttrAddBalance{
Tenant: "cgrates.org",
Account: "testAccount1",
BalanceType: utils.MONETARY,
Value: 12.765,
}
// add balance for the account on internal and this should be replicated to other engines
if err := internalRPC.Call(utils.APIerSv1AddBalance, addBal, &reply); err != nil {
t.Error(err)
}
time.Sleep(50 * time.Millisecond)
// verify account on engineOne
if err := engineOneRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 26.545 {
t.Errorf("Expecting: %+v, received: %+v",
26.545, val)
}
// verify account on engineTwo
if err := engineTwoRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 26.545 {
t.Errorf("Expecting: %+v, received: %+v",
26.545, val)
}
}
func testInternalSetAccount(t *testing.T) {
var reply string
if err := internalRPC.Call(utils.APIerSv1SetAccount,
utils.AttrSetAccount{
Tenant: "cgrates.org",
Account: "testSetAccount",
ActionPlanID: "AP_PACKAGE_10",
ReloadScheduler: true,
}, &reply); err != nil {
t.Error(err)
}
// give some time to scheduler to execute the action
time.Sleep(100 * time.Millisecond)
var acnt *engine.Account
attrAcc := &utils.AttrGetAccount{
Tenant: "cgrates.org",
Account: "testSetAccount",
}
// verify account on engineOne
if err := engineOneRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 10 {
t.Errorf("Expecting: %+v, received: %+v",
10, val)
}
// verify account on engineTwo
if err := engineTwoRPC.Call(utils.APIerSv2GetAccount, attrAcc, &acnt); err != nil {
t.Error(err)
} else if len(acnt.BalanceMap[utils.MONETARY]) != 1 {
t.Errorf("Expecting: %+v, received: %+v",
1, len(acnt.BalanceMap[utils.MONETARY]))
} else if val := acnt.BalanceMap[utils.MONETARY].GetTotalValue(); val != 10 {
t.Errorf("Expecting: %+v, received: %+v",
10, val)
}
}
func testInternalReplicateStats(t *testing.T) {
var reply string
statConfig = &engine.StatQueueWithCache{
StatQueueProfile: &engine.StatQueueProfile{
Tenant: "cgrates.org",
ID: "StatsToReplicate",
FilterIDs: []string{"*string:~*req.Account:1001"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC),
},
QueueLength: 10,
TTL: time.Duration(10) * time.Second,
Metrics: []*engine.MetricWithFilters{
{
MetricID: utils.MetaACD,
},
{
MetricID: utils.MetaTCD,
},
},
ThresholdIDs: []string{"*none"},
Weight: 20,
MinItems: 1,
},
}
if err := internalRPC.Call(utils.APIerSv1SetStatQueueProfile, statConfig, &reply); err != nil {
t.Error(err)
} else if reply != utils.OK {
t.Error("Unexpected reply returned", reply)
}
time.Sleep(50 * time.Millisecond) // wait for the replication
var rcv *engine.StatQueueProfile
if err := engineOneRPC.Call(utils.APIerSv1GetStatQueueProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "StatsToReplicate"}, &rcv); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(statConfig.StatQueueProfile, rcv) {
t.Errorf("Expecting: %+v, received: %+v", utils.ToJSON(statConfig.StatQueueProfile), utils.ToJSON(rcv))
}
if err := engineTwoRPC.Call(utils.APIerSv1GetStatQueueProfile,
&utils.TenantID{Tenant: "cgrates.org", ID: "StatsToReplicate"}, &rcv); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(statConfig.StatQueueProfile, rcv) {
t.Errorf("Expecting: %+v, received: %+v", utils.ToJSON(statConfig.StatQueueProfile), utils.ToJSON(rcv))
}
}
func testInternalRemoteITKillEngine(t *testing.T) {
if err := engine.KillEngine(100); err != nil {
t.Error(err)
}
}