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 from ricxappframe.xapp_frame import _BaseXapp, Xapp, RMRXapp, RIC_HEALTH_CHECK_REQ, RIC_HEALTH_CHECK_RESP
  21 from ricxappframe.constants import sdl_namespaces
  22
  23 rmr_xapp = None
  24 rmr_xapp_health = None
  25 gen_xapp = None
  26 rnib_xapp = None
  27
  28
  29 def test_rmr_init():
  30
  31     # test variables
  32     def_pay = None
  33     sixty_pay = None
  34
  35     # create rmr app
  36
  37     def default_handler(self, summary, sbuf):
  38         nonlocal def_pay
  39         def_pay = json.loads(summary["payload"])
  40         self.rmr_free(sbuf)
  41
  42     def sixtythou_handler(self, summary, sbuf):
  43         nonlocal sixty_pay
  44         sixty_pay = json.loads(summary["payload"])
  45         self.rmr_free(sbuf)
  46
  47     global rmr_xapp
  48     rmr_xapp = RMRXapp(default_handler, rmr_port=4564, use_fake_sdl=True)
  49     rmr_xapp.register_callback(sixtythou_handler, 60000)
  50     rmr_xapp.run(thread=True)  # in unit tests we need to thread here or else execution is not returned!
  51
  52     time.sleep(1)
  53
  54     # create a general xapp that will demonstrate some SDL functionality and launch some requests against the rmr xapp
  55
  56     def entry(self):
  57
  58         time.sleep(1)
  59
  60         self.sdl_set("testns", "mykey", 6)
  61         assert self.sdl_get("testns", "mykey") == 6
  62         assert self.sdl_find_and_get("testns", "myk") == {"mykey": 6}
  63         assert self.healthcheck()
  64
  65         val = json.dumps({"test send 60000": 1}).encode()
  66         self.rmr_send(val, 60000)
  67
  68         val = json.dumps({"test send 60001": 2}).encode()
  69         self.rmr_send(val, 60001)
  70
  71         self.sdl_delete("testns", "bogus")
  72
  73     global gen_xapp
  74     gen_xapp = Xapp(entrypoint=entry, use_fake_sdl=True)
  75     gen_xapp.run()
  76
  77     time.sleep(1)
  78
  79     assert def_pay == {"test send 60001": 2}
  80     assert sixty_pay == {"test send 60000": 1}
  81
  82
  83 def test_rmr_healthcheck():
  84     # thanos uses the rmr xapp to healthcheck the rmr xapp
  85
  86     # test variables
  87     health_pay = None
  88
  89     def post_init(self):
  90         self.rmr_send(b"", RIC_HEALTH_CHECK_REQ)
  91
  92     def default_handler(self, summary, sbuf):
  93         pass
  94
  95     global rmr_xapp_health
  96     rmr_xapp_health = RMRXapp(default_handler, post_init=post_init, rmr_port=4666, use_fake_sdl=True)
  97
  98     def health_handler(self, summary, sbuf):
  99         nonlocal health_pay
 100         health_pay = summary["payload"]
 101         self.rmr_free(sbuf)
 102
 103     rmr_xapp_health.register_callback(health_handler, RIC_HEALTH_CHECK_RESP)
 104     rmr_xapp_health.run(thread=True)  # in unit tests we need to thread here or else execution is not returned!
 105
 106     time.sleep(1)
 107
 108     assert health_pay == b"OK\n"
 109
 110
 111 def test_get_list_nodeb(rnib_information):
 112     global rnib_xapp
 113     rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
 114
 115     # Test there is no rnib information.
 116     gnb_list = rnib_xapp.get_list_gnb_ids()
 117     enb_list = rnib_xapp.get_list_enb_ids()
 118     assert len(gnb_list) == 0
 119     assert len(enb_list) == 0
 120
 121     # Add rnib information directly.
 122     for rnib in rnib_information:
 123         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENB", rnib, usemsgpack=False)
 124         rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "GNB", rnib, usemsgpack=False)
 125
 126     gnb_list = rnib_xapp.get_list_gnb_ids()
 127     assert len(gnb_list) == len(rnib_information)
 128     for gnb in gnb_list:
 129         assert gnb.SerializeToString() in rnib_information
 130
 131     enb_list = rnib_xapp.get_list_enb_ids()
 132     assert len(enb_list) == len(rnib_information)
 133     for enb in enb_list:
 134         assert enb.SerializeToString() in rnib_information
 135
 136
 137 def teardown_module():
 138     """
 139     this is like a "finally"; the name of this function is pytest magic
 140     safer to put down here since certain failures above can lead to pytest never returning
 141     for example if an exception gets raised before stop is called in any test function above, pytest will hang forever
 142     """
 143     with suppress(Exception):
 144         gen_xapp.stop()
 145     with suppress(Exception):
 146         rmr_xapp.stop()
 147     with suppress(Exception):
 148         rmr_xapp_health.stop()
 149     with suppress(Exception):
 150         rnib_xapp.stop()