-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
+import os
from onapsdk.configuration import settings
+from waiting import wait
from oransdk.dmaap.dmaap import OranDmaap
-import os
-
-resources_path="./resources"
+from oransdk.policy.policy import OranPolicy
+from oransdk.sdnc.sdnc import OranSdnc
+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()
+policy = OranPolicy()
+
+def policy_component_ready():
+ """Check if components are ready."""
+ logger.info("Verify policy components are ready")
+ policy_ready = {"api_ready": False, "pap_ready": False, "apex_ready": False}
+ policy_status = policy.get_components_status(settings.POLICY_BASICAUTH)
+ if (policy_status["api"]["healthy"] and not policy_ready["api_ready"]):
+ logger.info("Policy Api is ready")
+ policy_ready["api_ready"] = True
+ if (policy_status["pap"]["healthy"] and not policy_ready["pap_ready"]):
+ logger.info("Policy Pap is ready")
+ policy_ready["pap_ready"] = True
+ if (policy_status["pdps"]["apex"][0]["healthy"] == "HEALTHY" and not policy_ready["apex_ready"]):
+ logger.info("Policy Apex is ready")
+ policy_ready["apex_ready"] = True
+ return policy_ready["api_ready"] and policy_ready["pap_ready"] and policy_ready["apex_ready"]
+
+
+###### 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")
+ wait(lambda: policy_component_ready(), sleep_seconds=10, timeout_seconds=300, waiting_for="Policy to be ready")
+ ## Just kill any simulators that could already be runnin
+ network_sims.stop_network_simulators()
+ ###### END of FIRST start, now we can start the sims for the real tests.
+ logger.info("Tests session setup is ready")
Code Review / it / dep.git / blobdiff