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         "sync"
  33         "testing"
  34         "time"
  35 )
  36
  37 //-----------------------------------------------------------------------------
  38 //
  39 //-----------------------------------------------------------------------------
  40
  41 type httpEventWaiter struct {
  42         resultChan   chan bool
  43         nextActionOk bool
  44 }
  45
  46 func (msg *httpEventWaiter) SetResult(res bool) {
  47         msg.resultChan <- res
  48 }
  49
  50 func (msg *httpEventWaiter) WaitResult(t *testing.T) bool {
  51         select {
  52         case result := <-msg.resultChan:
  53                 return result
  54         case <-time.After(15 * time.Second):
  55                 testError(t, "Waiter not received result status from case within 15 secs")
  56                 return false
  57         }
  58         testError(t, "Waiter error in default branch")
  59         return false
  60 }
  61
  62 //-----------------------------------------------------------------------------
  63 //
  64 //-----------------------------------------------------------------------------
  65 type testingHttpRtmgrControl struct {
  66         sync.Mutex
  67         desc        string
  68         port        string
  69         eventWaiter *httpEventWaiter
  70 }
  71
  72 func (hc *testingHttpRtmgrControl) NextEvent(eventWaiter *httpEventWaiter) {
  73         hc.Lock()
  74         defer hc.Unlock()
  75         hc.eventWaiter = eventWaiter
  76 }
  77
  78 func (hc *testingHttpRtmgrControl) AllocNextEvent(nextAction bool) *httpEventWaiter {
  79         eventWaiter := &httpEventWaiter{
  80                 resultChan:   make(chan bool),
  81                 nextActionOk: nextAction,
  82         }
  83         hc.NextEvent(eventWaiter)
  84         return eventWaiter
  85 }
  86
  87 func (hc *testingHttpRtmgrControl) http_handler(w http.ResponseWriter, r *http.Request) {
  88
  89         hc.Lock()
  90         defer hc.Unlock()
  91
  92         var req rtmgr_models.XappSubscriptionData
  93         err := json.NewDecoder(r.Body).Decode(&req)
  94         if err != nil {
  95                 xapp.Logger.Error("%s", err.Error())
  96         }
  97         xapp.Logger.Info("(%s) handling Address=%s Port=%d SubscriptionID=%d", hc.desc, *req.Address, *req.Port, *req.SubscriptionID)
  98
  99         var code int = 0
 100         switch r.Method {
 101         case http.MethodPost:
 102                 code = 201
 103                 if hc.eventWaiter != nil {
 104                         if hc.eventWaiter.nextActionOk == false {
 105                                 code = 400
 106                         }
 107                 }
 108         case http.MethodDelete:
 109                 code = 200
 110                 if hc.eventWaiter != nil {
 111                         if hc.eventWaiter.nextActionOk == false {
 112                                 code = 400
 113                         }
 114                 }
 115         default:
 116                 code = 200
 117         }
 118
 119         waiter := hc.eventWaiter
 120         hc.eventWaiter = nil
 121         if waiter != nil {
 122                 waiter.SetResult(true)
 123         }
 124         xapp.Logger.Info("(%s) Method=%s Reply with code %d", hc.desc, r.Method, code)
 125         w.WriteHeader(code)
 126
 127 }
 128
 129 func (hc *testingHttpRtmgrControl) run() {
 130         http.HandleFunc("/", hc.http_handler)
 131         http.ListenAndServe("localhost:"+hc.port, nil)
 132 }
 133
 134 func initTestingHttpRtmgrControl(desc string, port string) *testingHttpRtmgrControl {
 135         hc := &testingHttpRtmgrControl{}
 136         hc.desc = desc
 137         hc.port = port
 138         return hc
 139 }
 140
 141 //-----------------------------------------------------------------------------
 142 //
 143 //-----------------------------------------------------------------------------
 144 type testingRmrControl struct {
 145         desc     string
 146         syncChan chan struct{}
 147 }
 148
 149 func (tc *testingRmrControl) ReadyCB(data interface{}) {
 150         xapp.Logger.Info("testingRmrControl(%s) ReadyCB", tc.desc)
 151         tc.syncChan <- struct{}{}
 152         return
 153 }
 154
 155 func (tc *testingRmrControl) WaitCB() {
 156         <-tc.syncChan
 157 }
 158
 159 func initTestingControl(desc string, rtfile string, port string) testingRmrControl {
 160         tc := testingRmrControl{}
 161         os.Setenv("RMR_SEED_RT", rtfile)
 162         os.Setenv("RMR_SRC_ID", "localhost:"+port)
 163         xapp.Logger.Info("Using rt file %s", os.Getenv("RMR_SEED_RT"))
 164         xapp.Logger.Info("Using src id  %s", os.Getenv("RMR_SRC_ID"))
 165         tc.desc = strings.ToUpper(desc)
 166         tc.syncChan = make(chan struct{})
 167         return tc
 168 }
 169
 170 //-----------------------------------------------------------------------------
 171 //
 172 //-----------------------------------------------------------------------------
 173 type testingRmrStubControl struct {
 174         testingRmrControl
 175         rmrClientTest *xapp.RMRClient
 176         active        bool
 177         msgCnt        uint64
 178 }
 179
 180 func (tc *testingRmrStubControl) GetMsgCnt() uint64 {
 181         return tc.msgCnt
 182 }
 183
 184 func (tc *testingRmrStubControl) IncMsgCnt() {
 185         tc.msgCnt++
 186 }
 187
 188 func (tc *testingRmrStubControl) DecMsgCnt() {
 189         if tc.msgCnt > 0 {
 190                 tc.msgCnt--
 191         }
 192 }
 193
 194 func (tc *testingRmrStubControl) TestMsgCnt(t *testing.T) {
 195         if tc.GetMsgCnt() > 0 {
 196                 testError(t, "(%s) message count expected 0 but is %d", tc.desc, tc.GetMsgCnt())
 197         }
 198 }
 199
 200 func (tc *testingRmrStubControl) RmrSend(params *RMRParams) (err error) {
 201         //
 202         //NOTE: Do this way until xapp-frame sending is improved
 203         //
 204         xapp.Logger.Info("(%s) RmrSend %s", tc.desc, params.String())
 205         status := false
 206         i := 1
 207         for ; i <= 10 && status == false; i++ {
 208                 status = tc.rmrClientTest.SendMsg(params.RMRParams)
 209                 if status == false {
 210                         xapp.Logger.Info("(%s) RmrSend failed. Retry count %v, %s", tc.desc, i, params.String())
 211                         time.Sleep(500 * time.Millisecond)
 212                 }
 213         }
 214         if status == false {
 215                 err = fmt.Errorf("(%s) RmrSend failed. Retry count %v, %s", tc.desc, i, params.String())
 216                 xapp.Rmr.Free(params.Mbuf)
 217         }
 218         return
 219 }
 220
 221 func initTestingRmrControl(desc string, rtfile string, port string, stat string, consumer xapp.MessageConsumer) testingRmrStubControl {
 222         tc := testingRmrStubControl{}
 223         tc.active = false
 224         tc.testingRmrControl = initTestingControl(desc, rtfile, port)
 225         tc.rmrClientTest = xapp.NewRMRClientWithParams("tcp:"+port, 4096, 1, stat)
 226         tc.rmrClientTest.SetReadyCB(tc.ReadyCB, nil)
 227         go tc.rmrClientTest.Start(consumer)
 228         tc.WaitCB()
 229         return tc
 230 }
 231
 232 //-----------------------------------------------------------------------------
 233 //
 234 //-----------------------------------------------------------------------------
 235 type testingMessageChannel struct {
 236         rmrConChan chan *RMRParams
 237 }
 238
 239 func initTestingMessageChannel() testingMessageChannel {
 240         mc := testingMessageChannel{}
 241         mc.rmrConChan = make(chan *RMRParams)
 242         return mc
 243 }
 244
 245 //-----------------------------------------------------------------------------
 246 //
 247 //-----------------------------------------------------------------------------
 248 type xappTransaction struct {
 249         tc   *testingXappControl
 250         xid  string
 251         meid *xapp.RMRMeid
 252 }
 253
 254 type testingXappControl struct {
 255         testingRmrStubControl
 256         testingMessageChannel
 257         xid_seq uint64
 258 }
 259
 260 func (tc *testingXappControl) newXid() string {
 261         var xid string
 262         xid = tc.desc + "_XID_" + strconv.FormatUint(uint64(tc.xid_seq), 10)
 263         tc.xid_seq++
 264         return xid
 265 }
 266
 267 func (tc *testingXappControl) newXappTransaction(xid *string, ranname string) *xappTransaction {
 268         trans := &xappTransaction{}
 269         trans.tc = tc
 270         if xid == nil {
 271                 trans.xid = tc.newXid()
 272         } else {
 273                 trans.xid = *xid
 274         }
 275         trans.meid = &xapp.RMRMeid{RanName: ranname}
 276         return trans
 277 }
 278
 279 func (tc *testingXappControl) Consume(params *xapp.RMRParams) (err error) {
 280         xapp.Rmr.Free(params.Mbuf)
 281         params.Mbuf = nil
 282         msg := &RMRParams{params}
 283
 284         if params.Mtype == 55555 {
 285                 xapp.Logger.Info("(%s) Testing message ignore %s", tc.desc, msg.String())
 286                 tc.active = true
 287                 return
 288         }
 289
 290         if strings.Contains(msg.Xid, tc.desc) {
 291                 xapp.Logger.Info("(%s) Consume %s", tc.desc, msg.String())
 292                 tc.IncMsgCnt()
 293                 tc.rmrConChan <- msg
 294         } else {
 295                 xapp.Logger.Info("(%s) Ignore %s", tc.desc, msg.String())
 296         }
 297         return
 298 }
 299
 300 func createNewXappControl(desc string, rtfile string, port string, stat string) *testingXappControl {
 301         xappCtrl := &testingXappControl{}
 302         xappCtrl.testingRmrStubControl = initTestingRmrControl(desc, rtfile, port, stat, xappCtrl)
 303         xappCtrl.testingMessageChannel = initTestingMessageChannel()
 304         xappCtrl.xid_seq = 1
 305         return xappCtrl
 306 }
 307
 308 //-----------------------------------------------------------------------------
 309 //
 310 //-----------------------------------------------------------------------------
 311 type testingE2termControl struct {
 312         testingRmrStubControl
 313         testingMessageChannel
 314 }
 315
 316 func (tc *testingE2termControl) Consume(params *xapp.RMRParams) (err error) {
 317         xapp.Rmr.Free(params.Mbuf)
 318         params.Mbuf = nil
 319         msg := &RMRParams{params}
 320
 321         if params.Mtype == 55555 {
 322                 xapp.Logger.Info("(%s) Testing message ignore %s", tc.desc, msg.String())
 323                 tc.active = true
 324                 return
 325         }
 326
 327         xapp.Logger.Info("(%s) Consume %s", tc.desc, msg.String())
 328         tc.rmrConChan <- msg
 329         return
 330 }
 331
 332 func createNewE2termControl(desc string, rtfile string, port string, stat string) *testingE2termControl {
 333         e2termCtrl := &testingE2termControl{}
 334         e2termCtrl.testingRmrStubControl = initTestingRmrControl(desc, rtfile, port, stat, e2termCtrl)
 335         e2termCtrl.testingMessageChannel = initTestingMessageChannel()
 336         return e2termCtrl
 337 }
 338
 339 //-----------------------------------------------------------------------------
 340 //
 341 //-----------------------------------------------------------------------------
 342 type testingMainControl struct {
 343         testingRmrControl
 344         c *Control
 345 }
 346
 347 func createNewMainControl(desc string, rtfile string, port string) *testingMainControl {
 348         mainCtrl = &testingMainControl{}
 349         mainCtrl.testingRmrControl = initTestingControl(desc, rtfile, port)
 350         mainCtrl.c = NewControl()
 351         xapp.SetReadyCB(mainCtrl.ReadyCB, nil)
 352         go xapp.RunWithParams(mainCtrl.c, false)
 353         mainCtrl.WaitCB()
 354         return mainCtrl
 355 }
 356
 357 //-----------------------------------------------------------------------------
 358 //
 359 //-----------------------------------------------------------------------------
 360
 361 func testError(t *testing.T, pattern string, args ...interface{}) {
 362         xapp.Logger.Error(fmt.Sprintf(pattern, args...))
 363         t.Errorf(fmt.Sprintf(pattern, args...))
 364 }
 365
 366 func testLog(t *testing.T, pattern string, args ...interface{}) {
 367         xapp.Logger.Info(fmt.Sprintf(pattern, args...))
 368         t.Logf(fmt.Sprintf(pattern, args...))
 369 }
 370
 371 func testCreateTmpFile(str string) (string, error) {
 372         file, err := ioutil.TempFile("/tmp", "*.rt")
 373         if err != nil {
 374                 return "", err
 375         }
 376         _, err = file.WriteString(str)
 377         if err != nil {
 378                 file.Close()
 379                 return "", err
 380         }
 381         return file.Name(), nil
 382 }
 383
 384 //-----------------------------------------------------------------------------
 385 //
 386 //-----------------------------------------------------------------------------
 387
 388 var xappConn1 *testingXappControl
 389 var xappConn2 *testingXappControl
 390 var e2termConn *testingE2termControl
 391 var mainCtrl *testingMainControl
 392 var rtmgrHttp *testingHttpRtmgrControl
 393
 394 func TestMain(m *testing.M) {
 395         xapp.Logger.Info("TestMain start")
 396
 397         //---------------------------------
 398         //
 399         //---------------------------------
 400         rtmgrHttp = initTestingHttpRtmgrControl("RTMGRSTUB", "8989")
 401         go rtmgrHttp.run()
 402
 403         //---------------------------------
 404         //
 405         //---------------------------------
 406
 407         //
 408         //Cfg creation won't work like this as xapp-frame reads it during init.
 409         //
 410         /*
 411             cfgstr:=`{
 412               "local": {
 413                   "host": ":8080"
 414               },
 415               "logger": {
 416                   "level": 4
 417               },
 418               "rmr": {
 419                  "protPort": "tcp:14560",
 420                  "maxSize": 4096,
 421                  "numWorkers": 1,
 422                  "txMessages": ["RIC_SUB_REQ", "RIC_SUB_DEL_REQ"],
 423                  "rxMessages": ["RIC_SUB_RESP", "RIC_SUB_FAILURE", "RIC_SUB_DEL_RESP", "RIC_SUB_DEL_FAILURE", "RIC_INDICATION"]
 424               },
 425               "db": {
 426                   "host": "localhost",
 427                   "port": 6379,
 428                   "namespaces": ["sdl", "rnib"]
 429               },
 430                  "rtmgr" : {
 431                    "HostAddr" : "localhost",
 432                    "port" : "8989",
 433                    "baseUrl" : "/"
 434                  }
 435            `
 436
 437            cfgfilename,_ := testCreateTmpFile(cfgstr)
 438            defer os.Remove(cfgfilename)
 439            os.Setenv("CFG_FILE", cfgfilename)
 440         */
 441         xapp.Logger.Info("Using cfg file %s", os.Getenv("CFG_FILE"))
 442
 443         //---------------------------------
 444         // Static routetable for rmr
 445         //
 446         // NOTE: Routing table is configured so, that responses
 447         //       are duplicated to xapp1 and xapp2 instances.
 448         //       If XID is not matching xapp stub will just
 449         //       drop message. (Messages 12011, 12012, 12021, 12022)
 450         //
 451         // 14560 submgr
 452         // 15560 e2term stub
 453         // 13560 xapp1 stub
 454         // 13660 xapp2 stub
 455         //
 456         //---------------------------------
 457
 458         allrt := `newrt|start
 459 mse|12010|-1|localhost:14560
 460 mse|12010,localhost:14560|-1|localhost:15560
 461 mse|12011,localhost:15560|-1|localhost:14560
 462 mse|12012,localhost:15560|-1|localhost:14560
 463 mse|12011,localhost:14560|-1|localhost:13660;localhost:13560
 464 mse|12012,localhost:14560|-1|localhost:13660;localhost:13560
 465 mse|12020|-1|localhost:14560
 466 mse|12020,localhost:14560|-1|localhost:15560
 467 mse|12021,localhost:15560|-1|localhost:14560
 468 mse|12022,localhost:15560|-1|localhost:14560
 469 mse|12021,localhost:14560|-1|localhost:13660;localhost:13560
 470 mse|12022,localhost:14560|-1|localhost:13660;localhost:13560
 471 mse|55555|-1|localhost:13660;localhost:13560,localhost:15560
 472 newrt|end
 473 `
 474
 475         //---------------------------------
 476         //
 477         //---------------------------------
 478         xapp.Logger.Info("### submgr main run ###")
 479
 480         subsrt := allrt
 481         /*
 482                 subsrt := `newrt|start
 483            mse|12010|-1|localhost:14560
 484            mse|12010,localhost:14560|-1|localhost:15560
 485            mse|12011,localhost:15560|-1|localhost:14560
 486            mse|12011|-1|localhost:13560;localhost:13660
 487            mse|12012,localhost:15560|-1|localhost:14560
 488            mse|12012|-1|localhost:13560;localhost:13660
 489            mse|12020|-1|localhost:14560
 490            mse|12020,localhost:14560|-1|localhost:15560
 491            mse|12021,localhost:15560|-1|localhost:14560
 492            mse|12021|-1|localhost:13560;localhost:13660
 493            mse|12022,localhost:15560|-1|localhost:14560
 494            mse|12022|-1|localhost:13560;localhost:13660
 495            newrt|end
 496            `
 497         */
 498
 499         subrtfilename, _ := testCreateTmpFile(subsrt)
 500         defer os.Remove(subrtfilename)
 501         mainCtrl = createNewMainControl("main", subrtfilename, "14560")
 502
 503         //---------------------------------
 504         //
 505         //---------------------------------
 506         xapp.Logger.Info("### xapp1 rmr run ###")
 507
 508         xapprt1 := allrt
 509         /*
 510                 xapprt1 := `newrt|start
 511            mse|12010|-1|localhost:14560
 512            mse|12011|-1|localhost:13560
 513            mse|12012|-1|localhost:13560
 514            mse|12020|-1|localhost:14560
 515            mse|12021|-1|localhost:13560
 516            mse|12022|-1|localhost:13560
 517            newrt|end
 518            `
 519         */
 520
 521         xapprtfilename1, _ := testCreateTmpFile(xapprt1)
 522         defer os.Remove(xapprtfilename1)
 523         xappConn1 = createNewXappControl("xappstub1", xapprtfilename1, "13560", "RMRXAPP1STUB")
 524
 525         //---------------------------------
 526         //
 527         //---------------------------------
 528
 529         xapp.Logger.Info("### xapp2 rmr run ###")
 530
 531         xapprt2 := allrt
 532         /*
 533                 xapprt2 := `newrt|start
 534            mse|12010|-1|localhost:14560
 535            mse|12011|-1|localhost:13660
 536            mse|12012|-1|localhost:13660
 537            mse|12020|-1|localhost:14560
 538            mse|12021|-1|localhost:13660
 539            mse|12022|-1|localhost:13660
 540            newrt|end
 541            `
 542         */
 543
 544         xapprtfilename2, _ := testCreateTmpFile(xapprt2)
 545         defer os.Remove(xapprtfilename2)
 546         xappConn2 = createNewXappControl("xappstub2", xapprtfilename2, "13660", "RMRXAPP2STUB")
 547
 548         //---------------------------------
 549         //
 550         //---------------------------------
 551         xapp.Logger.Info("### e2term rmr run ###")
 552
 553         e2termrt := allrt
 554         /*
 555                 e2termrt := `newrt|start
 556            mse|12010|-1|localhost:15560
 557            mse|12011|-1|localhost:14560
 558            mse|12012|-1|localhost:14560
 559            mse|12020|-1|localhost:15560
 560            mse|12021|-1|localhost:14560
 561            mse|12022|-1|localhost:14560
 562            newrt|end
 563            `
 564         */
 565
 566         e2termrtfilename, _ := testCreateTmpFile(e2termrt)
 567         defer os.Remove(e2termrtfilename)
 568         e2termConn = createNewE2termControl("e2termstub", e2termrtfilename, "15560", "RMRE2TERMSTUB")
 569
 570         //---------------------------------
 571         // Testing message sending
 572         //---------------------------------
 573         var dummyBuf []byte = make([]byte, 100)
 574
 575         params := &RMRParams{&xapp.RMRParams{}}
 576         params.Mtype = 55555
 577         params.SubId = -1
 578         params.Payload = dummyBuf
 579         params.PayloadLen = 100
 580         params.Meid = &xapp.RMRMeid{RanName: "NONEXISTINGRAN"}
 581         params.Xid = "THISISTESTFORSTUBS"
 582         params.Mbuf = nil
 583
 584         status := false
 585         i := 1
 586         for ; i <= 10 && status == false; i++ {
 587                 xapp.Rmr.Send(params.RMRParams, false)
 588                 if e2termConn.active == true && xappConn1.active == true && xappConn2.active == true {
 589                         status = true
 590                         break
 591                 } else {
 592                         xapp.Logger.Info("Sleep 0.5 secs and try routes again")
 593                         time.Sleep(500 * time.Millisecond)
 594                 }
 595         }
 596
 597         if status == false {
 598                 xapp.Logger.Error("Could not initialize routes")
 599                 os.Exit(1)
 600         }
 601
 602         //---------------------------------
 603         //
 604         //---------------------------------
 605         code := m.Run()
 606         os.Exit(code)
 607 }