X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=tests%2Ftest_xapps.py;h=3706f302af93f34bd66130dc56728585e0eaae79;hb=2650675e23003fa2d2dca319a72ba04e92316634;hp=cd806ac1a651a9ce26fa3d5e1be544e667a61e4d;hpb=09894e3c7bdd6eeaae4a84467f5ea6af1a061204;p=ric-plt%2Fxapp-frame-py.git

diff --git a/tests/test_xapps.py b/tests/test_xapps.py
index cd806ac..3706f30 100644
--- a/tests/test_xapps.py
+++ b/tests/test_xapps.py
@@ -17,11 +17,15 @@
 import json
 import time
 from contextlib import suppress
-from ricxappframe.xapp_frame import Xapp, RMRXapp, RIC_HEALTH_CHECK_REQ, RIC_HEALTH_CHECK_RESP
+
+from ricxappframe.util.constants import Constants
+from ricxappframe.xapp_frame import _BaseXapp, Xapp, RMRXapp
+from ricxappframe.constants import sdl_namespaces
 
 rmr_xapp = None
 rmr_xapp_health = None
 gen_xapp = None
+rnib_xapp = None
 
 
 def test_rmr_init():
@@ -66,6 +70,8 @@ def test_rmr_init():
         val = json.dumps({"test send 60001": 2}).encode()
         self.rmr_send(val, 60001)
 
+        self.sdl_delete("testns", "bogus")
+
     global gen_xapp
     gen_xapp = Xapp(entrypoint=entry, use_fake_sdl=True)
     gen_xapp.run()
@@ -83,7 +89,7 @@ def test_rmr_healthcheck():
     health_pay = None
 
     def post_init(self):
-        self.rmr_send(b"", RIC_HEALTH_CHECK_REQ)
+        self.rmr_send(b"", Constants.RIC_HEALTH_CHECK_REQ)
 
     def default_handler(self, summary, sbuf):
         pass
@@ -96,7 +102,7 @@ def test_rmr_healthcheck():
         health_pay = summary["payload"]
         self.rmr_free(sbuf)
 
-    rmr_xapp_health.register_callback(health_handler, RIC_HEALTH_CHECK_RESP)
+    rmr_xapp_health.register_callback(health_handler, Constants.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)
@@ -104,6 +110,32 @@ def test_rmr_healthcheck():
     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
+
+
 def teardown_module():
     """
     this is like a "finally"; the name of this function is pytest magic
@@ -116,3 +148,5 @@ def teardown_module():
         rmr_xapp.stop()
     with suppress(Exception):
         rmr_xapp_health.stop()
+    with suppress(Exception):
+        rnib_xapp.stop()