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

   1 import pytest
   2 import logging
   3 import logging.config
   4 from subprocess import check_output, run
   5 from waiting import wait
   6 import time
   7 from onapsdk.configuration import settings
   8 from oransdk.dmaap.dmaap import OranDmaap
   9 import os
  10 import json
  11 import requests
  12
  13 resources_path="./resources"
  14
  15 # Set working dir as python script location
  16 abspath = os.path.abspath(__file__)
  17 dname = os.path.dirname(abspath)
  18 os.chdir(dname)
  19
  20 logging.config.dictConfig(settings.LOG_CONFIG)
  21 logger = logging.getLogger("Test Session setup")
  22
  23
  24 def start_network_simulators():
  25         logger.info ("Clean up any network simulators")
  26         cmd="kubectl delete namespace network"
  27         run(cmd, shell=True, check=False)
  28         logger.info ("Start the network simulators")
  29         cmd="kubectl create namespace network"
  30         check_output(cmd, shell=True).decode('utf-8')
  31         cmd=f"helm install --debug oran-simulator local/ru-du-simulators --namespace network -f {resources_path}/network-simulators-topology/network-simulators-override.yaml -f {resources_path}/network-simulators-topology/network-simulators-topology-override.yaml"
  32         check_output(cmd, shell=True).decode('utf-8')
  33
  34 def get_all_simulators():
  35         dockerFilter = check_output("kubectl get services -n network -o name | awk -F \"/\" '{print $2}'", shell=True)
  36         return dockerFilter.splitlines()
  37
  38 def stop_network_simulators():
  39         cmd="kubectl delete namespace network"
  40         return check_output(cmd, shell=True).decode('utf-8')
  41
  42 def is_network_simulators_up():
  43         cmd="kubectl get pods --field-selector status.phase!=Running -n network"
  44         result=check_output(cmd, shell=True).decode('utf-8')
  45         logger.info (f"Checking if network simulators is UP:{result}")
  46         if "" == result:
  47                 logger.info ("Network sims is Up")
  48                 return True
  49         else:
  50                 logger.info ("Network sims is Down")
  51                 return False
  52
  53 def is_onap_up():
  54         cmd="kubectl get pods --field-selector status.phase!=Running -n onap | wc -l"
  55         result=check_output(cmd, shell=True).decode('utf-8')
  56         logger.info (f"Checking if ONAP is UP:{result}")
  57         if int(result) <= 9:
  58                 logger.info ("ONAP is Up")
  59                 return True
  60         else:
  61                 logger.info ("ONAP is Down")
  62                 return False
  63
  64 def is_nonrtric_up():
  65         cmd="kubectl get pods --field-selector status.phase!=Running -n nonrtric | wc -l"
  66         result=check_output(cmd, shell=True).decode('utf-8')
  67         logger.info (f"Checking if NONRTRIC is UP:{result}")
  68         if int(result) == 0:
  69                 logger.info ("NONRTRIC is Up")
  70                 return True
  71         else:
  72                 logger.info ("NONRTRIC is Down")
  73                 return False
  74
  75 def update_event_settings(nfName, nfType):
  76         file = f'{resources_path}/faults-config/event-settings-'+nfType+'.json'
  77         print ("File name:" + file)
  78         with open(file) as json_file:
  79                 body = json.load(json_file)
  80                 url = settings.SDNC_URL + '/rests/data/network-topology:network-topology/topology=topology-netconf/node=' + nfName + '/yang-ext:mount/nts-network-function:simulation/network-function'
  81                 print ("url:"+url)
  82                 headers = {
  83                         'content-type': 'application/yang-data+json',
  84                         'accept': 'application/yang-data+json',
  85                         'Authorization' : settings.SDNC_AUTH
  86                 }
  87                 try:
  88                         response = requests.put(url, verify=False, json=body, headers=headers)
  89                         print("Response:" + str(response))
  90                 except requests.exceptions.Timeout:
  91                         sys.exit('HTTP request failed, please check you internet connection.')
  92                 except requests.exceptions.TooManyRedirects:
  93                         sys.exit('HTTP request failed, please check your proxy settings.')
  94                 except requests.exceptions.RequestException as e:
  95                         raise SystemExit(e)
  96                 return response.status_code >= 200 and response.status_code < 300
  97
  98 def enable_events_for_all_simulators():
  99         for container in get_all_simulators():
 100                 name = container.decode("utf-8")
 101                 if "o-" in name:
 102                         if "o-ru" in name:
 103                                 print("Set", name, update_event_settings(name, "ru"))
 104                         if "o-du" in name:
 105                                 print("Set", name, update_event_settings(name, "du"))
 106
 107
 108 def wait_for_smo_to_be_running():
 109         wait(lambda: is_onap_up() and is_nonrtric_up(), sleep_seconds=10, timeout_seconds=300, waiting_for="SMO to be ready")
 110
 111 def wait_for_network_simulators_to_be_running():
 112         wait(lambda: is_network_simulators_up(), sleep_seconds=10, timeout_seconds=60, waiting_for="Network simulators to be ready")
 113
 114 def pytest_sessionstart(session):
 115         wait_for_smo_to_be_running()
 116         # Due to an Onap Ves bugs or dmaap ?? DU sims must send messages twice so we need to restart the sims
 117         start_network_simulators()
 118         wait_for_network_simulators_to_be_running()
 119         time.sleep(2)
 120         dmaap = OranDmaap()
 121         # Do a first get to register the o1test/o1test user in DMAAP, all messages will then be stored for him
 122         dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 10000, settings.DMAAP_GROUP, settings.DMAAP_USER)
 123
 124         ## Now kill the simulators and restart them for the test session
 125         stop_network_simulators()
 126
 127         start_network_simulators()
 128         wait_for_network_simulators_to_be_running()
 129         # Wait enough time to have at least the SDNR notifications sent
 130         logger.info ("Waiting 60s that SDNR sends all registration events to VES")
 131         time.sleep(60)
 132         logger.info ("Enabling faults/events reporting on SDNR")
 133         enable_events_for_all_simulators()
 134
 135         logger.info ("Test Session setup completed successfully")
 136
 137
 138 def pytest_sessionfinish(session, exitstatus):
 139         stop_network_simulators()
 140         logger.info ("Test Session cleanup done")
 141