smo-install/test/pythonsdk/src/orantests/test_o1.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2021-2022 AT&T Intellectual Property. All rights
   7 #                             reserved.
   8 # ================================================================================
   9 # Licensed under the Apache License, Version 2.0 (the "License");
  10 # you may not use this file except in compliance with the License.
  11 # You may obtain a copy of the License at
  12 #
  13 # http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS,
  17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 # See the License for the specific language governing permissions and
  19 # limitations under the License.
  20 # ============LICENSE_END============================================
  21 # ===================================================================
  22 #
  23 ###
  24 """O1 tests module."""
  25 import datetime
  26 import logging
  27 import json
  28 import os
  29 import time
  30 import pytest
  31 from onapsdk.configuration import settings
  32 from smo.network_simulators import NetworkSimulators
  33 from oransdk.dmaap.dmaap import OranDmaap
  34 from oransdk.sdnc.sdnc import OranSdnc
  35 from waiting import wait
  36
  37 # Set working dir as python script location
  38 abspath = os.path.abspath(__file__)
  39 dname = os.path.dirname(abspath)
  40 os.chdir(dname)
  41
  42 logging.config.dictConfig(settings.LOG_CONFIG)
  43 logger = logging.getLogger("Test O1")
  44
  45 network_simulators = NetworkSimulators("./resources")
  46 dmaap = OranDmaap()
  47 test_session_timestamp = datetime.datetime.now()
  48
  49 TOPIC_PNFREG = '{"topicName": "unauthenticated.VES_PNFREG_OUTPUT"}'
  50
  51 TOPIC_FAULT = '{"topicName": "unauthenticated.SEC_FAULT_OUTPUT"}'
  52
  53 @pytest.fixture(scope="module", autouse=True)
  54 def setup_simulators():
  55     """Setup the simulators before the executing the tests."""
  56     logger.info("Test class setup for O1 tests")
  57
  58     # Do a first get to register the o1test/o1test user in DMAAP
  59     # all registration messages will then be stored for the registration tests.
  60     # If it exists already it clears all cached events.
  61     dmaap.create_topic(TOPIC_PNFREG)
  62     dmaap.create_topic(TOPIC_FAULT)
  63     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")
  64     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")
  65     network_simulators.start_network_simulators()
  66     network_simulators.wait_for_network_simulators_to_be_running()
  67   # ADD DU RESTART just in case
  68     # Wait enough time to have at least the SDNR notifications sent
  69     logger.info("Waiting 20s that SDNR sends all registration events to VES...")
  70     time.sleep(10)
  71     logger.info("Enabling faults/events reporting on SDNR")
  72     network_simulators.enable_events_for_all_simulators()
  73 #    logger.info("Waiting 20s that the Dmaap faults topic is created...")
  74 #    time.sleep(20)
  75     # Preparing the DMaap to cache all the events for the fault topics.
  76     # If it exists already it clears all cached events.
  77     logger.info("Waiting 120s to have registration and faults events in DMaap")
  78     time.sleep(120)
  79     logger.info("Test Session setup completed successfully")
  80
  81     ### Cleanup code
  82     yield
  83     network_simulators.stop_network_simulators()
  84     logger.info("Test Session cleanup done")
  85
  86 def create_registration_structure(events):
  87     """Decode the registration events list."""
  88     devices_found_in_events = dict()
  89     for event in events:
  90         event_json = json.loads(event)
  91         logger.info("Registration json decoded: %s", str(event_json))
  92         devices_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = event_json["event"]["commonEventHeader"]["reportingEntityName"]
  93
  94     logger.info("Devices found in events:%s", devices_found_in_events)
  95     return devices_found_in_events
  96
  97 def create_faults_structure(events):
  98     """Decode the fault events list."""
  99     faults_found_in_events = dict()
 100     for event in events:
 101         event_json = json.loads(event)
 102         logger.info("Fault json decoded: %s", str(event_json))
 103         if event_json["event"]["commonEventHeader"]["sourceName"] in faults_found_in_events:
 104             faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] += 1
 105         else:
 106             faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = 1
 107     logger.info("Faults found in events: %s", faults_found_in_events)
 108     return faults_found_in_events
 109
 110 def test_devices_in_sdnc():
 111     """Verify that the devices are well defined in SDNC."""
 112     logger.info("Verify if devices are well in SDNC")
 113     for device in settings.NETWORK_SIMULATORS_LIST:
 114         logger.info("Verify if %s is well in SDNR", device)
 115         assert OranSdnc.get_devices(device, settings.SDNC_BASICAUTH) == 200
 116
 117 def validate_faults_timestamp(faults):
 118     """Extract only the faults returned by SDNC that have been raised during this test."""
 119     valid_faults = []
 120     for fault in faults['data-provider:output']['data']:
 121         converted_fault_timestamp = datetime.datetime.strptime(fault['timestamp'], "%Y-%m-%dT%H:%M:%S.%fZ")
 122         logger.info("Comparing fault timestamp %s (%s) to session test timestamp %s", converted_fault_timestamp, fault['timestamp'], test_session_timestamp)
 123         if converted_fault_timestamp > test_session_timestamp:
 124             valid_faults.append(fault)
 125     logger.info("Valid faults array: %s", valid_faults)
 126     return valid_faults
 127
 128 def test_device_faults_in_sdnc():
 129     """Verify that the device faults are well defined in SDNC."""
 130     logger.info("Verify is there is any events")
 131     for device in settings.NETWORK_SIMULATORS_DU_RU_LIST:
 132         faults = OranSdnc.get_events(settings.SDNC_BASICAUTH, device).json()
 133         logger.info("Verify if %s has events", device)
 134         assert len(validate_faults_timestamp(faults)) >= 3
 135
 136 def test_network_devices_registration_in_dmaap():
 137     """Validate that the devices are well registered in SDNR and forwarded to VES."""
 138     logger.info("Verify if SDNR sends well the RU registration to VES by checking in DMAAP")
 139     # As the user has been registered in DMAAP during test session init,
 140     # that call should return all sims registered by SDNR
 141     all_registrations = []
 142     events = []
 143
 144     while (events := dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()) != []:
 145         logger.info("Getting a first set of event: %s", events)
 146         all_registrations += events
 147
 148     logger.info("ALl registration events received: %s", all_registrations)
 149     # events should be a list of messages
 150     logger.info("Verify if the number of events is well >= to the number of expected devices")
 151     # The DU can send multiple times message to VES and SDNR can send multiple time event for RU
 152     assert len(all_registrations) >= (len(settings.NETWORK_SIMULATORS_DU_RU_LIST))
 153     devices_registered = create_registration_structure(all_registrations)
 154
 155     # Each device must be at least one time in the event list
 156     for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST:
 157         logger.info("Checking if %s is in events list", sim_name)
 158         assert sim_name in devices_registered
 159         if "o-ru" in sim_name:
 160             logger.info("RU event detected checking SDNR has well registered it")
 161             assert "ONAP SDN-R" in devices_registered[sim_name]
 162         elif "o-du" in sim_name:
 163             logger.info("DU detected checking it has well registered itself")
 164             assert "o-du" in devices_registered[sim_name]
 165
 166 def test_device_faults_in_dmaap():
 167     """Verify that device faults are well sent to DMAAP by SDNR."""
 168     logger.info("Verify if SDNR forwards well the faults sent by the simulators to DMAAP")
 169     events = dmaap.get_message_from_topic("unauthenticated.SEC_FAULT_OUTPUT", 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()
 170     logger.info("Verify if faults have well been received for each device")
 171     assert len(events) > 0
 172     faults_received = create_faults_structure(events)
 173
 174     # Each device must have some faults
 175     for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST:
 176         logger.info("Check if %s has at least >=3 faults", sim_name)
 177         assert sim_name in faults_received and faults_received[sim_name] >= 3