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

   1 import pytest
   2 import logging
   3 import logging.config
   4 from subprocess import check_output
   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
  11 resources_path="./resources"
  12
  13 # Set working dir as python script location
  14 abspath = os.path.abspath(__file__)
  15 dname = os.path.dirname(abspath)
  16 os.chdir(dname)
  17
  18 logging.config.dictConfig(settings.LOG_CONFIG)
  19 logger = logging.getLogger("Test Session setup")
  20
  21
  22 def start_network_simulators():
  23         cmd="kubectl create namespace network"
  24         check_output(cmd, shell=True).decode('utf-8')
  25         cmd=f"helm install --debug oran-simulator local/ru-du-simulators --namespace network -f {resources_path}/network-simulators-override.yaml -f {resources_path}/network-simulators-topology-override.yaml"
  26         check_output(cmd, shell=True).decode('utf-8')
  27
  28 def stop_network_simulators():
  29         cmd="kubectl delete namespace network"
  30         return check_output(cmd, shell=True).decode('utf-8')
  31
  32 def is_network_simulators_up():
  33         cmd="kubectl get pods --field-selector status.phase!=Running -n network"
  34         result=check_output(cmd, shell=True).decode('utf-8')
  35         logger.info (f"Checking if network simulators is UP:{result}")
  36         if "" == result:
  37                 logger.info ("Network sims is Up")
  38                 return True
  39         else:
  40                 logger.info ("Network sims is Down")
  41                 return False
  42
  43 def is_onap_up():
  44         cmd="kubectl get pods --field-selector status.phase!=Running -n onap | wc -l"
  45         result=check_output(cmd, shell=True).decode('utf-8')
  46         logger.info (f"Checking if ONAP is UP:{result}")
  47         if int(result) <= 9:
  48                 logger.info ("ONAP is Up")
  49                 return True
  50         else:
  51                 logger.info ("ONAP is Down")
  52                 return False
  53
  54 def is_nonrtric_up():
  55         cmd="kubectl get pods --field-selector status.phase!=Running -n nonrtric | wc -l"
  56         result=check_output(cmd, shell=True).decode('utf-8')
  57         logger.info (f"Checking if NONRTRIC is UP:{result}")
  58         if int(result) == 0:
  59                 logger.info ("NONRTRIC is Up")
  60                 return True
  61         else:
  62                 logger.info ("NONRTRIC is Down")
  63                 return False
  64
  65 def wait_for_smo_to_be_running():
  66         wait(lambda: is_onap_up() and is_nonrtric_up(), sleep_seconds=10, timeout_seconds=300, waiting_for="SMO to be ready")
  67
  68 def wait_for_network_simulators_to_be_running():
  69         wait(lambda: is_network_simulators_up(), sleep_seconds=10, timeout_seconds=60, waiting_for="Network simulators to be ready")
  70
  71 def pytest_sessionstart(session):
  72         wait_for_smo_to_be_running()
  73         # Due to an Onap Ves bugs or dmaap ?? DU sims must send messages twice so we need to restart the sims
  74         start_network_simulators()
  75         wait_for_network_simulators_to_be_running()
  76         time.sleep(2)
  77         dmaap = OranDmaap()
  78         # Do a first get to register the o1test/o1test user in DMAAP, all messages will then be stored for him
  79         dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 10000, settings.DMAAP_GROUP, settings.DMAAP_USER)
  80
  81         ## Now kill the simulators and restart them for the test session
  82         stop_network_simulators()
  83
  84         start_network_simulators()
  85         wait_for_network_simulators_to_be_running()
  86         # Wait enough time to have at least the SDNR notifications sent
  87         time.sleep(30)
  88         logger.info ("Test Session setup completed successfully")
  89
  90
  91
  92 def pytest_sessionfinish(session, exitstatus):
  93         stop_network_simulators()
  94         logger.info ("Test Session cleanup done")
  95