+ assert def_pay == {"test send 60001": 2}
+ assert sixty_pay == {"test send 60000": 1}
+
+
+def test_rmr_healthcheck():
+ # thanos uses the rmr xapp to healthcheck the rmr xapp
+
+ # test variables
+ health_pay = None
+
+ def post_init(self):
+ self.rmr_send(b"", RIC_HEALTH_CHECK_REQ)
+
+ def default_handler(self, summary, sbuf):
+ pass
+
+ global rmr_xapp_health
+ rmr_xapp_health = RMRXapp(default_handler, post_init=post_init, rmr_port=4666, use_fake_sdl=True)
+
+ def health_handler(self, summary, sbuf):
+ nonlocal health_pay
+ health_pay = summary["payload"]
+ self.rmr_free(sbuf)
+
+ rmr_xapp_health.register_callback(health_handler, RIC_HEALTH_CHECK_RESP)
+ rmr_xapp_health.run(thread=True) # in unit tests we need to thread here or else execution is not returned!
+
+ time.sleep(1)
+
+ assert health_pay == b"OK\n"
+
+
+def test_get_list_nodeb(rnib_information):
+ global rnib_xapp
+ rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True)
+
+ # Test there is no rnib information.
+ gnb_list = rnib_xapp.get_list_gnb_ids()
+ enb_list = rnib_xapp.get_list_enb_ids()
+ assert len(gnb_list) == 0
+ assert len(enb_list) == 0
+
+ # Add rnib information directly.
+ for rnib in rnib_information:
+ rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENB", rnib, usemsgpack=False)
+ rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "GNB", rnib, usemsgpack=False)
+
+ gnb_list = rnib_xapp.get_list_gnb_ids()
+ assert len(gnb_list) == len(rnib_information)
+ for gnb in gnb_list:
+ assert gnb.SerializeToString() in rnib_information
+
+ enb_list = rnib_xapp.get_list_enb_ids()
+ assert len(enb_list) == len(rnib_information)
+ for enb in enb_list:
+ assert enb.SerializeToString() in rnib_information