uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-11 10:06:24 +05:00
Code Issues Packages Projects Releases Wiki Activity

Add/Update unit and integration tests adjusting for sessions backup

Browse Source
This commit is contained in:
arberkatellari
2024-06-11 17:12:57 +02:00
committed by Dan Christian Bogos
parent f356695f6f
commit a0f94c6804
18 changed files with 1474 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

627
sessions/sessions_bkup_it_test.go Normal file
View File

@@ -0,0 +1,627 @@
//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 sessions
import (
"fmt"
"path"
"testing"
"time"
"github.com/cgrates/birpc"
"github.com/cgrates/birpc/context"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
var (
sBkupCfgPath string
sBkupCfgDIR string
sBkupCfg *config.CGRConfig
sBkupRPC *birpc.Client
dataDB engine.DataDB
updatedAt time.Time
SessionsBkupTests = []func(t *testing.T){
testSessionSBkupInitCfg,
testSessionSBkupResetDB,
testSessionSBkupStartEngine,
testSessionSBkupApierRpcConn,
testSessionSBkupTPFromFolder,
testSessionSBkupInitiate,
testSessionSBkupStopCgrEngine,
testSessionSBkupStartEngine,
testSessionSBkupApierRpcConn,
testSessionSBkupCheckRestored1,
testSessionSBkupTerminate1,
testSessionSBkupStopCgrEngine,
testSessionSBkupStartEngine,
testSessionSBkupApierRpcConn,
testSessionSBkupCheckRestored2,
testSessionSBkupTerminate2,
testSessionSBkupStopCgrEngine,
testSessionSBkupStartEngine,
testSessionSBkupApierRpcConn,
testSessionSBkupCheckRestored3,
testSessionSBkupStopCgrEngine,
testSessionSBkupStartEngine,
testSessionSBkupApierRpcConn,
testSessionSBkupCallBackup1,
testSessionSBkupInitiate,
testSessionSBkupCallBackup2,
testSessionSBkupGetBackedupSessions,
testSessionSBkupUpdateTerminate,
testSessionSBkupCallBackup3,
testSessionSBkupCheckUpdatedAt,
testSessionSBkupStopCgrEngine,
testSessionSBkupStartEngine,
testSessionSBkupApierRpcConn,
testSessionSBkupCheckUpdatedNotExpired,
testSessionSBkupStopCgrEngine,
testSessionSBkupStartEngine,
testSessionSBkupApierRpcConn,
testSessionSBkupCheckUpdatedExpired,
testSessionSBkupStopCgrEngine,
}
)
func TestSessionsBkup(t *testing.T) {
switch *utils.DBType {
case utils.MetaInternal:
t.SkipNow()
case utils.MetaMySQL:
sBkupCfgDIR = "sessions_backup_mysql"
case utils.MetaMongo:
sBkupCfgDIR = "sessions_backup_mongo"
case utils.MetaPostgres:
sBkupCfgDIR = "sessions_backup_postgres"
default:
t.Fatal("Unknown Database type")
}
for _, stest := range SessionsBkupTests {
t.Run(*utils.DBType, stest)
}
}
func testSessionSBkupInitCfg(t *testing.T) {
sBkupCfgPath = path.Join(*utils.DataDir, "conf", "samples", sBkupCfgDIR)
if sBkupCfg, err = config.NewCGRConfigFromPath(sBkupCfgPath); err != nil {
t.Fatal(err)
}
}
// Remove data in both rating and accounting db
func testSessionSBkupResetDB(t *testing.T) {
if err := engine.InitDataDb(sBkupCfg); err != nil {
t.Fatal(err)
}
if err := engine.InitStorDb(sBkupCfg); err != nil {
t.Fatal(err)
}
}
// Start CGR Engine
func testSessionSBkupStartEngine(t *testing.T) {
if _, err := engine.StartEngine(sBkupCfgPath, *utils.WaitRater); err != nil {
t.Fatal(err)
}
}
// Connect rpc client to rater
func testSessionSBkupApierRpcConn(t *testing.T) {
if sBkupRPC, err = newRPCClient(sBkupCfg.ListenCfg()); err != nil {
t.Fatal(err)
}
}
// Load the tariff plan, creating accounts and their balances
func testSessionSBkupTPFromFolder(t *testing.T) {
attrs := &utils.AttrLoadTpFromFolder{FolderPath: path.Join(*utils.DataDir, "tariffplans", "oldtutorial")}
var loadInst utils.LoadInstance
if err := sBkupRPC.Call(context.Background(), utils.APIerSv2LoadTariffPlanFromFolder, attrs, &loadInst); err != nil {
t.Error(err)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Give time for scheduler to execute topups
}
func testSessionSBkupInitiate(t *testing.T) {
var aSessions []*ExternalSession
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
usage := time.Duration(90 * time.Second)
argsInit := &V1InitSessionArgs{
InitSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestBkupSessionExpiresAfter3s",
Event: map[string]any{
utils.EventName: "TEST_EVENT1",
utils.Tenant: "cgrates.org",
utils.OriginID: "123450",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2024, time.March, 7, 16, 60, 0, 0, time.UTC),
utils.AnswerTime: time.Date(2024, time.March, 7, 16, 60, 10, 0, time.UTC),
utils.Usage: usage,
},
},
}
var initRpl V1InitSessionReply
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1InitiateSession,
argsInit, &initRpl); err != nil {
t.Error(err)
}
//compare the value
if initRpl.MaxUsage == nil || *initRpl.MaxUsage != usage {
t.Errorf("Expecting : %+v, received: %+v", usage, initRpl.MaxUsage)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Wait for the sessions to be populated
// Delay further 4 seconds to make the session unrestorable (session expiries in 4s in configs)
time.Sleep(4 * time.Second)
//check if the session was createad as active session
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 1 {
t.Errorf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
} else if aSessions[0].Usage != usage {
t.Errorf("Expecting : %+v, received: %+v", usage, aSessions[0].Usage)
}
usage = time.Duration(120 * time.Second)
argsInit = &V1InitSessionArgs{
InitSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestBkupSession2",
Event: map[string]any{
utils.EventName: "TEST_EVENT2",
utils.Tenant: "cgrates.org",
utils.OriginID: "123452",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2024, time.March, 7, 16, 60, 2, 0, time.UTC),
utils.AnswerTime: time.Date(2024, time.March, 7, 16, 60, 12, 0, time.UTC),
utils.Usage: usage,
},
},
}
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1InitiateSession,
argsInit, &initRpl); err != nil {
t.Error(err)
}
//compare the value
if initRpl.MaxUsage == nil || *initRpl.MaxUsage != usage {
t.Errorf("Expecting : %+v, received: %+v", usage, initRpl.MaxUsage)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Wait for the sessions to be populated
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 2 {
t.Errorf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
}
for _, extSess := range aSessions {
if extSess.OriginID == "123452" {
if extSess.Usage != usage {
t.Errorf("OriginID: <123452>, Expecting : %+v, received: %+v", usage, extSess.Usage)
}
}
}
usage = time.Duration(150 * time.Second)
argsInit = &V1InitSessionArgs{
InitSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestBkupSession1",
Event: map[string]any{
utils.EventName: "TEST_EVENT1",
utils.Tenant: "cgrates.org",
utils.OriginID: "123451",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2024, time.March, 7, 16, 60, 3, 0, time.UTC),
utils.AnswerTime: time.Date(2024, time.March, 7, 16, 60, 13, 0, time.UTC),
utils.Usage: usage,
},
},
}
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1InitiateSession,
argsInit, &initRpl); err != nil {
t.Error(err)
}
//compare the value
if initRpl.MaxUsage == nil || *initRpl.MaxUsage != usage {
t.Errorf("Expecting : %+v, received: %+v", usage, initRpl.MaxUsage)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Wait for the sessions to be populated
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 3 {
t.Errorf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
}
for _, extSess := range aSessions {
if extSess.OriginID == "123451" {
if extSess.Usage != usage {
t.Errorf("OriginID: <123451>, Expecting : %+v, received: %+v", usage, extSess.Usage)
}
}
}
}
func testSessionSBkupCheckRestored1(t *testing.T) {
var aSessions []*ExternalSession
usage1 := time.Duration(120 * time.Second)
usage2 := time.Duration(150 * time.Second)
//check if the sessions were restored
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 2 {
t.Fatalf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
}
for _, extSess := range aSessions {
switch {
case extSess.OriginID == "123451":
if extSess.Usage != usage2 {
t.Errorf("OriginID: <123451>, Expecting : %+v, received: %+v", usage2, extSess.Usage)
}
case extSess.OriginID == "123452":
if extSess.Usage != usage1 {
t.Errorf("OriginID: <123452>, Expecting : %+v, received: %+v", usage1, extSess.Usage)
}
default:
t.Fatalf("Unexpected OriginID <%v> present in the session", extSess.OriginID)
}
}
}
func testSessionSBkupTerminate1(t *testing.T) {
var replyTerminate string
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1TerminateSession,
&V1TerminateSessionArgs{
TerminateSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestBkupSession1",
Event: map[string]any{
utils.EventName: "TEST_EVENT1",
utils.Tenant: "cgrates.org",
utils.OriginID: "123451",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2024, time.March, 7, 16, 60, 3, 0, time.UTC),
utils.AnswerTime: time.Date(2024, time.March, 7, 16, 60, 13, 0, time.UTC),
},
},
}, &replyTerminate); err != nil {
t.Error(err)
}
if replyTerminate != utils.OK {
t.Errorf("Expected reply <OK>, received <%+v>", replyTerminate)
}
}
func testSessionSBkupCheckRestored2(t *testing.T) {
var aSessions []*ExternalSession
usage1 := time.Duration(120 * time.Second)
//check if the sessions were restored
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err != nil {
t.Fatal(err)
} else if len(aSessions) != 1 {
t.Fatalf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
}
if aSessions[0].OriginID == "123452" {
if aSessions[0].Usage != usage1 {
t.Errorf("OriginID: <123452>, Expecting : %+v, received: %+v", usage1, aSessions[0].Usage)
}
}
}
func testSessionSBkupTerminate2(t *testing.T) {
var replyTerminate string
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1TerminateSession,
&V1TerminateSessionArgs{
TerminateSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestBkupSession2",
Event: map[string]any{
utils.EventName: "TEST_EVENT2",
utils.Tenant: "cgrates.org",
utils.OriginID: "123452",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2024, time.March, 7, 16, 60, 2, 0, time.UTC),
utils.AnswerTime: time.Date(2024, time.March, 7, 16, 60, 12, 0, time.UTC),
},
},
}, &replyTerminate); err != nil {
t.Error(err)
}
if replyTerminate != utils.OK {
t.Errorf("Expected reply <OK>, received <%+v>", replyTerminate)
}
}
func testSessionSBkupCheckRestored3(t *testing.T) {
var aSessions []*ExternalSession
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
if len(aSessions) != 0 {
t.Errorf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
}
}
func testSessionSBkupCallBackup1(t *testing.T) {
var sessionsBackedup int
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1BackupActiveSessions,
utils.EmptyString, &sessionsBackedup); err != nil {
t.Error(err)
} else if sessionsBackedup != 0 {
t.Errorf("Expected 0 backedup sessions. Backed up: %+v", sessionsBackedup)
}
}
func testSessionSBkupCallBackup2(t *testing.T) {
var sessionsBackedup int
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1BackupActiveSessions,
utils.EmptyString, &sessionsBackedup); err != nil {
t.Fatal(err)
} else if sessionsBackedup != 3 {
t.Errorf("Expected 3 backedup sessions. Backed up: %+v", sessionsBackedup)
}
}
func testSessionSBkupGetBackedupSessions(t *testing.T) {
if *utils.DBType == utils.MetaMySQL || *utils.DBType == utils.MetaPostgres {
dataDB, err = engine.NewRedisStorage(
fmt.Sprintf("%s:%s", sBkupCfg.DataDbCfg().Host, sBkupCfg.DataDbCfg().Port),
10, sBkupCfg.DataDbCfg().User, sBkupCfg.DataDbCfg().Password, sBkupCfg.GeneralCfg().DBDataEncoding,
10, 20, "", false, 5*time.Second, 0, 0, 0, 0, false, utils.EmptyString, utils.EmptyString, utils.EmptyString)
if err != nil {
t.Fatal("Could not connect to Redis", err.Error())
}
}
if *utils.DBType == utils.MetaMongo {
dataDB, err = engine.NewMongoStorage("mongodb", sBkupCfg.DataDbCfg().Host,
sBkupCfg.DataDbCfg().Port, sBkupCfg.DataDbCfg().Name,
sBkupCfg.DataDbCfg().User, sBkupCfg.DataDbCfg().Password,
sBkupCfg.GeneralCfg().DBDataEncoding,
utils.StorDB, nil, 10*time.Second)
if err != nil {
t.Fatal(err)
}
}
var getBackSess []*Session
storedSessions, err := dataDB.GetSessionsBackupDrv(sBkupCfg.GeneralCfg().NodeID,
sBkupCfg.GeneralCfg().DefaultTenant)
if err != nil {
t.Error(err)
}
for _, storSess := range storedSessions {
sess := newSessionFromStoredSession(storSess)
getBackSess = append(getBackSess, sess)
}
if len(getBackSess) != 3 { // even though one of them expired, we are not restoring them currently so in db there should be 3 sessions stored
t.Fatalf("Expected 3 sessions stored, received %v", len(getBackSess))
}
for _, oneSess := range getBackSess {
switch {
case oneSess.ResourceID == "123450":
if oneSess.totalUsage() != time.Duration(90*time.Second) {
t.Errorf("Expected <%v>, received <%v>", time.Duration(90*time.Second), oneSess.totalUsage())
}
case oneSess.ResourceID == "123451":
if oneSess.totalUsage() != time.Duration(150*time.Second) {
t.Errorf("Expected <%v>, received <%v>", time.Duration(150*time.Second), oneSess.totalUsage())
}
case oneSess.ResourceID == "123452":
if oneSess.totalUsage() != time.Duration(120*time.Second) {
t.Errorf("Expected <%v>, received <%v>", time.Duration(120*time.Second), oneSess.totalUsage())
}
updatedAt = oneSess.UpdatedAt
}
}
}
func testSessionSBkupUpdateTerminate(t *testing.T) {
updatedUsage := time.Duration(200 * time.Second)
var upRply V1UpdateSessionReply
upArgs := &V1UpdateSessionArgs{
UpdateSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestBkupSession2",
Event: map[string]any{
utils.EventName: "TEST_EVENT2",
utils.Tenant: "cgrates.org",
utils.OriginID: "123452",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2024, time.March, 7, 16, 60, 2, 0, time.UTC),
utils.AnswerTime: time.Date(2024, time.March, 7, 16, 60, 12, 0, time.UTC),
utils.Usage: updatedUsage,
},
},
}
if err = sBkupRPC.Call(context.Background(), utils.SessionSv1UpdateSession, upArgs, &upRply); err != nil {
t.Error(err)
} else if *upRply.MaxUsage != updatedUsage {
t.Errorf("Expected <%+v>, Received <%+v>", updatedUsage, *upRply.MaxUsage)
}
args := &V1TerminateSessionArgs{
TerminateSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestBkupSession1",
Event: map[string]any{
utils.EventName: "TEST_EVENT1",
utils.Tenant: "cgrates.org",
utils.OriginID: "123451",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2024, time.March, 7, 16, 60, 3, 0, time.UTC),
utils.AnswerTime: time.Date(2024, time.March, 7, 16, 60, 13, 0, time.UTC),
utils.Usage: time.Duration(160 * time.Second),
},
},
}
var rply string
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1TerminateSession,
args, &rply); err != nil {
t.Error(err)
}
if rply != utils.OK {
t.Errorf("Unexpected reply: %s", rply)
}
}
func testSessionSBkupCallBackup3(t *testing.T) {
var sessionsBackedup int
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1BackupActiveSessions,
utils.EmptyString, &sessionsBackedup); err != nil {
t.Fatal(err)
} else if sessionsBackedup != 2 { // should've backed up 2 since we terminated one of them
t.Errorf("Expected 2 backedup sessions. Backed up: %+v", sessionsBackedup)
}
}
func testSessionSBkupCheckUpdatedAt(t *testing.T) {
var getBackSess []*Session
storedSessions, err := dataDB.GetSessionsBackupDrv(sBkupCfg.GeneralCfg().NodeID,
sBkupCfg.GeneralCfg().DefaultTenant)
if err != nil {
t.Error(err)
}
for _, storSess := range storedSessions {
sess := newSessionFromStoredSession(storSess)
getBackSess = append(getBackSess, sess)
}
if len(getBackSess) != 2 { // even though one of them expired, we are not restoring them currently so in db there should be 2 sessions stored
t.Fatalf("Expected 2 sessions stored, received %v", len(getBackSess))
}
for _, oneSess := range getBackSess {
switch {
case oneSess.ResourceID == "123450":
if oneSess.totalUsage() != time.Duration(90*time.Second) {
t.Errorf("Expected <%v>, received <%v>", time.Duration(90*time.Second), oneSess.totalUsage())
}
case oneSess.ResourceID == "123452":
if oneSess.totalUsage() != time.Duration(320*time.Second) { // usage should be updated to 320 seconds
t.Errorf("Expected <%v>, received <%v>", time.Duration(320*time.Second), oneSess.totalUsage())
}
if oneSess.UpdatedAt == updatedAt {
t.Errorf("Expected UpdatedAt field to be changed on update. Received the same time as before <%+v>", oneSess.UpdatedAt)
}
}
}
}
func testSessionSBkupCheckUpdatedNotExpired(t *testing.T) {
var aSessions []*ExternalSession
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 1 {
t.Fatalf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
}
if aSessions[0].OriginID != "123452" {
t.Errorf("Expected backed up session source 123452, received <%+v>", aSessions[0].OriginID)
}
time.Sleep(4 * time.Second) // Wait for updated session to expire
}
func testSessionSBkupCheckUpdatedExpired(t *testing.T) {
var aSessions []*ExternalSession
if err := sBkupRPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
} else if len(aSessions) != 0 {
t.Fatalf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
}
}
func testSessionSBkupStopCgrEngine(t *testing.T) {
if err := engine.KillEngine(1000); err != nil {
t.Error(err)
}
}

401
sessions/sessions_bkup_rpl_it_test.go Normal file
View File

@@ -0,0 +1,401 @@
//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 sessions
import (
"fmt"
"os/exec"
"path"
"testing"
"time"
"github.com/cgrates/birpc"
"github.com/cgrates/birpc/context"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
var (
sBRplcEng1CfgPath, sBRplcEng2CfgPath string
sBRplcEng1CfgDIR, sBRplcEng2CfgDIR string
sBRplcEng1Cfg, sBRplcEng2Cfg *config.CGRConfig
sBRplcEng1RPC, sBRplcEng2RPC *birpc.Client
sBRplcEng1Eng, sBRplcEng2Eng *exec.Cmd
SessionsBkupRplTests = []func(t *testing.T){
testSessionSBkupRplcInitCfg,
testSessionSBkupRplcResetDB,
testSessionSBkupRplcStartEngineBoth,
testSessionSBkupRplcApierRpcConnBoth,
testSessionSBkupRplcTPFromFolder,
testSessionSBkupRplcInitiate,
testSessionSBkupRplcGetSessions,
testSessionSBkupRplcUpdate,
testSessionSBkupRplcStartEngine2,
testSessionSBkupRplcApierRpcConn2,
testSessionSBkupRplcGetActvSessionsFromRestored,
testSessionSBkupRplcTerminate,
testSessionSBkupRplcStopCgrEngine,
testSessionSBkupRplcStartEngine2,
testSessionSBkupRplcApierRpcConn2,
testSessionSBkupRplcGetNoActvSessionsFromRestored,
testSessionSBkupRplcStopCgrEngine,
}
)
func TestSessionSBkupRplc(t *testing.T) {
switch *utils.DBType {
case utils.MetaInternal:
t.SkipNow()
case utils.MetaMySQL:
sBRplcEng1CfgDIR = "sbkupreplcengine1_mysql"
sBRplcEng2CfgDIR = "sbkupreplcengine2_mongo"
case utils.MetaMongo:
sBRplcEng1CfgDIR = "sbkupreplcengine1_mongo"
sBRplcEng2CfgDIR = "sbkupreplcengine2_mysql"
case utils.MetaPostgres:
sBRplcEng1CfgDIR = "sbkupreplcengine1_postgres"
sBRplcEng2CfgDIR = "sbkupreplcengine2_mongo"
default:
t.Fatal("Unknown Database type")
}
if *utils.Encoding == utils.MetaGOB {
sBRplcEng1CfgDIR += "_gob"
sBRplcEng2CfgDIR += "_gob"
}
for _, stest := range SessionsBkupRplTests {
t.Run(*utils.DBType, stest)
}
}
func testSessionSBkupRplcInitCfg(t *testing.T) {
sBRplcEng1CfgPath = path.Join(*utils.DataDir, "conf", "samples", sBRplcEng1CfgDIR)
if sBRplcEng1Cfg, err = config.NewCGRConfigFromPath(sBRplcEng1CfgPath); err != nil {
t.Fatal(err)
}
sBRplcEng2CfgPath = path.Join(*utils.DataDir, "conf", "samples", sBRplcEng2CfgDIR)
if sBRplcEng2Cfg, err = config.NewCGRConfigFromPath(sBRplcEng2CfgPath); err != nil {
t.Fatal(err)
}
}
// Remove data in both rating and accounting db
func testSessionSBkupRplcResetDB(t *testing.T) {
if err := engine.InitDataDb(sBRplcEng1Cfg); err != nil {
t.Fatal(err)
}
if err := engine.InitStorDb(sBRplcEng1Cfg); err != nil {
t.Fatal(err)
}
if err := engine.InitDataDb(sBRplcEng2Cfg); err != nil {
t.Fatal(err)
}
if err := engine.InitStorDb(sBRplcEng2Cfg); err != nil {
t.Fatal(err)
}
}
// Start CGR Engine
func testSessionSBkupRplcStartEngineBoth(t *testing.T) {
if sBRplcEng2Eng, err = engine.StopStartEngine(sBRplcEng2CfgPath, *utils.WaitRater); err != nil { // Start engine2 before engine1
t.Fatal(err)
}
if sBRplcEng1Eng, err = engine.StartEngine(sBRplcEng1CfgPath, *utils.WaitRater); err != nil {
t.Fatal(err)
}
}
// Connect rpc client to rater
func testSessionSBkupRplcApierRpcConnBoth(t *testing.T) {
if sBRplcEng1RPC, err = newRPCClient(sBRplcEng1Cfg.ListenCfg()); err != nil {
t.Fatal(err)
}
if sBRplcEng2RPC, err = newRPCClient(sBRplcEng2Cfg.ListenCfg()); err != nil {
t.Fatal(err)
}
}
// Load the tariff plan, creating accounts and their balances
func testSessionSBkupRplcTPFromFolder(t *testing.T) {
attrs := &utils.AttrLoadTpFromFolder{FolderPath: path.Join(*utils.DataDir, "tariffplans", "oldtutorial")}
var loadInst utils.LoadInstance
if err := sBRplcEng1RPC.Call(context.Background(), utils.APIerSv2LoadTariffPlanFromFolder, attrs, &loadInst); err != nil {
t.Error(err)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Give time for scheduler to execute topups
attrs = &utils.AttrLoadTpFromFolder{FolderPath: path.Join(*utils.DataDir, "tariffplans", "oldtutorial")}
if err := sBRplcEng2RPC.Call(context.Background(), utils.APIerSv2LoadTariffPlanFromFolder, attrs, &loadInst); err != nil {
t.Error(err)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Give time for scheduler to execute topups
}
func testSessionSBkupRplcInitiate(t *testing.T) {
var aSessions []*ExternalSession
//make sure we don't have active sessions on engine1 and passive on engine2
if err := sBRplcEng1RPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
usage := time.Minute + 30*time.Second
argsInit := &V1InitSessionArgs{
InitSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestSessionSBkupRplcInitiate",
Event: map[string]any{
utils.EventName: "TEST_EVENT",
utils.Tenant: "cgrates.org",
utils.OriginID: "123451",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2018, time.January, 7, 16, 60, 0, 0, time.UTC),
utils.AnswerTime: time.Date(2018, time.January, 7, 16, 60, 10, 0, time.UTC),
utils.Usage: usage,
},
},
}
var initRpl V1InitSessionReply
if err := sBRplcEng1RPC.Call(context.Background(), utils.SessionSv1InitiateSession,
argsInit, &initRpl); err != nil {
t.Error(err)
}
//compare the value
if initRpl.MaxUsage == nil || *initRpl.MaxUsage != usage {
t.Errorf("Expecting : %+v, received: %+v", usage, initRpl.MaxUsage)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Wait for the sessions to be populated
//check if the session was createad as active session on engine1
if err := sBRplcEng1RPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
utils.SessionFilter{
Filters: []string{
fmt.Sprintf("*string:~*req.%s:%s", utils.OriginID, "123451"),
},
}, &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 1 {
t.Errorf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
} else if aSessions[0].Usage != 90*time.Second {
t.Errorf("Expecting : %+v, received: %+v", 90*time.Second, aSessions[0].Usage)
}
//check if the session was created as passive session on engine2
var pSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions,
utils.SessionFilter{
Filters: []string{
fmt.Sprintf("*string:~*req.%s:%s", utils.OriginID, "123451"),
},
}, &pSessions); err != nil {
t.Error(err)
} else if len(pSessions) != 1 {
t.Errorf("PassiveSessions: %+v", pSessions)
} else if pSessions[0].Usage != 90*time.Second {
t.Errorf("Expecting : %+v, received: %+v", 90*time.Second, pSessions[0].Usage)
}
}
func testSessionSBkupRplcGetSessions(t *testing.T) {
time.Sleep(501 * time.Millisecond) // make sure active sessions are backed up from "backup_interval"
if err := sBRplcEng1Eng.Process.Kill(); err != nil {
t.Errorf("Failed to kill process, error: %v", err.Error())
}
// make sure we have no active sessions on engine2
var aSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
// make sure we have passive sessions on engine2
var pSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions,
utils.SessionFilter{}, &pSessions); err != nil {
t.Error(err)
} else if len(pSessions) != 1 {
t.Errorf("PassiveSessions: %+v", pSessions)
} else if pSessions[0].Usage != 90*time.Second {
t.Errorf("Expecting : %+v, received: %+v", 90*time.Second, pSessions[0].Usage)
}
}
func testSessionSBkupRplcUpdate(t *testing.T) {
//update the session on engine2 so the session should became active
usage := time.Minute
argsUpdate := &V1UpdateSessionArgs{
UpdateSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestSessionSBkupRplcUpdate",
Event: map[string]any{
utils.EventName: "TEST_EVENT",
utils.Tenant: "cgrates.org",
utils.OriginID: "123451",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2018, time.January, 7, 16, 60, 0, 0, time.UTC),
utils.AnswerTime: time.Date(2018, time.January, 7, 16, 60, 10, 0, time.UTC),
utils.Usage: usage,
},
},
}
var updtRpl V1UpdateSessionReply
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1UpdateSession,
argsUpdate, &updtRpl); err != nil {
t.Error(err)
}
if updtRpl.MaxUsage == nil || *updtRpl.MaxUsage != usage {
t.Errorf("Expecting : %+v, received: %+v", usage, updtRpl.MaxUsage)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Wait for the sessions to be populated
var aSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
utils.SessionFilter{
Filters: []string{
fmt.Sprintf("*string:~*req.%s:%s", utils.OriginID, "123451"),
},
}, &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 1 {
t.Errorf("Unexpected number of sessions received: %+v", aSessions)
} else if aSessions[0].Usage != 150*time.Second {
t.Errorf("Expecting : %+v, received: %+v", 150*time.Second, aSessions[0].Usage)
}
var pSessions []*ExternalSession
// Make sure we don't have passive session on active host
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions, new(utils.SessionFilter),
&pSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
time.Sleep(501 * time.Millisecond) // make sure active sessions are backed up from "backup_interval"
if err := sBRplcEng2Eng.Process.Kill(); err != nil {
t.Errorf("Failed to kill process, error: %v", err.Error())
}
}
func testSessionSBkupRplcStartEngine2(t *testing.T) {
if sBRplcEng2Eng, err = engine.StopStartEngine(sBRplcEng2CfgPath, *utils.WaitRater); err != nil {
t.Fatal(err)
}
}
func testSessionSBkupRplcApierRpcConn2(t *testing.T) {
if sBRplcEng2RPC, err = newRPCClient(sBRplcEng2Cfg.ListenCfg()); err != nil {
t.Fatal(err)
}
}
func testSessionSBkupRplcGetActvSessionsFromRestored(t *testing.T) {
var aSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
utils.SessionFilter{}, &aSessions); err != nil {
t.Error(err)
} else if len(aSessions) != 1 {
t.Errorf("Unexpected number of sessions received: %+v", utils.ToIJSON(aSessions))
} else if aSessions[0].Usage != 150*time.Second {
t.Errorf("Expecting : %+v, received: %+v", 150*time.Second, aSessions[0].Usage)
}
var pSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions,
new(utils.SessionFilter), &pSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
}
func testSessionSBkupRplcTerminate(t *testing.T) {
args := &V1TerminateSessionArgs{
TerminateSession: true,
CGREvent: &utils.CGREvent{
Tenant: "cgrates.org",
ID: "TestSessionSBkupRplcTerminate",
Event: map[string]any{
utils.EventName: "TEST_EVENT",
utils.Tenant: "cgrates.org",
utils.OriginID: "123451",
utils.ToR: utils.MetaVoice,
utils.RequestType: utils.MetaPrepaid,
utils.AccountField: "1001",
utils.Subject: "1001",
utils.Destination: "1004",
utils.Category: "call",
utils.SetupTime: time.Date(2018, time.January, 7, 16, 60, 0, 0, time.UTC),
utils.AnswerTime: time.Date(2018, time.January, 7, 16, 60, 10, 0, time.UTC),
utils.Usage: 2*time.Minute + 30*time.Second,
},
},
}
var reply string
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1TerminateSession, args, &reply); err != nil {
t.Error(err)
}
time.Sleep(time.Duration(*utils.WaitRater) * time.Millisecond) // Wait for the sessions to be populated
var aSessions []*ExternalSession
//check if the session was terminated on engine2
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
new(utils.SessionFilter), &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Errorf("Error: %v with len(aSessions)=%v , session : %+v", err, len(aSessions), utils.ToIJSON(aSessions))
}
// make sure we don't have passive session on engine2
var pSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions, new(utils.SessionFilter),
&pSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Errorf("Error: %v with len(pSessions)=%v , session : %+v", err, len(pSessions), utils.ToIJSON(pSessions))
}
}
func testSessionSBkupRplcGetNoActvSessionsFromRestored(t *testing.T) {
var aSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetActiveSessions,
utils.SessionFilter{}, &aSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
var pSessions []*ExternalSession
if err := sBRplcEng2RPC.Call(context.Background(), utils.SessionSv1GetPassiveSessions,
new(utils.SessionFilter), &pSessions); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
}
func testSessionSBkupRplcStopCgrEngine(t *testing.T) {
if err := engine.KillEngine(1000); err != nil {
t.Error(err)
}
}

29
sessions/sessions_test.go
View File

@@ -1938,21 +1938,23 @@ func (*testRPCClientConnection) Call(string, any, any) error { return nil }
func TestNewSessionS(t *testing.T) {
cgrCGF := config.NewDefaultCGRConfig()
eOut := &SessionS{
cgrCfg: cgrCGF,
dm: nil,
biJClnts: make(map[birpc.ClientConnector]string),
biJIDs: make(map[string]*biJClient),
aSessions: make(map[string]*Session),
aSessionsIdx: make(map[string]map[string]map[string]utils.StringSet),
aSessionsRIdx: make(map[string][]*riFieldNameVal),
pSessions: make(map[string]*Session),
pSessionsIdx: make(map[string]map[string]map[string]utils.StringSet),
pSessionsRIdx: make(map[string][]*riFieldNameVal),
}
sS := NewSessionS(cgrCGF, nil, nil)
eOut := &SessionS{
cgrCfg: cgrCGF,
dm: nil,
biJClnts: make(map[birpc.ClientConnector]string),
biJIDs: make(map[string]*biJClient),
aSessions: make(map[string]*Session),
aSessionsIdx: make(map[string]map[string]map[string]utils.StringSet),
aSessionsRIdx: make(map[string][]*riFieldNameVal),
pSessions: make(map[string]*Session),
pSessionsIdx: make(map[string]map[string]map[string]utils.StringSet),
pSessionsRIdx: make(map[string][]*riFieldNameVal),
markedSsCGRIDs: make(utils.StringSet),
removeSsCGRIDs: make(utils.StringSet),
}
if !reflect.DeepEqual(sS, eOut) {
t.Errorf("Expected %s , received: %s", utils.ToJSON(sS), utils.ToJSON(eOut))
t.Errorf("Expected <%+v> , \nreceived: <%+v>", sS, eOut)
}
}
@@ -2663,6 +2665,7 @@ func TestInitSession(t *testing.T) {
},
DebitInterval: 0,
Chargeable: true,
UpdatedAt: s.UpdatedAt,
}
s.SRuns = nil
if !reflect.DeepEqual(exp, s) {

23
sessions/sessionscover_test.go
View File

@@ -1545,11 +1545,12 @@ func TestTransitSState(t *testing.T) {
CGRID: "TEST_CGRID",
},
}
expected := &Session{
CGRID: "TEST_CGRID",
}
rcv = sessions.getActivateSession("test")
expected := &Session{
CGRID: "TEST_CGRID",
UpdatedAt: rcv.UpdatedAt,
}
if !reflect.DeepEqual(rcv, expected) {
t.Errorf("Expected %+v \n, received %+v", utils.ToJSON(expected), utils.ToJSON(rcv))
}
@@ -1579,6 +1580,7 @@ func TestRelocateSession(t *testing.T) {
},
},
}
rcv := sessions.relocateSession("111", "222", "127.0.0.1")
expected := &Session{
CGRID: "dfa2adaa5ab49349777c1ab3bcf3455df0259880",
EventStart: map[string]any{
@@ -1593,10 +1595,9 @@ func TestRelocateSession(t *testing.T) {
},
},
},
UpdatedAt: rcv.UpdatedAt,
}
if rcv := sessions.relocateSession("111", "222", "127.0.0.1"); rcv == nil {
t.Errorf("Expected to not be nil")
} else if !reflect.DeepEqual(rcv, expected) {
if !reflect.DeepEqual(rcv, expected) {
t.Errorf("Expected %+v \n, received %+v", utils.ToJSON(expected), utils.ToJSON(rcv))
}
@@ -1604,7 +1605,7 @@ func TestRelocateSession(t *testing.T) {
"0d0fe8779b54c88f121e26c5d83abee5935127e5": nil,
}
rcv := sessions.relocateSession("111", "222", utils.EmptyString)
rcv = sessions.relocateSession("111", "222", utils.EmptyString)
if rcv != nil {
t.Errorf("Expected to be nil")
}
@@ -1629,12 +1630,12 @@ func TestGetRelocateSession(t *testing.T) {
},
}
rcv = sessions.getRelocateSession("test", utils.EmptyString, "222", "127.0.0.1")
expected := &Session{
CGRID: "TEST_CGRID",
CGRID: "TEST_CGRID",
UpdatedAt: rcv.UpdatedAt,
}
if rcv = sessions.getRelocateSession("test", utils.EmptyString, "222", "127.0.0.1"); rcv == nil {
t.Errorf("Expected to be nil")
} else if !reflect.DeepEqual(rcv, expected) {
if !reflect.DeepEqual(rcv, expected) {
t.Errorf("Expected %+v \n, received %+v", utils.ToJSON(expected), utils.ToJSON(rcv))
}
}