X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=smo-install%2Ftest%2Fpythonsdk%2Fsrc%2Forantests%2Fconftest.py;h=2e78b5ba488e8d6fe459f05e9f1f7d02bb7257d2;hb=6c7e3c3c5e51182a891a2a21676f1cc2646ba484;hp=50bb0ec0388bce4b763119f492566b81d3cd3442;hpb=8b713ed09d061baacf624a533be3351e7dcb0500;p=it%2Fdep.git diff --git a/smo-install/test/pythonsdk/src/orantests/conftest.py b/smo-install/test/pythonsdk/src/orantests/conftest.py index 50bb0ec0..2e78b5ba 100644 --- a/smo-install/test/pythonsdk/src/orantests/conftest.py +++ b/smo-install/test/pythonsdk/src/orantests/conftest.py @@ -1,16 +1,36 @@ -import pytest +#!/usr/bin/env python3 +### +# ============LICENSE_START======================================================= +# ORAN SMO PACKAGE - PYTHONSDK TESTS +# ================================================================================ +# Copyright (C) 2021-2022 AT&T Intellectual Property. All rights +# reserved. +# ================================================================================ +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ============LICENSE_END============================================ +# =================================================================== +# +### +"""Module called by pytest.""" import logging import logging.config -from subprocess import check_output, run -from waiting import wait -import time -from onapsdk.configuration import settings -from oransdk.dmaap.dmaap import OranDmaap import os -import json -import requests - -resources_path="./resources" +from oransdk.dmaap.dmaap import OranDmaap +from oransdk.sdnc.sdnc import OranSdnc +from onapsdk.configuration import settings +from waiting import wait +from smo.smo import Smo +from smo.network_simulators import NetworkSimulators # Set working dir as python script location abspath = os.path.abspath(__file__) @@ -20,122 +40,58 @@ os.chdir(dname) logging.config.dictConfig(settings.LOG_CONFIG) logger = logging.getLogger("Test Session setup") - -def start_network_simulators(): - logger.info ("Clean up any network simulators") - cmd="kubectl delete namespace network" - run(cmd, shell=True, check=False) - logger.info ("Start the network simulators") - cmd="kubectl create namespace network" - check_output(cmd, shell=True).decode('utf-8') - 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" - check_output(cmd, shell=True).decode('utf-8') - -def get_all_simulators(): - dockerFilter = check_output("kubectl get services -n network -o name | awk -F \"/\" '{print $2}'", shell=True) - return dockerFilter.splitlines() - -def stop_network_simulators(): - cmd="kubectl delete namespace network" - return check_output(cmd, shell=True).decode('utf-8') - -def is_network_simulators_up(): - cmd="kubectl get pods --field-selector status.phase!=Running -n network" - result=check_output(cmd, shell=True).decode('utf-8') - logger.info (f"Checking if network simulators is UP:{result}") - if "" == result: - logger.info ("Network sims is Up") - return True - else: - logger.info ("Network sims is Down") - return False - -def is_onap_up(): - cmd="kubectl get pods --field-selector status.phase!=Running -n onap | wc -l" - result=check_output(cmd, shell=True).decode('utf-8') - logger.info (f"Checking if ONAP is UP:{result}") - if int(result) <= 9: - logger.info ("ONAP is Up") - return True - else: - logger.info ("ONAP is Down") - return False - -def is_nonrtric_up(): - cmd="kubectl get pods --field-selector status.phase!=Running -n nonrtric | wc -l" - result=check_output(cmd, shell=True).decode('utf-8') - logger.info (f"Checking if NONRTRIC is UP:{result}") - if int(result) == 0: - logger.info ("NONRTRIC is Up") - return True - else: - logger.info ("NONRTRIC is Down") - return False - -def update_event_settings(nfName, nfType): - file = f'{resources_path}/faults-config/event-settings-'+nfType+'.json' - print ("File name:" + file) - with open(file) as json_file: - body = json.load(json_file) - url = settings.SDNC_URL + '/rests/data/network-topology:network-topology/topology=topology-netconf/node=' + nfName + '/yang-ext:mount/nts-network-function:simulation/network-function' - print ("url:"+url) - headers = { - 'content-type': 'application/yang-data+json', - 'accept': 'application/yang-data+json', - 'Authorization' : settings.SDNC_AUTH - } - try: - response = requests.put(url, verify=False, json=body, headers=headers) - print("Response:" + str(response)) - except requests.exceptions.Timeout: - sys.exit('HTTP request failed, please check you internet connection.') - except requests.exceptions.TooManyRedirects: - sys.exit('HTTP request failed, please check your proxy settings.') - except requests.exceptions.RequestException as e: - raise SystemExit(e) - return response.status_code >= 200 and response.status_code < 300 - -def enable_events_for_all_simulators(): - for container in get_all_simulators(): - name = container.decode("utf-8") - if "o-" in name: - if "o-ru" in name: - print("Set", name, update_event_settings(name, "ru")) - if "o-du" in name: - print("Set", name, update_event_settings(name, "du")) - - -def wait_for_smo_to_be_running(): - wait(lambda: is_onap_up() and is_nonrtric_up(), sleep_seconds=10, timeout_seconds=300, waiting_for="SMO to be ready") - -def wait_for_network_simulators_to_be_running(): - wait(lambda: is_network_simulators_up(), sleep_seconds=10, timeout_seconds=60, waiting_for="Network simulators to be ready") - -def pytest_sessionstart(session): - wait_for_smo_to_be_running() - # Due to an Onap Ves bugs or dmaap ?? DU sims must send messages twice so we need to restart the sims - start_network_simulators() - wait_for_network_simulators_to_be_running() - time.sleep(2) - dmaap = OranDmaap() - # Do a first get to register the o1test/o1test user in DMAAP, all messages will then be stored for him - dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 10000, settings.DMAAP_GROUP, settings.DMAAP_USER) - - ## Now kill the simulators and restart them for the test session - stop_network_simulators() - - start_network_simulators() - wait_for_network_simulators_to_be_running() - # Wait enough time to have at least the SDNR notifications sent - logger.info ("Waiting 60s that SDNR sends all registration events to VES") - time.sleep(60) - logger.info ("Enabling faults/events reporting on SDNR") - enable_events_for_all_simulators() - - logger.info ("Test Session setup completed successfully") - - -def pytest_sessionfinish(session, exitstatus): - stop_network_simulators() - logger.info ("Test Session cleanup done") - +network_sims = NetworkSimulators("./resources") +smo = Smo() +dmaap = OranDmaap() +sdnc = OranSdnc() + +TOPIC_PNFREG = '{"topicName": "unauthenticated.VES_PNFREG_OUTPUT"}' + +TOPIC_FAULT = '{"topicName": "unauthenticated.SEC_FAULT_OUTPUT"}' + +#TOPIC_PNFREG = { +# "owner": "", +# "readerAcl": { +# "enabled": "true", +# "users": [] +# }, +# "name": "unauthenticated.VES_PNFREG_OUTPUT", +# "description": "", +# "writerAcl": { +# "enabled": "true", +# "users": [] +# } +#} + +#TOPIC_FAULT = { +# "owner": "", +# "readerAcl": { +# "enabled": "true", +# "users": [] +# }, +# "name": "unauthenticated.SEC_FAULT_OUTPUT", +# "description": "", +# "writerAcl": { +# "enabled": "true", +# "users": [] +# } +#} + +###### Entry points of PYTEST Session +def pytest_sessionstart(): + """Pytest calls it when starting a test session.""" + logger.info("Check and wait for SMO to be running") + smo.wait_for_smo_to_be_running() + logger.info("Check and for for SDNC to be running") + wait(lambda: OranSdnc.get_events(settings.SDNC_BASICAUTH, "test").status_code == 200, sleep_seconds=10, timeout_seconds=300, waiting_for="SDNC to be ready") + + dmaap.create_topic(TOPIC_PNFREG) + dmaap.create_topic(TOPIC_FAULT) + ### Due to an Onap Ves/dmaap behavior !!! DU sims must send messages + ### twice so we need to create/delete the sims +# network_sims.start_network_simulators() +# network_sims.wait_for_network_simulators_to_be_running() +# time.sleep(20) + ## Now kill the simulators and restart them for the test session + network_sims.stop_network_simulators() + ###### END of FIRST start, now we can start the sims for the real tests.