smo-install/test/pythonsdk/src/orantests/oran_tests/smo/network_simulators.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2021-2022 AT&T Intellectual Property. All rights
   7 #                             reserved.
   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
  12 #
  13 # http://www.apache.org/licenses/LICENSE-2.0
  14 #
  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 # ===================================================================
  22 #
  23 ###
  24
  25 """NetworkSimulators module."""
  26 import logging
  27 import logging.config
  28 import json
  29 from subprocess import check_output, run
  30 import sys
  31 import requests
  32 from onapsdk.configuration import settings
  33 from waiting import wait
  34
  35 logging.config.dictConfig(settings.LOG_CONFIG)
  36 logger = logging.getLogger("Network Simulators k8s")
  37
  38 class NetworkSimulators():
  39     """Network simulators controls the simulators k8s deployment."""
  40
  41     resources_path = ""
  42
  43     def __init__(self, resources_dir):
  44         """Create NetworkSimulators class."""
  45         self.resources_path = resources_dir
  46
  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')
  54
  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()
  60
  61     @staticmethod
  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()
  66
  67     @staticmethod
  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)
  73
  74     @staticmethod
  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)
  80         if result == '':
  81             logger.info("Network sims is Up")
  82             return True
  83         logger.info("Network sims is Down")
  84         return False
  85
  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)
  94             headers = {
  95                 'content-type': 'application/yang-data+json',
  96                 'accept': 'application/yang-data+json',
  97                 'Authorization' : settings.SDNC_AUTH
  98             }
  99             try:
 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:
 107                 raise SystemExit(e)
 108             return response.status_code >= 200 and response.status_code < 300
 109
 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")
 114             if "o-" in name:
 115                 if "o-ru" in name:
 116                     logger.info("Set %s %s", name, self.update_event_settings(name, "ru"))
 117                 if "o-du" in name:
 118                     logger.info("Set %s %s", name, self.update_event_settings(name, "du"))
 119
 120     @staticmethod
 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")