tests/test_xapps.py

   1 # ==================================================================================
   2 #       Copyright (c) 2020 Nokia
   3 #       Copyright (c) 2020 AT&T Intellectual Property.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #          http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 # ==================================================================================
  17 import json
  18 import time
  19 from contextlib import suppress
  20
  21 from ricxappframe.util.constants import Constants
  22 from ricxappframe.xapp_frame import _BaseXapp, Xapp, RMRXapp
  23 from ricxappframe.constants import sdl_namespaces
  24
  25 import ricxappframe.entities.rnib.nb_identity_pb2 as pb_nb
  26
  27 rmr_xapp = None
  28 rmr_xapp_health = None
  29 gen_xapp = None
  30 rnib_xapp = None
  31
  32
  33 def test_rmr_init():
  34
  35     # test variables
  36     def_pay = None
  37     sixty_pay = None
  38
  39     # create rmr app
  40
  41     def default_handler(self, summary, sbuf):
  42         nonlocal def_pay
  43         def_pay = json.loads(summary["payload"])
  44         self.rmr_free(sbuf)
  45
  46     def sixtythou_handler(self, summary, sbuf):
  47         nonlocal sixty_pay
  48         sixty_pay = json.loads(summary["payload"])
  49         self.rmr_free(sbuf)
  50
  51     global rmr_xapp
  52     rmr_xapp = RMRXapp(default_handler, rmr_port=4564, use_fake_sdl=True)
  53     rmr_xapp.register_callback(sixtythou_handler, 60000)
  54     rmr_xapp.run(thread=True)  # in unit tests we need to thread here or else execution is not returned!
  55
  56     time.sleep(1)
  57
  58     # create a general xapp that will demonstrate some SDL functionality and launch some requests against the rmr xapp
  59
  60     def entry(self):
  61
  62         time.sleep(1)
  63
  64         self.sdl_set("testns", "mykey", 6)
  65         assert self.sdl_get("testns", "mykey") == 6
  66         assert self.sdl_find_and_get("testns", "myk") == {"mykey": 6}
  67         assert self.healthcheck()
  68
  69         val = json.dumps({"test send 60000": 1}).encode()
  70         self.rmr_send(val, 60000)
  71
  72         val = json.dumps({"test send 60001": 2}).encode()
  73         self.rmr_send(val, 60001)
  74
  75         self.sdl_delete("testns", "bogus")
  76
  77     global gen_xapp
  78     gen_xapp = Xapp(entrypoint=entry, use_fake_sdl=True)
  79     gen_xapp.run()
  80
  81     time.sleep(1)
  82
  83     assert def_pay == {"test send 60001": 2}
  84     assert sixty_pay == {"test send 60000": 1}
  85
  86
  87 def test_rmr_healthcheck():
  88     # thanos uses the rmr xapp to healthcheck the rmr xapp
  89
  90     # test variables
  91     health_pay = None
  92
  93     def post_init(self):
  94         self.rmr_send(b"", Constants.RIC_HEALTH_CHECK_REQ)
  95
  96     def default_handler(self, summary, sbuf):
  97         pass
  98
  99     global rmr_xapp_health
 100     rmr_xapp_health = RMRXapp(default_handler, post_init=post_init, rmr_port=4666, use_fake_sdl=True)
 101
 102     def health_handler(self, summary, sbuf):
 103         nonlocal health_pay
 104         health_pay = summary["payload"]
 105         self.rmr_free(sbuf)
 106
 107     rmr_xapp_health.register_callback(health_handler, Constants.RIC_HEALTH_CHECK_RESP)
 108     rmr_xapp_health.run(thread=True)  # in unit tests we need to thread here or else execution is not returned!
 109
 110     time.sleep(1)
 111
 112     assert health_pay == b"OK\n"
 113
 114
 115 def test_rnib_get_list_nodeb(rnib_information):
 116     global rnib_xapp
 117     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 118
 119     # Test there is no rnib information.
 120     gnb_list = rnib_xapp.get_list_gnb_ids()
 121     enb_list = rnib_xapp.get_list_enb_ids()
 122     assert len(gnb_list) == 0
 123     assert len(enb_list) == 0
 124
 125     # Add rnib information directly.
 126     for rnib in rnib_information:
 127         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENB", rnib, usemsgpack=False)
 128         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "GNB", rnib, usemsgpack=False)
 129
 130     gnb_list = rnib_xapp.get_list_gnb_ids()
 131     assert len(gnb_list) == len(rnib_information)
 132     for gnb in gnb_list:
 133         assert gnb.SerializeToString() in rnib_information
 134
 135     enb_list = rnib_xapp.get_list_enb_ids()
 136     assert len(enb_list) == len(rnib_information)
 137     for enb in enb_list:
 138         assert enb.SerializeToString() in rnib_information
 139
 140     rnib_xapp.stop()
 141
 142
 143 def test_rnib_get_list_all_nodeb(rnib_information):
 144     global rnib_xapp
 145     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 146
 147     # Add rnib information directly.
 148     for rnib in rnib_information:
 149         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "GNB", rnib, usemsgpack=False)
 150
 151     nb_list = rnib_xapp.GetListNodebIds()
 152     assert len(nb_list) == 2
 153
 154     for rnib in rnib_information:
 155         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENB", rnib, usemsgpack=False)
 156
 157     nb_list = rnib_xapp.GetListNodebIds()
 158     assert len(nb_list) == 4
 159
 160     rnib_xapp.stop()
 161
 162
 163 def test_rnib_get_list_cells(rnib_cellinformation):
 164     global rnib_xapp
 165
 166     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 167
 168     mynb = pb_nb.NbIdentity()
 169     mynb.inventory_name = "nodeb_1234"
 170     mynb.global_nb_id.plmn_id = "plmn_1234"
 171     mynb.global_nb_id.nb_id = "nb_1234"
 172     mynb.connection_status = 1
 173     rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENB", mynb.SerializeToString(), usemsgpack=False)
 174
 175     # Add rnib information directly.
 176     for rnib in rnib_cellinformation:
 177         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENBCELL1", rnib, usemsgpack=False)
 178         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENBCELL2", rnib, usemsgpack=False)
 179     rnib_xapp.stop()
 180
 181
 182 def test_rnib_get_nodeb(rnib_helpers):
 183     global rnib_xapp
 184     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 185     nb1 = rnib_helpers.createNodebInfo('nodeb_1234', 'GNB', '192.168.1.1', 8088)
 186     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False)
 187     nb2 = rnib_helpers.createNodebInfo('nodeb_1234', 'ENB', '192.168.1.2', 8088)
 188     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False)
 189
 190     gnb = rnib_xapp.GetNodeb('nodeb_1235')
 191     assert gnb == nb2
 192     gnb = rnib_xapp.GetNodeb('nodeb_1234')
 193     assert gnb == nb1
 194     gnb = rnib_xapp.GetNodeb('nodeb_1230')
 195     assert gnb is None
 196     rnib_xapp.stop()
 197
 198
 199 def test_rnib_get_cell(rnib_helpers):
 200     global rnib_xapp
 201     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 202     c1 = rnib_helpers.createCell('c1234', 8)
 203     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "PCI:c1234:08", c1.SerializeToString(), usemsgpack=False)
 204     c2 = rnib_helpers.createCell('c1235', 11)
 205     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "PCI:c1235:0b", c2.SerializeToString(), usemsgpack=False)
 206
 207     cell = rnib_xapp.GetCell('c1235', 11)
 208     assert cell == c2
 209     cell = rnib_xapp.GetCell('c1234', 8)
 210     assert cell == c1
 211     cell = rnib_xapp.GetCell('c1236', 11)
 212     assert cell is None
 213     rnib_xapp.stop()
 214
 215
 216 def test_rnib_get_cell_by_id(rnib_helpers):
 217     global rnib_xapp
 218     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 219     c1 = rnib_helpers.createCell('c1234', 8)
 220     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "CELL:c1234", c1.SerializeToString(), usemsgpack=False)
 221     c2 = rnib_helpers.createCell('c1235', 11)
 222     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "CELL:c1235", c2.SerializeToString(), usemsgpack=False)
 223
 224     cell = rnib_xapp.GetCellById('LTE_CELL', 'c1235')
 225     assert cell == c2
 226     cell = rnib_xapp.GetCellById('LTE_CELL', 'c1234')
 227     assert cell == c1
 228     cell = rnib_xapp.GetCellById('LTE_CELL', 'c1236')
 229     assert cell is None
 230     rnib_xapp.stop()
 231
 232
 233 def test_rnib_get_cells(rnib_helpers):
 234     global rnib_xapp
 235     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 236     nb1 = rnib_helpers.createNodebInfo('nodeb_1234', 'GNB', '192.168.1.1', 8088)
 237     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False)
 238     nb2 = rnib_helpers.createNodebInfo('nodeb_1234', 'ENB', '192.168.1.2', 8088)
 239     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False)
 240
 241     sc = rnib_xapp.GetCellList('nodeb_1235')
 242     assert sc == nb2.enb.served_cells
 243     sc = rnib_xapp.GetCellList('nodeb_1234')
 244     assert sc == nb1.gnb.served_nr_cells
 245     sc = rnib_xapp.GetCellList('nodeb_1230')
 246     assert sc is None
 247     rnib_xapp.stop()
 248
 249
 250 def test_rnib_get_global_nodeb(rnib_helpers):
 251     global rnib_xapp
 252     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 253     nb1 = rnib_helpers.createNodeb('nodeb_1234', '358', 'nb_1234')
 254     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "GNB:" + '358:' + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False)
 255     nb2 = rnib_helpers.createNodeb('nodeb_1235', '356', 'nb_1235')
 256     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "GNB:" + '356:' + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False)
 257
 258     gnb = rnib_xapp.GetNodebByGlobalNbId('GNB', '356', 'nodeb_1235')
 259     assert gnb == nb2
 260     gnb = rnib_xapp.GetNodebByGlobalNbId('GNB', '358', 'nodeb_1234')
 261     assert gnb == nb1
 262     gnb = rnib_xapp.GetNodebByGlobalNbId('GNB', '356', 'nodeb_1230')
 263     assert gnb is None
 264     rnib_xapp.stop()
 265
 266
 267 def test_rnib_get_ranfunction(rnib_helpers):
 268     global rnib_xapp
 269     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 270     nb1 = rnib_helpers.createNodebInfo('nodeb_1234', 'GNB', '192.168.1.1', 8088)
 271     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False)
 272     nb2 = rnib_helpers.createNodebInfo('nodeb_1235', 'GNB', '192.168.1.2', 8088)
 273     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False)
 274     nb3 = rnib_helpers.createNodebInfo('nodeb_1236', 'GNB', '192.168.1.2', 8088)
 275     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1236', nb3.SerializeToString(), usemsgpack=False)
 276     nb4 = rnib_helpers.createNodebInfo('nodeb_1237', 'GNB', '192.168.1.2', 8088)
 277     rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1237', nb4.SerializeToString(), usemsgpack=False)
 278
 279     sc = rnib_xapp.GetRanFunctionDefinition('nodeb_1235', "1.3.6.1.4.1.1.2.2.2")
 280     assert sc == ['te524367153']
 281     sc = rnib_xapp.GetRanFunctionDefinition('nodeb_1235', "1.3.6.1.4.1.1.2.2.5")
 282     assert sc == []
 283     rnib_xapp.stop()
 284
 285
 286 def teardown_module():
 287     """
 288     this is like a "finally"; the name of this function is pytest magic
 289     safer to put down here since certain failures above can lead to pytest never returning
 290     for example if an exception gets raised before stop is called in any test function above, pytest will hang forever
 291     """
 292     with suppress(Exception):
 293         gen_xapp.stop()
 294     with suppress(Exception):
 295         rmr_xapp.stop()
 296     with suppress(Exception):
 297         rmr_xapp_health.stop()