50bb0ec0388bce4b763119f492566b81d3cd3442
[it/dep.git] / 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