X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=smo-install%2Ftest%2Fpythonsdk%2Fsrc%2Forantests%2Ftest_o1.py;h=8506ec46dc4225b5a6e6691a1e5d64b727b9ec69;hb=6c7e3c3c5e51182a891a2a21676f1cc2646ba484;hp=85771e83ec37a64f6b5390e4633cfcae8328d45c;hpb=8b713ed09d061baacf624a533be3351e7dcb0500;p=it%2Fdep.git diff --git a/smo-install/test/pythonsdk/src/orantests/test_o1.py b/smo-install/test/pythonsdk/src/orantests/test_o1.py index 85771e83..8506ec46 100644 --- a/smo-install/test/pythonsdk/src/orantests/test_o1.py +++ b/smo-install/test/pythonsdk/src/orantests/test_o1.py @@ -1,34 +1,171 @@ #!/usr/bin/env python3 -# SPDX-License-Identifier: Apache-2.0 - +### +# ============LICENSE_START======================================================= +# ORAN SMO PACKAGE - PYTHONSDK TESTS +# ================================================================================ +# Copyright (C) 2021-2022 AT&T Intellectual Property. All rights +# reserved. +# ================================================================================ +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ============LICENSE_END============================================ +# =================================================================== +# +### +"""O1 tests module.""" +import datetime import logging +import json +import os +import time +import pytest from onapsdk.configuration import settings +from smo.network_simulators import NetworkSimulators from oransdk.dmaap.dmaap import OranDmaap -import json +from oransdk.sdnc.sdnc import OranSdnc +from waiting import wait -BASIC_AUTH = {} +# Set working dir as python script location +abspath = os.path.abspath(__file__) +dname = os.path.dirname(abspath) +os.chdir(dname) logging.config.dictConfig(settings.LOG_CONFIG) -logger = logging.getLogger("test O1") - -def test_network_devices_registration(): - logger.info("Verify if SDNR sends well the RU registration to VES by checking in DMAAP") - dmaap = OranDmaap() - # As the user has been registered in DMAAP during test session init, that call should return all sims registered by SDNR - events = dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 10000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() - # events should be a list of messages - logger.info("Verify if the number of events is well equals to the number of devices") - assert (len(events) >= 6) - - for event in events: - logger.info(f"Checking event: {event}") - eventjson = json.loads(event) - logger.info("json ?"+str(eventjson)) - assert(eventjson["event"]["commonEventHeader"]["sourceName"] in settings.NETWORK_SIMULATOR_DEVICES_LIST) - - if "o-ru" in eventjson["event"]["commonEventHeader"]["sourceName"]: - logger.info(f"RU event detected checking SDNR has well registered it") - assert(eventjson["event"]["commonEventHeader"]["reportingEntityName"] == "ONAP SDN-R") - elif "o-du" in eventjson["event"]["commonEventHeader"]["sourceName"]: - logger.info(f"DU detected checking it has well registered itself") - assert("o-du" in eventjson["event"]["commonEventHeader"]["reportingEntityName"]) +logger = logging.getLogger("Test O1") + +network_simulators = NetworkSimulators("./resources") +dmaap = OranDmaap() +test_session_timestamp = datetime.datetime.now() + +@pytest.fixture(scope="module", autouse=True) +def setup_simulators(): + """Setup the simulators before the executing the tests.""" + logger.info("Test class setup for O1 tests") + + # Do a first get to register the o1test/o1test user in DMAAP + # all registration messages will then be stored for the registration tests. + # If it exists already it clears all cached events. + wait(lambda: (dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic unauthenticated.VES_PNFREG_OUTPUT to be empty") + wait(lambda: (dmaap.get_message_from_topic("unauthenticated.SEC_FAULT_OUTPUT", 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic unauthenticated.SEC_FAULT_OUTPUT to be empty") + network_simulators.start_network_simulators() + network_simulators.wait_for_network_simulators_to_be_running() + # ADD DU RESTART just in case + # Wait enough time to have at least the SDNR notifications sent + logger.info("Waiting 20s that SDNR sends all registration events to VES...") + time.sleep(10) + logger.info("Enabling faults/events reporting on SDNR") + network_simulators.enable_events_for_all_simulators() +# logger.info("Waiting 20s that the Dmaap faults topic is created...") +# time.sleep(20) + # Preparing the DMaap to cache all the events for the fault topics. + # If it exists already it clears all cached events. + logger.info("Waiting 120s to have registration and faults events in DMaap") + time.sleep(120) + logger.info("Test Session setup completed successfully") + + ### Cleanup code + yield + network_simulators.stop_network_simulators() + logger.info("Test Session cleanup done") + +def create_registration_structure(events): + """Decode the registration events list.""" + devices_found_in_events = dict() + for event in events: + event_json = json.loads(event) + logger.info("Registration json decoded: %s", str(event_json)) + devices_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = event_json["event"]["commonEventHeader"]["reportingEntityName"] + + logger.info("Devices found in events:%s", devices_found_in_events) + return devices_found_in_events + +def create_faults_structure(events): + """Decode the fault events list.""" + faults_found_in_events = dict() + for event in events: + event_json = json.loads(event) + logger.info("Fault json decoded: %s", str(event_json)) + if event_json["event"]["commonEventHeader"]["sourceName"] in faults_found_in_events: + faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] += 1 + else: + faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = 1 + logger.info("Faults found in events: %s", faults_found_in_events) + return faults_found_in_events + +def test_devices_in_sdnc(): + """Verify that the devices are well defined in SDNC.""" + logger.info("Verify if devices are well in SDNC") + for device in settings.NETWORK_SIMULATORS_LIST: + logger.info("Verify if %s is well in SDNR", device) + assert OranSdnc.get_devices(device, settings.SDNC_BASICAUTH) == 200 + +def validate_faults_timestamp(faults): + """Extract only the faults returned by SDNC that have been raised during this test.""" + valid_faults = [] + for fault in faults['data-provider:output']['data']: + converted_fault_timestamp = datetime.datetime.strptime(fault['timestamp'], "%Y-%m-%dT%H:%M:%S.%fZ") + logger.info("Comparing fault timestamp %s (%s) to session test timestamp %s", converted_fault_timestamp, fault['timestamp'], test_session_timestamp) + if converted_fault_timestamp > test_session_timestamp: + valid_faults.append(fault) + logger.info("Valid faults array: %s", valid_faults) + return valid_faults + +def test_device_faults_in_sdnc(): + """Verify that the device faults are well defined in SDNC.""" + logger.info("Verify is there is any events") + for device in settings.NETWORK_SIMULATORS_DU_RU_LIST: + faults = OranSdnc.get_events(settings.SDNC_BASICAUTH, device).json() + logger.info("Verify if %s has events", device) + assert len(validate_faults_timestamp(faults)) >= 3 + +def test_network_devices_registration_in_dmaap(): + """Validate that the devices are well registered in SDNR and forwarded to VES.""" + logger.info("Verify if SDNR sends well the RU registration to VES by checking in DMAAP") + # As the user has been registered in DMAAP during test session init, + # that call should return all sims registered by SDNR + all_registrations = [] + events = [] + + while (events := dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()) != []: + logger.info("Getting a first set of event: %s", events) + all_registrations += events + + logger.info("ALl registration events received: %s", all_registrations) + # events should be a list of messages + logger.info("Verify if the number of events is well >= to the number of expected devices") + # The DU can send multiple times message to VES and SDNR can send multiple time event for RU + assert len(all_registrations) >= (len(settings.NETWORK_SIMULATORS_DU_RU_LIST)) + devices_registered = create_registration_structure(all_registrations) + + # Each device must be at least one time in the event list + for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST: + logger.info("Checking if %s is in events list", sim_name) + assert sim_name in devices_registered + if "o-ru" in sim_name: + logger.info("RU event detected checking SDNR has well registered it") + assert "ONAP SDN-R" in devices_registered[sim_name] + elif "o-du" in sim_name: + logger.info("DU detected checking it has well registered itself") + assert "o-du" in devices_registered[sim_name] + +def test_device_faults_in_dmaap(): + """Verify that device faults are well sent to DMAAP by SDNR.""" + logger.info("Verify if SDNR forwards well the faults sent by the simulators to DMAAP") + events = dmaap.get_message_from_topic("unauthenticated.SEC_FAULT_OUTPUT", 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() + logger.info("Verify if faults have well been received for each device") + assert len(events) > 0 + faults_received = create_faults_structure(events) + + # Each device must have some faults + for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST: + logger.info("Check if %s has at least >=3 faults", sim_name) + assert sim_name in faults_received and faults_received[sim_name] >= 3