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         "errors"
  25         "fmt"
  26         "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_models"
  27         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  28         "io/ioutil"
  29         "net/http"
  30         "os"
  31         "strconv"
  32         "strings"
  33         "testing"
  34         "time"
  35 )
  36
  37 //-----------------------------------------------------------------------------
  38 //
  39 //-----------------------------------------------------------------------------
  40 type testingControl struct {
  41         desc     string
  42         syncChan chan struct{}
  43 }
  44
  45 func (tc *testingControl) ReadyCB(data interface{}) {
  46         xapp.Logger.Info("testingControl(%s) ReadyCB", tc.desc)
  47         tc.syncChan <- struct{}{}
  48         return
  49 }
  50
  51 func (tc *testingControl) WaitCB() {
  52         <-tc.syncChan
  53 }
  54
  55 func initTestingControl(desc string, rtfile string, port string) testingControl {
  56         tc := testingControl{}
  57         os.Setenv("RMR_SEED_RT", rtfile)
  58         os.Setenv("RMR_SRC_ID", "localhost:"+port)
  59         xapp.Logger.Info("Using rt file %s", os.Getenv("RMR_SEED_RT"))
  60         xapp.Logger.Info("Using src id  %s", os.Getenv("RMR_SRC_ID"))
  61         tc.desc = strings.ToUpper(desc)
  62         tc.syncChan = make(chan struct{})
  63         return tc
  64 }
  65
  66 //-----------------------------------------------------------------------------
  67 //
  68 //-----------------------------------------------------------------------------
  69 type testingRmrControl struct {
  70         testingControl
  71         rmrClientTest *xapp.RMRClient
  72 }
  73
  74 func (tc *testingRmrControl) RmrSend(params *xapp.RMRParams) (err error) {
  75         //
  76         //NOTE: Do this way until xapp-frame sending is improved
  77         //
  78         status := false
  79         i := 1
  80         for ; i <= 10 && status == false; i++ {
  81                 status = tc.rmrClientTest.SendMsg(params)
  82                 if status == false {
  83                         xapp.Logger.Info("rmr.Send() failed. Retry count %v, Mtype: %v, SubId: %v, Xid %s", i, params.Mtype, params.SubId, params.Xid)
  84                         time.Sleep(500 * time.Millisecond)
  85                 }
  86         }
  87         if status == false {
  88                 err = errors.New("rmr.Send() failed")
  89                 tc.rmrClientTest.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 *xapp.RMRParams
 109 }
 110
 111 func initTestingMessageChannel() testingMessageChannel {
 112         mc := testingMessageChannel{}
 113         mc.rmrConChan = make(chan *xapp.RMRParams)
 114         return mc
 115 }
 116
 117 //-----------------------------------------------------------------------------
 118 //
 119 //-----------------------------------------------------------------------------
 120
 121 type testingXappControl struct {
 122         testingRmrControl
 123         testingMessageChannel
 124         meid    *xapp.RMRMeid
 125         xid_seq uint64
 126         xid     string
 127 }
 128
 129 func (tc *testingXappControl) newXid() string {
 130         tc.xid = tc.desc + "_XID_" + strconv.FormatUint(uint64(tc.xid_seq), 10)
 131         tc.xid_seq++
 132         return tc.xid
 133 }
 134
 135 func (tc *testingXappControl) Consume(msg *xapp.RMRParams) (err error) {
 136
 137         //if msg.Xid == tc.xid {
 138         if strings.Contains(msg.Xid, tc.desc) {
 139                 xapp.Logger.Info("(%s) Consume mtype=%s subid=%d xid=%s", tc.desc, xapp.RicMessageTypeToName[msg.Mtype], msg.SubId, msg.Xid)
 140                 tc.rmrConChan <- msg
 141         } else {
 142                 xapp.Logger.Info("(%s) Ignore mtype=%s subid=%d xid=%s, Expected xid to contain %s", tc.desc, xapp.RicMessageTypeToName[msg.Mtype], msg.SubId, msg.Xid, tc.desc)
 143         }
 144         return
 145 }
 146
 147 func createNewXappControl(desc string, rtfile string, port string, stat string, ranname string) *testingXappControl {
 148         xappCtrl := &testingXappControl{}
 149         xappCtrl.testingRmrControl = initTestingRmrControl(desc, rtfile, port, stat, xappCtrl)
 150         xappCtrl.testingMessageChannel = initTestingMessageChannel()
 151         xappCtrl.meid = &xapp.RMRMeid{RanName: ranname}
 152         xappCtrl.xid_seq = 0
 153         xappCtrl.newXid()
 154         return xappCtrl
 155 }
 156
 157 //-----------------------------------------------------------------------------
 158 //
 159 //-----------------------------------------------------------------------------
 160 type testingE2termControl struct {
 161         testingRmrControl
 162         testingMessageChannel
 163 }
 164
 165 func (tc *testingE2termControl) Consume(msg *xapp.RMRParams) (err error) {
 166         xapp.Logger.Info("(%s) Consume mtype=%s subid=%d xid=%s", tc.desc, xapp.RicMessageTypeToName[msg.Mtype], msg.SubId, msg.Xid)
 167         tc.rmrConChan <- msg
 168         return
 169 }
 170
 171 func createNewE2termControl(desc string, rtfile string, port string, stat string) *testingE2termControl {
 172         e2termCtrl := &testingE2termControl{}
 173         e2termCtrl.testingRmrControl = initTestingRmrControl(desc, rtfile, port, stat, e2termCtrl)
 174         e2termCtrl.testingMessageChannel = initTestingMessageChannel()
 175         return e2termCtrl
 176 }
 177
 178 //-----------------------------------------------------------------------------
 179 //
 180 //-----------------------------------------------------------------------------
 181 type testingMainControl struct {
 182         testingControl
 183         c *Control
 184 }
 185
 186 func createNewMainControl(desc string, rtfile string, port string) *testingMainControl {
 187         mainCtrl = &testingMainControl{}
 188         mainCtrl.testingControl = initTestingControl(desc, rtfile, port)
 189         mainCtrl.c = NewControl()
 190         xapp.SetReadyCB(mainCtrl.ReadyCB, nil)
 191         go xapp.RunWithParams(mainCtrl.c, false)
 192         mainCtrl.WaitCB()
 193         return mainCtrl
 194 }
 195
 196 //-----------------------------------------------------------------------------
 197 //
 198 //-----------------------------------------------------------------------------
 199
 200 func testError(t *testing.T, pattern string, args ...interface{}) {
 201         xapp.Logger.Error(fmt.Sprintf(pattern, args...))
 202         t.Errorf(fmt.Sprintf(pattern, args...))
 203 }
 204
 205 func testLog(t *testing.T, pattern string, args ...interface{}) {
 206         xapp.Logger.Info(fmt.Sprintf(pattern, args...))
 207         t.Logf(fmt.Sprintf(pattern, args...))
 208 }
 209
 210 func testCreateTmpFile(str string) (string, error) {
 211         file, err := ioutil.TempFile("/tmp", "*.rt")
 212         if err != nil {
 213                 return "", err
 214         }
 215         _, err = file.WriteString(str)
 216         if err != nil {
 217                 file.Close()
 218                 return "", err
 219         }
 220         return file.Name(), nil
 221 }
 222
 223 //-----------------------------------------------------------------------------
 224 //
 225 //-----------------------------------------------------------------------------
 226
 227 var xappConn1 *testingXappControl
 228 var xappConn2 *testingXappControl
 229 var e2termConn *testingE2termControl
 230 var mainCtrl *testingMainControl
 231
 232 func TestMain(m *testing.M) {
 233         xapp.Logger.Info("TestMain start")
 234
 235         //
 236         //Cfg creation won't work like this as xapp-frame reads it during init.
 237         //
 238         /*
 239             cfgstr:=`{
 240               "local": {
 241                   "host": ":8080"
 242               },
 243               "logger": {
 244                   "level": 4
 245               },
 246               "rmr": {
 247                  "protPort": "tcp:14560",
 248                  "maxSize": 4096,
 249                  "numWorkers": 1,
 250                  "txMessages": ["RIC_SUB_REQ", "RIC_SUB_DEL_REQ"],
 251                  "rxMessages": ["RIC_SUB_RESP", "RIC_SUB_FAILURE", "RIC_SUB_DEL_RESP", "RIC_SUB_DEL_FAILURE", "RIC_INDICATION"]
 252               },
 253               "db": {
 254                   "host": "localhost",
 255                   "port": 6379,
 256                   "namespaces": ["sdl", "rnib"]
 257               },
 258                  "rtmgr" : {
 259                    "HostAddr" : "localhost",
 260                    "port" : "8989",
 261                    "baseUrl" : "/"
 262                  }
 263            `
 264
 265            cfgfilename,_ := testCreateTmpFile(cfgstr)
 266            defer os.Remove(cfgfilename)
 267            os.Setenv("CFG_FILE", cfgfilename)
 268         */
 269         xapp.Logger.Info("Using cfg file %s", os.Getenv("CFG_FILE"))
 270
 271         //---------------------------------
 272         // NOTE: Routing table is configured so, that responses
 273         //       are duplicated to xapp1 and xapp2 instances.
 274         //       If XID is not matching xapp stub will just
 275         //       drop message. (Messages 12011, 12012, 12021, 12022)
 276         //---------------------------------
 277         xapp.Logger.Info("### submgr main run ###")
 278
 279         subsrt := `newrt|start
 280 mse|12010|-1|localhost:14560
 281 mse|12010,localhost:14560|-1|localhost:15560
 282 mse|12011,localhost:15560|-1|localhost:14560
 283 mse|12011|-1|localhost:13560;localhost:13660
 284 mse|12012,localhost:15560|-1|localhost:14560
 285 mse|12012|-1|localhost:13560;localhost:13660
 286 mse|12020|-1|localhost:14560
 287 mse|12020,localhost:14560|-1|localhost:15560
 288 mse|12021,localhost:15560|-1|localhost:14560
 289 mse|12021|-1|localhost:13560;localhost:13660
 290 mse|12022,localhost:15560|-1|localhost:14560
 291 mse|12022|-1|localhost:13560;localhost:13660
 292 newrt|end
 293 `
 294         subrtfilename, _ := testCreateTmpFile(subsrt)
 295         defer os.Remove(subrtfilename)
 296         mainCtrl = createNewMainControl("main", subrtfilename, "14560")
 297
 298         //---------------------------------
 299         //
 300         //---------------------------------
 301         xapp.Logger.Info("### xapp1 rmr run ###")
 302
 303         xapprt1 := `newrt|start
 304 mse|12010|-1|localhost:14560
 305 mse|12011|-1|localhost:13560
 306 mse|12012|-1|localhost:13560
 307 mse|12020|-1|localhost:14560
 308 mse|12021|-1|localhost:13560
 309 mse|12022|-1|localhost:13560
 310 newrt|end
 311 `
 312
 313         xapprtfilename1, _ := testCreateTmpFile(xapprt1)
 314         defer os.Remove(xapprtfilename1)
 315         xappConn1 = createNewXappControl("xappConn1", xapprtfilename1, "13560", "RMRXAPP1STUB", "RAN_NAME_1")
 316
 317         //---------------------------------
 318         //
 319         //---------------------------------
 320
 321         xapp.Logger.Info("### xapp2 rmr run ###")
 322
 323         xapprt2 := `newrt|start
 324 mse|12010|-1|localhost:14560
 325 mse|12011|-1|localhost:13660
 326 mse|12012|-1|localhost:13660
 327 mse|12020|-1|localhost:14560
 328 mse|12021|-1|localhost:13660
 329 mse|12022|-1|localhost:13660
 330 newrt|end
 331 `
 332
 333         xapprtfilename2, _ := testCreateTmpFile(xapprt2)
 334         defer os.Remove(xapprtfilename2)
 335         xappConn2 = createNewXappControl("xappConn2", xapprtfilename2, "13660", "RMRXAPP2STUB", "RAN_NAME_1")
 336
 337         //---------------------------------
 338         //
 339         //---------------------------------
 340         xapp.Logger.Info("### e2term rmr run ###")
 341
 342         e2termrt := `newrt|start
 343 mse|12010|-1|localhost:15560
 344 mse|12011|-1|localhost:14560
 345 mse|12012|-1|localhost:14560
 346 mse|12020|-1|localhost:15560
 347 mse|12021|-1|localhost:14560
 348 mse|12022|-1|localhost:14560
 349 newrt|end
 350 `
 351
 352         e2termrtfilename, _ := testCreateTmpFile(e2termrt)
 353         defer os.Remove(e2termrtfilename)
 354         e2termConn = createNewE2termControl("e2termConn", e2termrtfilename, "15560", "RMRE2TERMSTUB")
 355
 356         //---------------------------------
 357         //
 358         //---------------------------------
 359         http_handler := func(w http.ResponseWriter, r *http.Request) {
 360                 var req rtmgr_models.XappSubscriptionData
 361                 err := json.NewDecoder(r.Body).Decode(&req)
 362                 if err != nil {
 363                         xapp.Logger.Error("%s", err.Error())
 364                 }
 365                 xapp.Logger.Info("(http handler) handling Address=%s Port=%d SubscriptionID=%d", *req.Address, *req.Port, *req.SubscriptionID)
 366
 367                 w.WriteHeader(200)
 368         }
 369
 370         go func() {
 371                 http.HandleFunc("/", http_handler)
 372                 http.ListenAndServe("localhost:8989", nil)
 373         }()
 374
 375         //---------------------------------
 376         //
 377         //---------------------------------
 378         code := m.Run()
 379         os.Exit(code)
 380 }