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 waiting import wait
  33 from smo.network_simulators import NetworkSimulators
  34 from oransdk.dmaap.dmaap import OranDmaap
  35 from oransdk.sdnc.sdnc import OranSdnc
  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
  50 @pytest.fixture(scope="module", autouse=True)
  51 def setup_simulators():
  52     """Setup the simulators before the executing the tests."""
  53     logger.info("Test class setup for O1 tests")
  54
  55     # Do a first get to register the o1test/o1test user in DMAAP
  56     # all registration messages will then be stored for the registration tests.
  57     # If it exists already it clears all cached events.
  58     dmaap.create_topic(settings.DMAAP_TOPIC_PNFREG_JSON)
  59     dmaap.create_topic(settings.DMAAP_TOPIC_FAULT_JSON)
  60     wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_PNFREG, 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")
  61     wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_FAULT, 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")
  62     network_simulators.start_network_simulators()
  63     network_simulators.wait_for_network_simulators_to_be_running()
  64   # ADD DU RESTART just in case
  65     # Wait enough time to have at least the SDNR notifications sent
  66     logger.info("Waiting 20s that SDNR sends all registration events to VES...")
  67     time.sleep(20)
  68     logger.info("Enabling faults/events reporting on SDNR")
  69     network_simulators.enable_events_for_all_simulators()
  70 #    logger.info("Waiting 20s that the Dmaap faults topic is created...")
  71 #    time.sleep(20)
  72     # Preparing the DMaap to cache all the events for the fault topics.
  73     # If it exists already it clears all cached events.
  74     logger.info("Waiting 120s to have registration and faults events in DMaap")
  75     time.sleep(120)
  76     logger.info("Test Session setup completed successfully")
  77
  78     ### Cleanup code
  79     yield
  80     network_simulators.stop_network_simulators()
  81     logger.info("Test Session cleanup done")
  82
  83 def create_registration_structure(events):
  84     """Decode the registration events list."""
  85     devices_found_in_events = dict()
  86     for event in events:
  87         event_json = json.loads(event)
  88         logger.info("Registration json decoded: %s", str(event_json))
  89         devices_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = event_json["event"]["commonEventHeader"]["reportingEntityName"]
  90
  91     logger.info("Devices found in events:%s", devices_found_in_events)
  92     return devices_found_in_events
  93
  94 def create_faults_structure(events):
  95     """Decode the fault events list."""
  96     faults_found_in_events = dict()
  97     for event in events:
  98         event_json = json.loads(event)
  99         logger.info("Fault json decoded: %s", str(event_json))
 100         if event_json["event"]["commonEventHeader"]["sourceName"] in faults_found_in_events:
 101             faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] += 1
 102         else:
 103             faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = 1
 104     logger.info("Faults found in events: %s", faults_found_in_events)
 105     return faults_found_in_events
 106
 107 def test_devices_in_sdnc():
 108     """Verify that the devices are well defined in SDNC."""
 109     logger.info("Verify if devices are well in SDNC")
 110     for device in settings.NETWORK_SIMULATORS_LIST:
 111         logger.info("Verify if %s is well in SDNR", device)
 112         assert OranSdnc.get_devices(device, settings.SDNC_BASICAUTH) == 200
 113
 114 def validate_faults_timestamp(faults):
 115     """Extract only the faults returned by SDNC that have been raised during this test."""
 116     valid_faults = []
 117     for fault in faults['data-provider:output']['data']:
 118         converted_fault_timestamp = datetime.datetime.strptime(fault['timestamp'], "%Y-%m-%dT%H:%M:%S.%fZ")
 119         logger.info("Comparing fault timestamp %s (%s) to session test timestamp %s", converted_fault_timestamp, fault['timestamp'], test_session_timestamp)
 120         if converted_fault_timestamp > test_session_timestamp:
 121             valid_faults.append(fault)
 122     logger.info("Valid faults array: %s", valid_faults)
 123     return valid_faults
 124
 125 def test_device_faults_in_sdnc():
 126     """Verify that the device faults are well defined in SDNC."""
 127     logger.info("Verify is there is any events")
 128     for device in settings.NETWORK_SIMULATORS_DU_RU_LIST:
 129         faults = OranSdnc.get_events(settings.SDNC_BASICAUTH, device).json()
 130         logger.info("Verify if %s has events", device)
 131         assert len(validate_faults_timestamp(faults)) >= 3
 132
 133 def test_network_devices_registration_in_dmaap():
 134     """Validate that the devices are well registered in SDNR and forwarded to VES."""
 135     logger.info("Verify if SDNR sends well the RU registration to VES by checking in DMAAP")
 136     # As the user has been registered in DMAAP during test session init,
 137     # that call should return all sims registered by SDNR
 138     all_registrations = []
 139     events = []
 140
 141     while (events := dmaap.get_message_from_topic(settings.DMAAP_TOPIC_PNFREG, 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()) != []:
 142         logger.info("Getting a first set of event: %s", events)
 143         all_registrations += events
 144
 145     logger.info("ALl registration events received: %s", all_registrations)
 146     # events should be a list of messages
 147     logger.info("Verify if the number of events is well >= to the number of expected devices")
 148     # The DU can send multiple times message to VES and SDNR can send multiple time event for RU
 149     assert len(all_registrations) >= (len(settings.NETWORK_SIMULATORS_DU_RU_LIST))
 150     devices_registered = create_registration_structure(all_registrations)
 151
 152     # Each device must be at least one time in the event list
 153     for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST:
 154         logger.info("Checking if %s is in events list", sim_name)
 155         assert sim_name in devices_registered
 156         if "o-ru" in sim_name:
 157             logger.info("RU event detected checking SDNR has well registered it")
 158             assert "ONAP SDN-R" in devices_registered[sim_name]
 159         elif "o-du" in sim_name:
 160             logger.info("DU detected checking it has well registered itself")
 161             assert "o-du" in devices_registered[sim_name]
 162
 163 def test_device_faults_in_dmaap():
 164     """Verify that device faults are well sent to DMAAP by SDNR."""
 165     logger.info("Verify if SDNR forwards well the faults sent by the simulators to DMAAP")
 166     events = dmaap.get_message_from_topic(settings.DMAAP_TOPIC_FAULT, 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()
 167     logger.info("Verify if faults have well been received for each device")
 168     assert len(events) > 0
 169     faults_received = create_faults_structure(events)
 170
 171     # Each device must have some faults
 172     for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST:
 173         logger.info("Check if %s has at least >=3 faults", sim_name)
 174         assert sim_name in faults_received and faults_received[sim_name] >= 3