/* 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 Affero 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 Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see */ package ers import ( "reflect" "testing" "time" "github.com/cgrates/cgrates/config" "github.com/cgrates/cgrates/engine" "github.com/cgrates/cgrates/utils" ) func TestAMQPv1ERProcessMessage(t *testing.T) { cfg := config.NewDefaultCGRConfig() rdr := &AMQPv1ER{ cgrCfg: cfg, cfgIdx: 0, fltrS: new(engine.FilterS), rdrEvents: make(chan *erEvent, 1), rdrExit: make(chan struct{}), rdrErr: make(chan error, 1), cap: nil, queueID: "cgrates_cdrs", } expEvent := &utils.CGREvent{ Tenant: "cgrates.org", Event: map[string]any{ utils.CGRID: "testCgrId", }, APIOpts: map[string]any{}, } body := []byte(`{"CGRID":"testCgrId"}`) rdr.Config().Fields = []*config.FCTemplate{ { Tag: "CGRID", Type: utils.MetaConstant, Value: config.NewRSRParsersMustCompile("testCgrId", utils.InfieldSep), Path: "*cgreq.CGRID", }, } rdr.Config().Fields[0].ComputePath() if err := rdr.processMessage(body); err != nil { t.Error(err) } select { case data := <-rdr.rdrEvents: expEvent.ID = data.cgrEvent.ID expEvent.Time = data.cgrEvent.Time if !reflect.DeepEqual(data.cgrEvent, expEvent) { t.Errorf("Expected %v but received %v", utils.ToJSON(expEvent), utils.ToJSON(data.cgrEvent)) } case <-time.After(50 * time.Millisecond): t.Error("Time limit exceeded") } } func TestAMQPv1ERProcessMessageError1(t *testing.T) { cfg := config.NewDefaultCGRConfig() rdr := &AMQPv1ER{ cgrCfg: cfg, cfgIdx: 0, fltrS: new(engine.FilterS), rdrEvents: make(chan *erEvent, 1), rdrExit: make(chan struct{}), rdrErr: make(chan error, 1), cap: nil, queueID: "cgrates_cdrs", } rdr.Config().Fields = []*config.FCTemplate{ {}, } body := []byte(`{"CGRID":"testCgrId"}`) errExpect := "unsupported type: <>" if err := rdr.processMessage(body); err == nil || err.Error() != errExpect { t.Errorf("Expected %v but received %v", errExpect, err) } } func TestAMQPv1ERProcessMessageError2(t *testing.T) { cfg := config.NewDefaultCGRConfig() data, err := engine.NewInternalDB(nil, nil, true, nil, cfg.DataDbCfg().Items) if err != nil { t.Error(err) } dm := engine.NewDataManager(data, cfg.CacheCfg(), nil) cfg.ERsCfg().Readers[0].ProcessedPath = "" fltrs := engine.NewFilterS(cfg, nil, dm) rdr := &AMQPv1ER{ cgrCfg: cfg, cfgIdx: 0, fltrS: fltrs, rdrEvents: make(chan *erEvent, 1), rdrExit: make(chan struct{}), rdrErr: make(chan error, 1), cap: nil, queueID: "cgrates_cdrs", } body := []byte(`{"CGRID":"testCgrId"}`) rdr.Config().Filters = []string{"Filter1"} errExpect := "NOT_FOUND:Filter1" if err := rdr.processMessage(body); err == nil || err.Error() != errExpect { t.Errorf("Expected %v but received %v", errExpect, err) } // rdr.Config().Filters = []string{"*exists:~*req..Account:"} if err := rdr.processMessage(body); err != nil { t.Error(err) } } func TestAMQPv1ERProcessMessageError3(t *testing.T) { cfg := config.NewDefaultCGRConfig() rdr := &AMQPv1ER{ cgrCfg: cfg, cfgIdx: 0, fltrS: new(engine.FilterS), rdrEvents: make(chan *erEvent, 1), rdrExit: make(chan struct{}), rdrErr: make(chan error, 1), cap: nil, queueID: "cgrates_cdrs", } body := []byte("invalid_format") errExpect := "invalid character 'i' looking for beginning of value" if err := rdr.processMessage(body); err == nil || err.Error() != errExpect { t.Errorf("Expected %v but received %v", errExpect, err) } }