-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
-from waiting import wait
-import time
-from onapsdk.configuration import settings
-from oransdk.dmaap.dmaap import OranDmaap
import os
-
-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__)
logging.config.dictConfig(settings.LOG_CONFIG)
logger = logging.getLogger("Test Session setup")
-
-def start_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-override.yaml -f {resources_path}/network-simulators-topology-override.yaml"
- check_output(cmd, shell=True).decode('utf-8')
-
-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 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
- time.sleep(30)
- 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.