3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 2021-2022 AT&T Intellectual Property. All rights
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
13 # http://www.apache.org/licenses/LICENSE-2.0
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 # ===================================================================
25 """NetworkSimulators module."""
29 from subprocess import check_output, run
32 from onapsdk.configuration import settings
33 from waiting import wait
35 logging.config.dictConfig(settings.LOG_CONFIG)
36 logger = logging.getLogger("Network Simulators k8s")
38 class NetworkSimulators():
39 """Network simulators controls the simulators k8s deployment."""
43 def __init__(self, resources_dir):
44 """Create NetworkSimulators class."""
45 self.resources_path = resources_dir
47 def start_network_simulators(self):
48 """Start all simulators defined in resources_path."""
49 logger.info("Start the network simulators")
50 cmd = "kubectl create namespace network"
51 check_output(cmd, shell=True).decode('utf-8')
52 cmd = f"helm install --debug oran-simulator local/ru-du-simulators --namespace network -f {self.resources_path}/network-simulators-topology/network-simulators-override.yaml -f {self.resources_path}/network-simulators-topology/network-simulators-topology-override.yaml"
53 check_output(cmd, shell=True).decode('utf-8')
55 def start_and_wait_network_simulators(self):
56 """Start and wait for all simulators defined in resources_path."""
57 logger.info("Start the network simulators")
58 self.start_network_simulators()
59 NetworkSimulators.wait_for_network_simulators_to_be_running()
62 def get_all_simulators():
63 """Retrieve all simulators defined in k8s services."""
64 dockerFilter = check_output("kubectl get services -n network -o name | awk -F \"/\" '{print $2}'", shell=True)
65 return dockerFilter.splitlines()
68 def stop_network_simulators():
69 """Stop the simulators."""
70 logger.info("Clean up any network simulators")
71 cmd = "kubectl delete namespace network"
72 run(cmd, shell=True, check=False)
75 def is_network_simulators_up() -> bool:
76 """Check if the network simulators are up."""
77 cmd = "kubectl get pods --field-selector status.phase!=Running -n network"
78 result = check_output(cmd, shell=True).decode('utf-8')
79 logger.info("Checking if network simulators is UP: %s", result)
81 logger.info("Network sims is Up")
83 logger.info("Network sims is Down")
86 def update_event_settings(self, nfName, nfType):
87 """Send one event for specific simulator of a specific type."""
88 file = f'{self.resources_path}/faults-config/event-settings-'+nfType+'.json'
89 logger.info("Faults parameters path: %s", file)
90 with open(file) as json_file:
91 body = json.load(json_file)
92 url = settings.SDNC_URL + '/rests/data/network-topology:network-topology/topology=topology-netconf/node=' + nfName + '/yang-ext:mount/nts-network-function:simulation/network-function'
93 logger.info("Using SDNC URL: %s", url)
95 'content-type': 'application/yang-data+json',
96 'accept': 'application/yang-data+json',
97 'Authorization' : settings.SDNC_AUTH
100 response = requests.put(url, verify=False, json=body, headers=headers)
101 logger.info("Response: %s", str(response))
102 except requests.exceptions.Timeout:
103 sys.exit('HTTP request failed, please check you internet connection.')
104 except requests.exceptions.TooManyRedirects:
105 sys.exit('HTTP request failed, please check your proxy settings.')
106 except requests.exceptions.RequestException as e:
108 return response.status_code >= 200 and response.status_code < 300
110 def enable_events_for_all_simulators(self):
111 """Send event to sdnc for all sim containers."""
112 for container in NetworkSimulators.get_all_simulators():
113 name = container.decode("utf-8")
116 logger.info("Set %s %s", name, self.update_event_settings(name, "ru"))
118 logger.info("Set %s %s", name, self.update_event_settings(name, "du"))
121 def wait_for_network_simulators_to_be_running():
122 """Check and wait for the network sims to be running."""
123 wait(lambda: NetworkSimulators.is_network_simulators_up(), sleep_seconds=settings.NETWORK_SIMULATOR_CHECK_RETRY, timeout_seconds=settings.NETWORK_SIMULATOR_CHECK_TIMEOUT, waiting_for="Network simulators to be ready")