Updating radius_coa integration test to use thresholds for generating SessionSv1.ReAuthorize

agents/radagent.go

@@ -116,7 +116,8 @@ type radiusDAClientCfg struct {
 }
 
 // newRadiusDAClientCfg is a constructor for the radiusDAClientCfg type.
-func newRadiusDAClientCfg(dicts *radigo.Dictionaries, secrets *radigo.Secrets, radAgentCfg *config.RadiusAgentCfg) radiusDAClientCfg {
+func newRadiusDAClientCfg(dicts *radigo.Dictionaries, secrets *radigo.Secrets,
+	radAgentCfg *config.RadiusAgentCfg) radiusDAClientCfg {
 	dacDicts := make(map[string]*radigo.Dictionary, len(radAgentCfg.ClientDaAddresses))
 	dacSecrets := make(map[string]string, len(radAgentCfg.ClientDaAddresses))
 	for client := range radAgentCfg.ClientDaAddresses {
@@ -229,7 +230,8 @@ func (ra *RadiusAgent) handleAcct(req *radigo.Packet) (rpl *radigo.Packet, err e
 	rplyNM := utils.NewOrderedNavigableMap()
 	opts := utils.MapStorage{}
 	var processed bool
-	reqVars := &utils.DataNode{Type: utils.NMMapType, Map: map[string]*utils.DataNode{utils.RemoteHost: utils.NewLeafNode(req.RemoteAddr().String())}}
+	reqVars := &utils.DataNode{Type: utils.NMMapType,
+		Map: map[string]*utils.DataNode{utils.RemoteHost: utils.NewLeafNode(req.RemoteAddr().String())}}
 	for _, reqProcessor := range ra.cgrCfg.RadiusAgentCfg().RequestProcessors {
 		agReq := NewAgentRequest(dcdr, reqVars, cgrRplyNM, rplyNM, opts,
 			reqProcessor.Tenant, ra.cgrCfg.GeneralCfg().DefaultTenant,
@@ -384,7 +386,8 @@ func (ra *RadiusAgent) processRequest(req *radigo.Packet, reqProcessor *config.R
 			reqProcessor.Flags.ParamValue(utils.MetaRoutesMaxCost),
 		)
 		rply := new(sessions.V1ProcessMessageReply)
-		err = ra.connMgr.Call(ra.ctx, ra.cgrCfg.RadiusAgentCfg().SessionSConns, utils.SessionSv1ProcessMessage, evArgs, rply)
+		err = ra.connMgr.Call(ra.ctx, ra.cgrCfg.RadiusAgentCfg().SessionSConns,
+			utils.SessionSv1ProcessMessage, evArgs, rply)
 		if utils.ErrHasPrefix(err, utils.RalsErrorPrfx) {
 			cgrEv.Event[utils.Usage] = 0 // avoid further debits
 		} else if evArgs.Debit {
@@ -399,8 +402,8 @@ func (ra *RadiusAgent) processRequest(req *radigo.Packet, reqProcessor *config.R
 			Paginator: cgrArgs,
 		}
 		rply := new(sessions.V1ProcessEventReply)
-		err = ra.connMgr.Call(ra.ctx, ra.cgrCfg.RadiusAgentCfg().SessionSConns, utils.SessionSv1ProcessEvent,
-			evArgs, rply)
+		err = ra.connMgr.Call(ra.ctx, ra.cgrCfg.RadiusAgentCfg().SessionSConns,
+			utils.SessionSv1ProcessEvent, evArgs, rply)
 		if utils.ErrHasPrefix(err, utils.RalsErrorPrfx) {
 			cgrEv.Event[utils.Usage] = 0 // avoid further debits
 		} else if needsMaxUsage(reqProcessor.Flags[utils.MetaRALs]) {
@@ -418,8 +421,8 @@ func (ra *RadiusAgent) processRequest(req *radigo.Packet, reqProcessor *config.R
 	// separate request so we can capture the Terminate/Event also here
 	if reqProcessor.Flags.GetBool(utils.MetaCDRs) {
 		var rplyCDRs string
-		if err = ra.connMgr.Call(ra.ctx, ra.cgrCfg.RadiusAgentCfg().SessionSConns, utils.SessionSv1ProcessCDR,
-			cgrEv, &rplyCDRs); err != nil {
+		if err = ra.connMgr.Call(ra.ctx, ra.cgrCfg.RadiusAgentCfg().SessionSConns,
+			utils.SessionSv1ProcessCDR, cgrEv, &rplyCDRs); err != nil {
 			agReq.CGRReply.Map[utils.Error] = utils.NewLeafNode(err.Error())
 		}
 	}
@@ -549,7 +552,8 @@ func (ra *RadiusAgent) V1ReAuthorize(_ *context.Context, cgrEv utils.CGREvent, r
 }
 
 // sendRadDaReq prepares and sends a Radius CoA/Disconnect Request and returns the reply code or an error.
-func (ra *RadiusAgent) sendRadDaReq(requestType radigo.PacketCode, requestTemplate, sessionID string, requestEv utils.DataProvider, requestVars *utils.DataNode) (radigo.PacketCode, error) {
+func (ra *RadiusAgent) sendRadDaReq(requestType radigo.PacketCode, requestTemplate, sessionID string,
+	requestEv utils.DataProvider, requestVars *utils.DataNode) (radigo.PacketCode, error) {
 	cachedPacket, has := engine.Cache.Get(utils.CacheRadiusPackets, sessionID)
 	if !has {
 		return 0, fmt.Errorf("failed to retrieve packet from cache: %w", utils.ErrNotFound)
@@ -568,7 +572,8 @@ func (ra *RadiusAgent) sendRadDaReq(requestType radigo.PacketCode, requestTempla
 		return 0, fmt.Errorf("could not set attributes: %w", err)
 	}
 
-	remoteAddr, remoteHost, err := dmRemoteAddr(packet.RemoteAddr().String(), ra.cgrCfg.RadiusAgentCfg().ClientDaAddresses)
+	remoteAddr, remoteHost, err := dmRemoteAddr(packet.RemoteAddr().String(),
+		ra.cgrCfg.RadiusAgentCfg().ClientDaAddresses)
 	if err != nil {
 		return 0, fmt.Errorf("retrieving remote address failed: %w", err)
 	}

agents/radius_coa_it_test.go

@@ -94,6 +94,52 @@ func TestRadiusCoADisconnect(t *testing.T) {
 	}
 	time.Sleep(time.Duration(*waitRater) * time.Millisecond)
 
+	var reply string
+
+	// Set Action which will be called by Threshold when account gets debitted
+	actRadCoaAcnt1001 := &utils.AttrSetActions{
+		ActionsId: "ACT_RAD_COA_ACNT_1001",
+		Actions: []*utils.TPAction{{
+			Identifier: utils.MetaCgrRpc,
+			ExtraParameters: `{
+		                    "Address":"localhost:2012",
+		                    "Attempts":1,
+		                    "Transport":"*json",
+		                    "Method":"SessionSv1.ReAuthorize",
+		                    "Params":{
+		                        "Filters":["*string:~*req.Account:1001"],
+		                        "Tenant":"cgrates.org",
+		                        "APIOpts":{"*radCoATemplate":"mycoa"},
+		                        "Event":{"CustomFilter":"custom_filter"}},
+		                    "Id":2}`,
+		}}}
+	if err := raDiscRPC.Call(context.Background(), utils.APIerSv2SetActions,
+		actRadCoaAcnt1001, &reply); err != nil {
+		t.Error("Got error on APIerSv2.SetActions: ", err.Error())
+	} else if reply != utils.OK {
+		t.Errorf("Calling APIerSv2.SetActions received: %s", reply)
+	}
+
+	// Set the Threshold profile which will call the action when account will be modified
+
+	tPrfl := &engine.ThresholdProfileWithAPIOpts{
+		ThresholdProfile: &engine.ThresholdProfile{
+			Tenant:    "cgrates.org",
+			ID:        "THD_ACNT_1001",
+			FilterIDs: []string{"*string:~*opts.*eventType:AccountUpdate", "*string:~*req.ID:1001"},
+			//MinHits:   1,
+			MaxHits:   1,
+			ActionIDs: []string{"LOG_WARNING", "ACT_RAD_COA_ACNT_1001"},
+			Async:     true,
+		},
+	}
+	if err := raDiscRPC.Call(context.Background(), utils.APIerSv1SetThresholdProfile,
+		tPrfl, &reply); err != nil {
+		t.Error(err)
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply returned", reply)
+	}
+
 	// Testing the functionality itself starts here.
 	var wg sync.WaitGroup
 	done := make(chan struct{}) // signal to end the test when the handlers have finished processing
@@ -231,22 +277,23 @@ func TestRadiusCoADisconnect(t *testing.T) {
 	if replyPacket.Code != radigo.AccountingResponse {
 		t.Errorf("unexpected reply received to AccountingRequest: %+v", replyPacket)
 	}
-
-	var reply string
-	if err := raDiscRPC.Call(context.Background(), utils.SessionSv1ReAuthorize,
-		utils.SessionFilterWithEvent{
-			SessionFilter: &utils.SessionFilter{
-				APIOpts: map[string]any{
-					utils.MetaRadCoATemplate: "mycoa",
-				}},
-			Event: map[string]any{
-				"CustomFilter": "custom_filter",
-			},
-		}, &reply); err != nil {
-		t.Error(err)
-	}
-
-	if err = raDiscRPC.Call(context.Background(), utils.SessionSv1ForceDisconnect, nil, &reply); err != nil {
+	/*
+		if err := raDiscRPC.Call(context.Background(), utils.SessionSv1ReAuthorize,
+			utils.SessionFilterWithEvent{
+				SessionFilter: &utils.SessionFilter{
+					APIOpts: map[string]any{
+						utils.MetaRadCoATemplate: "mycoa",
+					}},
+				Event: map[string]any{
+					"CustomFilter": "custom_filter",
+				},
+			}, &reply); err != nil {
+			t.Error(err)
+		}
+	*/
+	time.Sleep(1 * time.Second)
+	if err = raDiscRPC.Call(context.Background(), utils.SessionSv1ForceDisconnect,
+		nil, &reply); err != nil {
 		t.Error(err)
 	}
 

data/conf/samples/radius_coa_disconnect/cgrates.json

@@ -13,7 +13,8 @@
 },
 
 "rals": {
-	"enabled": true
+	"enabled": true,
+	"thresholds_conns": ["*localhost"],
 },
 
 "schedulers": {
@@ -38,6 +39,11 @@
 	"enabled": true
 },
 
+"thresholds": {
+	"enabled": true,
+	"store_interval": "-1"
+},
+
 "sessions": {
 	"enabled": true,
 	"attributes_conns": ["*localhost"],

data/tariffplans/oldtutorial/Actions.csv

@@ -10,3 +10,4 @@ LOG_WARNING,*log,,,,,,,,,,,,,false,false,10
 DISABLE_AND_LOG,*log,,,,,,,,,,,,,false,false,10
 DISABLE_AND_LOG,*disable_account,,,,,,,,,,,,,false,false,10
 TOPUP_100SMS_DE_MOBILE,*topup,,,,*sms,,DST_DE_MOBILE,,,,,100,10,false,false,10
+#ACT_RAD_COA_ACNT_1001,*cgr_rpc,"{""Address"":""localhost:2012"",""Transport"":""*json"",""Method"":""SessionSv1.ReAuthorize"",""Attempts"":1,""Async"":false,""Params"":{""Filters"":[""*string:~*req.Account:1001""],""Tenant"":""cgrates.org"",""APIOpts"":{""*radCoATemplate"":""mycoa""},""Event"":{""CustomFilter"":""custom_filter""}}}",,,,,,,,,,,,,,20