pkg/control/main_test.go

   1 /*
   2 ==================================================================================
   3   Copyright (c) 2019 AT&T Intellectual Property.
   4   Copyright (c) 2019 Nokia
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18 */
  19
  20 package control
  21
  22 import (
  23         "encoding/json"
  24         "fmt"
  25         "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_models"
  26         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  27         "io/ioutil"
  28         "net/http"
  29         "os"
  30         "strconv"
  31         "strings"
  32         "testing"
  33         "time"
  34 )
  35
  36 //-----------------------------------------------------------------------------
  37 //
  38 //-----------------------------------------------------------------------------
  39 type testingControl struct {
  40         desc     string
  41         syncChan chan struct{}
  42 }
  43
  44 func (tc *testingControl) ReadyCB(data interface{}) {
  45         xapp.Logger.Info("testingControl(%s) ReadyCB", tc.desc)
  46         tc.syncChan <- struct{}{}
  47         return
  48 }
  49
  50 func (tc *testingControl) WaitCB() {
  51         <-tc.syncChan
  52 }
  53
  54 func initTestingControl(desc string, rtfile string, port string) testingControl {
  55         tc := testingControl{}
  56         os.Setenv("RMR_SEED_RT", rtfile)
  57         os.Setenv("RMR_SRC_ID", "localhost:"+port)
  58         xapp.Logger.Info("Using rt file %s", os.Getenv("RMR_SEED_RT"))
  59         xapp.Logger.Info("Using src id  %s", os.Getenv("RMR_SRC_ID"))
  60         tc.desc = strings.ToUpper(desc)
  61         tc.syncChan = make(chan struct{})
  62         return tc
  63 }
  64
  65 //-----------------------------------------------------------------------------
  66 //
  67 //-----------------------------------------------------------------------------
  68 type testingRmrControl struct {
  69         testingControl
  70         rmrClientTest *xapp.RMRClient
  71 }
  72
  73 func (tc *testingRmrControl) RmrSend(params *RMRParams) (err error) {
  74         //
  75         //NOTE: Do this way until xapp-frame sending is improved
  76         //
  77         xapp.Logger.Info("(%s) RmrSend %s", tc.desc, params.String())
  78         status := false
  79         i := 1
  80         for ; i <= 10 && status == false; i++ {
  81                 status = tc.rmrClientTest.SendMsg(params.RMRParams)
  82                 if status == false {
  83                         xapp.Logger.Info("(%s) RmrSend failed. Retry count %v, %s", tc.desc, i, params.String())
  84                         time.Sleep(500 * time.Millisecond)
  85                 }
  86         }
  87         if status == false {
  88                 err = fmt.Errorf("(%s) RmrSend failed. Retry count %v, %s", tc.desc, i, params.String())
  89                 xapp.Rmr.Free(params.Mbuf)
  90         }
  91         return
  92 }
  93
  94 func initTestingRmrControl(desc string, rtfile string, port string, stat string, consumer xapp.MessageConsumer) testingRmrControl {
  95         tc := testingRmrControl{}
  96         tc.testingControl = initTestingControl(desc, rtfile, port)
  97         tc.rmrClientTest = xapp.NewRMRClientWithParams("tcp:"+port, 4096, 1, stat)
  98         tc.rmrClientTest.SetReadyCB(tc.ReadyCB, nil)
  99         go tc.rmrClientTest.Start(consumer)
 100         tc.WaitCB()
 101         return tc
 102 }
 103
 104 //-----------------------------------------------------------------------------
 105 //
 106 //-----------------------------------------------------------------------------
 107 type testingMessageChannel struct {
 108         rmrConChan chan *RMRParams
 109 }
 110
 111 func initTestingMessageChannel() testingMessageChannel {
 112         mc := testingMessageChannel{}
 113         mc.rmrConChan = make(chan *RMRParams)
 114         return mc
 115 }
 116
 117 //-----------------------------------------------------------------------------
 118 //
 119 //-----------------------------------------------------------------------------
 120 type xappTransaction struct {
 121         tc   *testingXappControl
 122         xid  string
 123         meid *xapp.RMRMeid
 124 }
 125
 126 type testingXappControl struct {
 127         testingRmrControl
 128         testingMessageChannel
 129         xid_seq uint64
 130 }
 131
 132 func (tc *testingXappControl) newXid() string {
 133         var xid string
 134         xid = tc.desc + "_XID_" + strconv.FormatUint(uint64(tc.xid_seq), 10)
 135         tc.xid_seq++
 136         return xid
 137 }
 138
 139 func (tc *testingXappControl) newXappTransaction(xid *string, ranname string) *xappTransaction {
 140         trans := &xappTransaction{}
 141         trans.tc = tc
 142         if xid == nil {
 143                 trans.xid = tc.newXid()
 144         } else {
 145                 trans.xid = *xid
 146         }
 147         trans.meid = &xapp.RMRMeid{RanName: ranname}
 148         return trans
 149 }
 150
 151 func (tc *testingXappControl) Consume(params *xapp.RMRParams) (err error) {
 152         xapp.Rmr.Free(params.Mbuf)
 153         params.Mbuf = nil
 154         msg := &RMRParams{params}
 155
 156         if strings.Contains(msg.Xid, tc.desc) {
 157                 xapp.Logger.Info("(%s) Consume %s", tc.desc, msg.String())
 158                 tc.rmrConChan <- msg
 159         } else {
 160                 xapp.Logger.Info("(%s) Ignore %s", tc.desc, msg.String())
 161         }
 162         return
 163 }
 164
 165 func createNewXappControl(desc string, rtfile string, port string, stat string) *testingXappControl {
 166         xappCtrl := &testingXappControl{}
 167         xappCtrl.testingRmrControl = initTestingRmrControl(desc, rtfile, port, stat, xappCtrl)
 168         xappCtrl.testingMessageChannel = initTestingMessageChannel()
 169         xappCtrl.xid_seq = 1
 170         return xappCtrl
 171 }
 172
 173 //-----------------------------------------------------------------------------
 174 //
 175 //-----------------------------------------------------------------------------
 176 type testingE2termControl struct {
 177         testingRmrControl
 178         testingMessageChannel
 179 }
 180
 181 func (tc *testingE2termControl) Consume(params *xapp.RMRParams) (err error) {
 182         xapp.Rmr.Free(params.Mbuf)
 183         params.Mbuf = nil
 184         msg := &RMRParams{params}
 185         xapp.Logger.Info("(%s) Consume %s", tc.desc, msg.String())
 186         tc.rmrConChan <- msg
 187         return
 188 }
 189
 190 func createNewE2termControl(desc string, rtfile string, port string, stat string) *testingE2termControl {
 191         e2termCtrl := &testingE2termControl{}
 192         e2termCtrl.testingRmrControl = initTestingRmrControl(desc, rtfile, port, stat, e2termCtrl)
 193         e2termCtrl.testingMessageChannel = initTestingMessageChannel()
 194         return e2termCtrl
 195 }
 196
 197 //-----------------------------------------------------------------------------
 198 //
 199 //-----------------------------------------------------------------------------
 200 type testingMainControl struct {
 201         testingControl
 202         c *Control
 203 }
 204
 205 func createNewMainControl(desc string, rtfile string, port string) *testingMainControl {
 206         mainCtrl = &testingMainControl{}
 207         mainCtrl.testingControl = initTestingControl(desc, rtfile, port)
 208         mainCtrl.c = NewControl()
 209         xapp.SetReadyCB(mainCtrl.ReadyCB, nil)
 210         go xapp.RunWithParams(mainCtrl.c, false)
 211         mainCtrl.WaitCB()
 212         return mainCtrl
 213 }
 214
 215 //-----------------------------------------------------------------------------
 216 //
 217 //-----------------------------------------------------------------------------
 218
 219 func testError(t *testing.T, pattern string, args ...interface{}) {
 220         xapp.Logger.Error(fmt.Sprintf(pattern, args...))
 221         t.Errorf(fmt.Sprintf(pattern, args...))
 222 }
 223
 224 func testLog(t *testing.T, pattern string, args ...interface{}) {
 225         xapp.Logger.Info(fmt.Sprintf(pattern, args...))
 226         t.Logf(fmt.Sprintf(pattern, args...))
 227 }
 228
 229 func testCreateTmpFile(str string) (string, error) {
 230         file, err := ioutil.TempFile("/tmp", "*.rt")
 231         if err != nil {
 232                 return "", err
 233         }
 234         _, err = file.WriteString(str)
 235         if err != nil {
 236                 file.Close()
 237                 return "", err
 238         }
 239         return file.Name(), nil
 240 }
 241
 242 //-----------------------------------------------------------------------------
 243 //
 244 //-----------------------------------------------------------------------------
 245
 246 var xappConn1 *testingXappControl
 247 var xappConn2 *testingXappControl
 248 var e2termConn *testingE2termControl
 249 var mainCtrl *testingMainControl
 250
 251 func TestMain(m *testing.M) {
 252         xapp.Logger.Info("TestMain start")
 253
 254         //---------------------------------
 255         //
 256         //---------------------------------
 257         http_handler := func(w http.ResponseWriter, r *http.Request) {
 258                 var req rtmgr_models.XappSubscriptionData
 259                 err := json.NewDecoder(r.Body).Decode(&req)
 260                 if err != nil {
 261                         xapp.Logger.Error("%s", err.Error())
 262                 }
 263                 xapp.Logger.Info("(http handler) handling Address=%s Port=%d SubscriptionID=%d", *req.Address, *req.Port, *req.SubscriptionID)
 264
 265                 w.WriteHeader(200)
 266         }
 267
 268         go func() {
 269                 http.HandleFunc("/", http_handler)
 270                 http.ListenAndServe("localhost:8989", nil)
 271         }()
 272
 273         //---------------------------------
 274         //
 275         //---------------------------------
 276
 277         //
 278         //Cfg creation won't work like this as xapp-frame reads it during init.
 279         //
 280         /*
 281             cfgstr:=`{
 282               "local": {
 283                   "host": ":8080"
 284               },
 285               "logger": {
 286                   "level": 4
 287               },
 288               "rmr": {
 289                  "protPort": "tcp:14560",
 290                  "maxSize": 4096,
 291                  "numWorkers": 1,
 292                  "txMessages": ["RIC_SUB_REQ", "RIC_SUB_DEL_REQ"],
 293                  "rxMessages": ["RIC_SUB_RESP", "RIC_SUB_FAILURE", "RIC_SUB_DEL_RESP", "RIC_SUB_DEL_FAILURE", "RIC_INDICATION"]
 294               },
 295               "db": {
 296                   "host": "localhost",
 297                   "port": 6379,
 298                   "namespaces": ["sdl", "rnib"]
 299               },
 300                  "rtmgr" : {
 301                    "HostAddr" : "localhost",
 302                    "port" : "8989",
 303                    "baseUrl" : "/"
 304                  }
 305            `
 306
 307            cfgfilename,_ := testCreateTmpFile(cfgstr)
 308            defer os.Remove(cfgfilename)
 309            os.Setenv("CFG_FILE", cfgfilename)
 310         */
 311         xapp.Logger.Info("Using cfg file %s", os.Getenv("CFG_FILE"))
 312
 313         //---------------------------------
 314         // Static routetable for rmr
 315         //
 316         // NOTE: Routing table is configured so, that responses
 317         //       are duplicated to xapp1 and xapp2 instances.
 318         //       If XID is not matching xapp stub will just
 319         //       drop message. (Messages 12011, 12012, 12021, 12022)
 320         //
 321         // 14560 submgr
 322         // 15560 e2term stub
 323         // 13560 xapp1 stub
 324         // 13660 xapp2 stub
 325         //
 326         //---------------------------------
 327
 328         allrt := `newrt|start
 329 mse|12010|-1|localhost:14560
 330 mse|12010,localhost:14560|-1|localhost:15560
 331 mse|12011,localhost:15560|-1|localhost:14560
 332 mse|12012,localhost:15560|-1|localhost:14560
 333 mse|12011,localhost:14560|-1|localhost:13660;localhost:13560
 334 mse|12012,localhost:14560|-1|localhost:13660;localhost:13560
 335 mse|12020|-1|localhost:14560
 336 mse|12020,localhost:14560|-1|localhost:15560
 337 mse|12021,localhost:15560|-1|localhost:14560
 338 mse|12022,localhost:15560|-1|localhost:14560
 339 mse|12021,localhost:14560|-1|localhost:13660;localhost:13560
 340 mse|12022,localhost:14560|-1|localhost:13660;localhost:13560
 341 newrt|end
 342 `
 343
 344         //---------------------------------
 345         //
 346         //---------------------------------
 347         xapp.Logger.Info("### submgr main run ###")
 348
 349         subsrt := allrt
 350         /*
 351                 subsrt := `newrt|start
 352            mse|12010|-1|localhost:14560
 353            mse|12010,localhost:14560|-1|localhost:15560
 354            mse|12011,localhost:15560|-1|localhost:14560
 355            mse|12011|-1|localhost:13560;localhost:13660
 356            mse|12012,localhost:15560|-1|localhost:14560
 357            mse|12012|-1|localhost:13560;localhost:13660
 358            mse|12020|-1|localhost:14560
 359            mse|12020,localhost:14560|-1|localhost:15560
 360            mse|12021,localhost:15560|-1|localhost:14560
 361            mse|12021|-1|localhost:13560;localhost:13660
 362            mse|12022,localhost:15560|-1|localhost:14560
 363            mse|12022|-1|localhost:13560;localhost:13660
 364            newrt|end
 365            `
 366         */
 367
 368         subrtfilename, _ := testCreateTmpFile(subsrt)
 369         defer os.Remove(subrtfilename)
 370         mainCtrl = createNewMainControl("main", subrtfilename, "14560")
 371
 372         //---------------------------------
 373         //
 374         //---------------------------------
 375         xapp.Logger.Info("### xapp1 rmr run ###")
 376
 377         xapprt1 := allrt
 378         /*
 379                 xapprt1 := `newrt|start
 380            mse|12010|-1|localhost:14560
 381            mse|12011|-1|localhost:13560
 382            mse|12012|-1|localhost:13560
 383            mse|12020|-1|localhost:14560
 384            mse|12021|-1|localhost:13560
 385            mse|12022|-1|localhost:13560
 386            newrt|end
 387            `
 388         */
 389
 390         xapprtfilename1, _ := testCreateTmpFile(xapprt1)
 391         defer os.Remove(xapprtfilename1)
 392         xappConn1 = createNewXappControl("xappstub1", xapprtfilename1, "13560", "RMRXAPP1STUB")
 393
 394         //---------------------------------
 395         //
 396         //---------------------------------
 397
 398         xapp.Logger.Info("### xapp2 rmr run ###")
 399
 400         xapprt2 := allrt
 401         /*
 402                 xapprt2 := `newrt|start
 403            mse|12010|-1|localhost:14560
 404            mse|12011|-1|localhost:13660
 405            mse|12012|-1|localhost:13660
 406            mse|12020|-1|localhost:14560
 407            mse|12021|-1|localhost:13660
 408            mse|12022|-1|localhost:13660
 409            newrt|end
 410            `
 411         */
 412
 413         xapprtfilename2, _ := testCreateTmpFile(xapprt2)
 414         defer os.Remove(xapprtfilename2)
 415         xappConn2 = createNewXappControl("xappstub2", xapprtfilename2, "13660", "RMRXAPP2STUB")
 416
 417         //---------------------------------
 418         //
 419         //---------------------------------
 420         xapp.Logger.Info("### e2term rmr run ###")
 421
 422         e2termrt := allrt
 423         /*
 424                 e2termrt := `newrt|start
 425            mse|12010|-1|localhost:15560
 426            mse|12011|-1|localhost:14560
 427            mse|12012|-1|localhost:14560
 428            mse|12020|-1|localhost:15560
 429            mse|12021|-1|localhost:14560
 430            mse|12022|-1|localhost:14560
 431            newrt|end
 432            `
 433         */
 434
 435         e2termrtfilename, _ := testCreateTmpFile(e2termrt)
 436         defer os.Remove(e2termrtfilename)
 437         e2termConn = createNewE2termControl("e2termstub", e2termrtfilename, "15560", "RMRE2TERMSTUB")
 438
 439         //---------------------------------
 440         // Stupid sleep to try improve robustness
 441         // due: http handler and rmr routes init delays
 442         //---------------------------------
 443         <-time.After(2 * time.Second)
 444
 445         //---------------------------------
 446         //
 447         //---------------------------------
 448         code := m.Run()
 449         os.Exit(code)
 450 }