smo-install/test/pythonsdk/src/orantests/test_cl_k8s.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 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 """Closed Loop Apex usecase tests module."""
  25 # This usecase has limitations due to Clamp issue.
  26 # 1. make sure using the policy-clamp-be version 6.2.0-snapshot-latest at this the moment
  27 import time
  28 import logging.config
  29 import subprocess
  30 import os
  31 from subprocess import check_output
  32 import pytest
  33 from waiting import wait
  34 from onapsdk.configuration import settings
  35 from oransdk.utils.jinja import jinja_env
  36 from oransdk.policy.clamp import ClampToscaTemplate
  37 from smo.cl_usecase import ClCommissioningUtils
  38
  39 # Set working dir as python script location
  40 abspath = os.path.abspath(__file__)
  41 dname = os.path.dirname(abspath)
  42 os.chdir(dname)
  43
  44 logging.config.dictConfig(settings.LOG_CONFIG)
  45 logger = logging.getLogger("test Control Loops for Clamp K8S usecase")
  46 clcommissioning_utils = ClCommissioningUtils()
  47 clamp = ClampToscaTemplate(settings.CLAMP_BASICAUTH)
  48
  49 chartmuseum_port = "8080"
  50 usecase_name = "script_usecase"
  51 chartmuseum_ip = "http://test-chartmuseum.test:8080"
  52 global app_name
  53
  54 @pytest.fixture(autouse=True)
  55 def setup_simulators(request):
  56     """Prepare the test environment before the executing the tests."""
  57     logger.info("Test class setup for Closed Loop tests of %s", request.node.name)
  58
  59     deploy_chartmuseum()
  60
  61     # Add the remote repo to Clamp k8s pod
  62     logger.info("Add the remote repo to Clamp k8s pod")
  63     k8s_pod = subprocess.run("kubectl get pods -n onap | grep k8s | awk '{print $1}'", shell=True, check=True, stdout=subprocess.PIPE).stdout.decode('utf-8').strip()
  64
  65     repo_url = subprocess.run("kubectl get services -n test | grep test-chartmuseum | awk '{print $3}'", shell=True, check=True, stdout=subprocess.PIPE).stdout.decode('utf-8').strip()+":8080"
  66     logger.info("k8s: %s, repo_url:%s", k8s_pod, repo_url)
  67     cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm repo add chartmuseum http://{repo_url}\""
  68     check_output(cmd, shell=True).decode('utf-8')
  69     cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm repo update\""
  70     check_output(cmd, shell=True).decode('utf-8')
  71     cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm search repo -l oru-app\""
  72     result = check_output(cmd, shell=True).decode('utf-8')
  73     if result == '':
  74         logger.info("Failed to update the K8s pod repo")
  75     logger.info("Test Session setup completed successfully for %s", request.node.name)
  76
  77     ### Cleanup code
  78     yield
  79     # Finish and delete the cl instance
  80     clcommissioning_utils.clean_instance(usecase_name)
  81     wait(lambda: is_rapp_down(app_name), sleep_seconds=5, timeout_seconds=60, waiting_for="Rapp is down")
  82     # Remove the remote repo to Clamp k8s pod
  83     cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm repo remove chartmuseum\""
  84     check_output(cmd, shell=True).decode('utf-8')
  85     cmd = "kubectl delete namespace test"
  86     check_output(cmd, shell=True).decode('utf-8')
  87     cmd = "helm repo remove test"
  88     check_output(cmd, shell=True).decode('utf-8')
  89     time.sleep(10)
  90     logger.info("Test Session cleanup done for %s", request.node.name)
  91
  92
  93 def deploy_chartmuseum():
  94     """Start chartmuseum pod and populate with the nedded helm chart."""
  95     logger.info("Start to deploy chartmuseum")
  96     cmd = "helm repo add test https://chartmuseum.github.io/charts"
  97     check_output(cmd, shell=True).decode('utf-8')
  98     cmd = "kubectl create namespace test"
  99     check_output(cmd, shell=True).decode('utf-8')
 100
 101     cmd = "helm install test test/chartmuseum --version 3.1.0 --namespace test --set env.open.DISABLE_API=false"
 102     check_output(cmd, shell=True).decode('utf-8')
 103     wait(lambda: is_chartmuseum_up(), sleep_seconds=10, timeout_seconds=60, waiting_for="chartmuseum to be ready")
 104
 105     time.sleep(10)
 106     chartmuseum_url = subprocess.run("kubectl get services -n test | grep test-chartmuseum | awk '{print $3}'", shell=True, check=True, stdout=subprocess.PIPE).stdout.decode('utf-8').strip()+":8080"
 107     cmd = f"curl -X POST --data-binary @{dname}/resources/cl-test-helm-chart/oru-app-1.0.0.tgz http://{chartmuseum_url}/api/charts"
 108     check_output(cmd, shell=True).decode('utf-8')
 109     cmd = f"curl -X POST --data-binary @{dname}/resources/cl-test-helm-chart/odu-app-1.0.0.tgz http://{chartmuseum_url}/api/charts"
 110     check_output(cmd, shell=True).decode('utf-8')
 111     cmd = f"curl -X POST --data-binary @{dname}/resources/cl-test-helm-chart/odu-app-ics-version-1.0.0.tgz http://{chartmuseum_url}/api/charts"
 112     check_output(cmd, shell=True).decode('utf-8')
 113
 114
 115 def is_chartmuseum_up() -> bool:
 116     """Check if the chartmuseum is up."""
 117     cmd = "kubectl get pods --field-selector status.phase=Running -n test"
 118     result = check_output(cmd, shell=True).decode('utf-8')
 119     logger.info("Checking if chartmuseum is UP: %s", result)
 120     if result == '':
 121         logger.info("chartmuseum is Down")
 122         return False
 123     logger.info("chartmuseum is Up")
 124     return True
 125
 126
 127 def is_rapp_up(appname) -> bool:
 128     """Check if the rapp is up."""
 129     cmd = "kubectl get pods -n nonrtric | grep " + appname + " | wc -l"
 130     result = check_output(cmd, shell=True).decode('utf-8')
 131     logger.info("Checking if %s is up :%s", appname, result)
 132     if int(result) == 1:
 133         logger.info("%s is Up", appname.upper())
 134         return True
 135     logger.info("%s is Down", appname.upper())
 136     return False
 137
 138 def is_rapp_down(appname) -> bool:
 139     """Check if the rapp is down."""
 140     cmd = "kubectl get pods -n nonrtric | grep " + appname + " | wc -l"
 141     result = check_output(cmd, shell=True).decode('utf-8')
 142     logger.info("Checking if %s is down :%s", appname, result)
 143     if int(result) == 0:
 144         logger.info("%s is Down", appname.upper())
 145         return True
 146     logger.info("%s is Up", appname.upper())
 147     return False
 148
 149 def test_cl_oru_app_deploy():
 150     chart_version = "1.0.0"
 151     chart_name = "oru-app"
 152     release_name = "oru-app"
 153     global app_name
 154     app_name = chart_name
 155     """The Closed Loop O-RU Fronthaul Recovery usecase Apex version."""
 156     logger.info("Upload tosca to commissioning")
 157     commissioning_payload = jinja_env().get_template("commission_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartmuseumPort=chartmuseum_port, chartVersion=chart_version, chartName=chart_name, releaseName=release_name)
 158     instance_payload = jinja_env().get_template("create_instance_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartmuseumPort=chartmuseum_port, chartVersion=chart_version, chartName=chart_name, releaseName=release_name, instanceName=usecase_name)
 159     assert clcommissioning_utils.create_instance(usecase_name, commissioning_payload, instance_payload) is True
 160
 161     logger.info("Check if oru-app is up")
 162     wait(lambda: is_rapp_up(chart_name), sleep_seconds=5, timeout_seconds=300, waiting_for="Oru app to be up")
 163
 164 def test_cl_odu_app_smo_deploy():
 165     chart_version = "1.0.0"
 166     chart_name = "odu-app"
 167     release_name = "odu-app"
 168     global app_name
 169     app_name = chart_name
 170     """The O-DU Slice Assurance SMO Version use case."""
 171     logger.info("Upload tosca to commissioning")
 172     commissioning_payload = jinja_env().get_template("commission_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartmuseumPort=chartmuseum_port, chartVersion=chart_version, chartName=chart_name, releaseName=release_name)
 173     instance_payload = jinja_env().get_template("create_instance_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartmuseumPort=chartmuseum_port, chartVersion=chart_version, chartName=chart_name, releaseName=release_name, instanceName=usecase_name)
 174     assert clcommissioning_utils.create_instance(usecase_name, commissioning_payload, instance_payload) is True
 175
 176     logger.info("Check if odu-app smo version is up")
 177     wait(lambda: is_rapp_up(chart_name), sleep_seconds=5, timeout_seconds=300, waiting_for="Odu app smo version to be up")
 178
 179 def test_cl_odu_app_ics_deploy():
 180     chart_version = "1.0.0"
 181     chart_name = "odu-app-ics-version"
 182     release_name = "odu-app-ics-version"
 183     global app_name
 184     app_name = chart_name
 185     """The O-DU Slice Assurance ICS Version use case."""
 186     logger.info("Upload tosca to commissioning")
 187     commissioning_payload = jinja_env().get_template("commission_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartmuseumPort=chartmuseum_port, chartVersion=chart_version, chartName=chart_name, releaseName=release_name)
 188     instance_payload = jinja_env().get_template("create_instance_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartmuseumPort=chartmuseum_port, chartVersion=chart_version, chartName=chart_name, releaseName=release_name, instanceName=usecase_name)
 189     assert clcommissioning_utils.create_instance(usecase_name, commissioning_payload, instance_payload) is True
 190
 191     logger.info("Check if odu-app ics version is up")
 192     wait(lambda: is_rapp_up(chart_name), sleep_seconds=5, timeout_seconds=300, waiting_for="Odu app ics version to be up")