4 from subprocess import check_output
5 from waiting import wait
7 from onapsdk.configuration import settings
8 from oransdk.dmaap.dmaap import OranDmaap
11 resources_path="./resources"
13 # Set working dir as python script location
14 abspath = os.path.abspath(__file__)
15 dname = os.path.dirname(abspath)
18 logging.config.dictConfig(settings.LOG_CONFIG)
19 logger = logging.getLogger("Test Session setup")
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')
28 def stop_network_simulators():
29 cmd="kubectl delete namespace network"
30 return check_output(cmd, shell=True).decode('utf-8')
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}")
37 logger.info ("Network sims is Up")
40 logger.info ("Network sims is Down")
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}")
48 logger.info ("ONAP is Up")
51 logger.info ("ONAP is Down")
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}")
59 logger.info ("NONRTRIC is Up")
62 logger.info ("NONRTRIC is Down")
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")
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")
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()
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)
81 ## Now kill the simulators and restart them for the test session
82 stop_network_simulators()
84 start_network_simulators()
85 wait_for_network_simulators_to_be_running()
86 # Wait enough time to have at least the SDNR notifications sent
88 logger.info ("Test Session setup completed successfully")
92 def pytest_sessionfinish(session, exitstatus):
93 stop_network_simulators()
94 logger.info ("Test Session cleanup done")