X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=tests%2Ftest_xapps.py;h=20ebe86246141e651de32183691dfa29aa367e1e;hb=HEAD;hp=c52f54fce35c8cc32caf10dd9292937aeaa31bf7;hpb=0f8305b167601ce6ca87c55208038455a65eab62;p=ric-plt%2Fxapp-frame-py.git diff --git a/tests/test_xapps.py b/tests/test_xapps.py index c52f54f..20ebe86 100644 --- a/tests/test_xapps.py +++ b/tests/test_xapps.py @@ -17,46 +17,50 @@ import json import time from contextlib import suppress -from ricxappframe.xapp_frame import Xapp, RMRXapp + +from ricxappframe.util.constants import Constants +from ricxappframe.xapp_frame import _BaseXapp, Xapp, RMRXapp +from ricxappframe.constants import sdl_namespaces + +import ricxappframe.entities.rnib.nb_identity_pb2 as pb_nb rmr_xapp = None +rmr_xapp_health = None gen_xapp = None +rnib_xapp = None -def test_flow(): +def test_rmr_init(): # test variables - def_called = 0 - sixty_called = 0 + def_pay = None + sixty_pay = None # create rmr app - global rmr_xapp - - def post_init(self): - self.logger.info("suppp info") - self.logger.debug("suppp debug") def default_handler(self, summary, sbuf): - nonlocal def_called - def_called += 1 - assert json.loads(summary["payload"]) == {"test send 60001": 1} + nonlocal def_pay + def_pay = json.loads(summary["payload"]) self.rmr_free(sbuf) - rmr_xapp = RMRXapp(default_handler, post_init=post_init, rmr_port=4564, rmr_wait_for_ready=False, use_fake_sdl=True) - def sixtythou_handler(self, summary, sbuf): - nonlocal sixty_called - sixty_called += 1 - assert json.loads(summary["payload"]) == {"test send 60000": 1} + nonlocal sixty_pay + sixty_pay = json.loads(summary["payload"]) self.rmr_free(sbuf) + global rmr_xapp + rmr_xapp = RMRXapp(default_handler, rmr_port=4564, use_fake_sdl=True) rmr_xapp.register_callback(sixtythou_handler, 60000) rmr_xapp.run(thread=True) # in unit tests we need to thread here or else execution is not returned! time.sleep(1) + # create a general xapp that will demonstrate some SDL functionality and launch some requests against the rmr xapp + def entry(self): + time.sleep(1) + self.sdl_set("testns", "mykey", 6) assert self.sdl_get("testns", "mykey") == 6 assert self.sdl_find_and_get("testns", "myk") == {"mykey": 6} @@ -68,14 +72,215 @@ def test_flow(): 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, rmr_wait_for_ready=False, use_fake_sdl=True) + gen_xapp = Xapp(entrypoint=entry, use_fake_sdl=True) gen_xapp.run() time.sleep(1) - assert def_called == 1 - assert sixty_called == 1 + 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"", Constants.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, 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) + + assert health_pay == b"OK\n" + + +def test_rnib_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 + + rnib_xapp.stop() + + +def test_rnib_get_list_all_nodeb(rnib_information): + global rnib_xapp + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + + # Add rnib information directly. + for rnib in rnib_information: + rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "GNB", rnib, usemsgpack=False) + + nb_list = rnib_xapp.GetListNodebIds() + assert len(nb_list) == 2 + + for rnib in rnib_information: + rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENB", rnib, usemsgpack=False) + + nb_list = rnib_xapp.GetListNodebIds() + assert len(nb_list) == 4 + + rnib_xapp.stop() + + +def test_rnib_get_list_cells(rnib_cellinformation): + global rnib_xapp + + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + + mynb = pb_nb.NbIdentity() + mynb.inventory_name = "nodeb_1234" + mynb.global_nb_id.plmn_id = "plmn_1234" + mynb.global_nb_id.nb_id = "nb_1234" + mynb.connection_status = 1 + rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENB", mynb.SerializeToString(), usemsgpack=False) + + # Add rnib information directly. + for rnib in rnib_cellinformation: + rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENBCELL1", rnib, usemsgpack=False) + rnib_xapp.sdl.add_member(sdl_namespaces.E2_MANAGER, "ENBCELL2", rnib, usemsgpack=False) + rnib_xapp.stop() + + +def test_rnib_get_nodeb(rnib_helpers): + global rnib_xapp + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + nb1 = rnib_helpers.createNodebInfo('nodeb_1234', 'GNB', '192.168.1.1', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False) + nb2 = rnib_helpers.createNodebInfo('nodeb_1234', 'ENB', '192.168.1.2', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False) + + gnb = rnib_xapp.GetNodeb('nodeb_1235') + assert gnb == nb2 + gnb = rnib_xapp.GetNodeb('nodeb_1234') + assert gnb == nb1 + gnb = rnib_xapp.GetNodeb('nodeb_1230') + assert gnb is None + rnib_xapp.stop() + + +def test_rnib_get_cell(rnib_helpers): + global rnib_xapp + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + c1 = rnib_helpers.createCell('c1234', 8) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "PCI:c1234:08", c1.SerializeToString(), usemsgpack=False) + c2 = rnib_helpers.createCell('c1235', 11) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "PCI:c1235:0b", c2.SerializeToString(), usemsgpack=False) + + cell = rnib_xapp.GetCell('c1235', 11) + assert cell == c2 + cell = rnib_xapp.GetCell('c1234', 8) + assert cell == c1 + cell = rnib_xapp.GetCell('c1236', 11) + assert cell is None + rnib_xapp.stop() + + +def test_rnib_get_cell_by_id(rnib_helpers): + global rnib_xapp + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + c1 = rnib_helpers.createCell('c1234', 8) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "CELL:c1234", c1.SerializeToString(), usemsgpack=False) + c2 = rnib_helpers.createCell('c1235', 11) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "CELL:c1235", c2.SerializeToString(), usemsgpack=False) + + cell = rnib_xapp.GetCellById('LTE_CELL', 'c1235') + assert cell == c2 + cell = rnib_xapp.GetCellById('LTE_CELL', 'c1234') + assert cell == c1 + cell = rnib_xapp.GetCellById('LTE_CELL', 'c1236') + assert cell is None + rnib_xapp.stop() + + +def test_rnib_get_cells(rnib_helpers): + global rnib_xapp + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + nb1 = rnib_helpers.createNodebInfo('nodeb_1234', 'GNB', '192.168.1.1', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False) + nb2 = rnib_helpers.createNodebInfo('nodeb_1234', 'ENB', '192.168.1.2', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False) + + sc = rnib_xapp.GetCellList('nodeb_1235') + assert sc == nb2.enb.served_cells + sc = rnib_xapp.GetCellList('nodeb_1234') + assert sc == nb1.gnb.served_nr_cells + sc = rnib_xapp.GetCellList('nodeb_1230') + assert sc is None + rnib_xapp.stop() + + +def test_rnib_get_global_nodeb(rnib_helpers): + global rnib_xapp + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + nb1 = rnib_helpers.createNodeb('nodeb_1234', '358', 'nb_1234') + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "GNB:" + '358:' + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False) + nb2 = rnib_helpers.createNodeb('nodeb_1235', '356', 'nb_1235') + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "GNB:" + '356:' + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False) + + gnb = rnib_xapp.GetNodebByGlobalNbId('GNB', '356', 'nodeb_1235') + assert gnb == nb2 + gnb = rnib_xapp.GetNodebByGlobalNbId('GNB', '358', 'nodeb_1234') + assert gnb == nb1 + gnb = rnib_xapp.GetNodebByGlobalNbId('GNB', '356', 'nodeb_1230') + assert gnb is None + rnib_xapp.stop() + + +def test_rnib_get_ranfunction(rnib_helpers): + global rnib_xapp + rnib_xapp = _BaseXapp(rmr_port=4777, rmr_wait_for_ready=False, use_fake_sdl=True) + nb1 = rnib_helpers.createNodebInfo('nodeb_1234', 'GNB', '192.168.1.1', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1234', nb1.SerializeToString(), usemsgpack=False) + nb2 = rnib_helpers.createNodebInfo('nodeb_1235', 'GNB', '192.168.1.2', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1235', nb2.SerializeToString(), usemsgpack=False) + nb3 = rnib_helpers.createNodebInfo('nodeb_1236', 'GNB', '192.168.1.2', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1236', nb3.SerializeToString(), usemsgpack=False) + nb4 = rnib_helpers.createNodebInfo('nodeb_1237', 'GNB', '192.168.1.2', 8088) + rnib_xapp.sdl.set(sdl_namespaces.E2_MANAGER, "RAN:" + 'nodeb_1237', nb4.SerializeToString(), usemsgpack=False) + + sc = rnib_xapp.GetRanFunctionDefinition('nodeb_1235', "1.3.6.1.4.1.1.2.2.2") + assert sc == ['te524367153'] + sc = rnib_xapp.GetRanFunctionDefinition('nodeb_1235', "1.3.6.1.4.1.1.2.2.5") + assert sc == [] + rnib_xapp.stop() def teardown_module(): @@ -88,3 +293,5 @@ def teardown_module(): gen_xapp.stop() with suppress(Exception): rmr_xapp.stop() + with suppress(Exception): + rmr_xapp_health.stop()