3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 2022 AT&T Intellectual Property. All rights
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
13 # http://www.apache.org/licenses/LICENSE-2.0
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 # ===================================================================
24 """Closed Loop Apex usecase tests module."""
25 # This usecase has limitations due to Clamp issue.
26 # 1. make sure using the policy-k8s-participant version is higher than 6.3.0
31 from subprocess import check_output
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
39 # Set working dir as python script location
40 abspath = os.path.abspath(__file__)
41 dname = os.path.dirname(abspath)
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)
49 usecase_name = "script_usecase"
50 chartmuseum_ip = "http://test-chartmuseum.test:8080"
52 @pytest.fixture(autouse=True)
53 def setup_simulators(request):
54 """Prepare the test environment before the executing the tests."""
55 logger.info("Test class setup for Closed Loop tests of %s", request.node.name)
59 # Add the remote repo to Clamp k8s pod
60 logger.info("Add the remote repo to Clamp k8s pod")
61 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()
63 cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm repo add chartmuseum http://test-chartmuseum.test:8080\""
64 check_output(cmd, shell=True).decode('utf-8')
65 cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm repo update\""
66 check_output(cmd, shell=True).decode('utf-8')
67 cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm search repo -l oru-app\""
68 result = check_output(cmd, shell=True).decode('utf-8')
70 logger.info("Failed to update the K8s pod repo")
71 logger.info("Test Session setup completed successfully for %s", request.node.name)
75 # Finish and delete the cl instance
76 clcommissioning_utils.clean_instance(usecase_name)
77 wait(lambda: is_rapp_down(pytest.app_name), sleep_seconds=5, timeout_seconds=60, waiting_for="Rapp is down")
78 # Remove the remote repo to Clamp k8s pod
79 cmd = f"kubectl exec -it -n onap {k8s_pod} -- sh -c \"helm repo remove chartmuseum\""
80 check_output(cmd, shell=True).decode('utf-8')
81 cmd = "kubectl delete namespace test"
82 check_output(cmd, shell=True).decode('utf-8')
83 cmd = "helm repo remove test"
84 check_output(cmd, shell=True).decode('utf-8')
86 logger.info("Test Session cleanup done for %s", request.node.name)
89 def deploy_chartmuseum():
90 """Start chartmuseum pod and populate with the nedded helm chart."""
91 logger.info("Start to deploy chartmuseum")
92 cmd = "helm repo add test https://chartmuseum.github.io/charts"
93 check_output(cmd, shell=True).decode('utf-8')
94 cmd = "kubectl create namespace test"
95 check_output(cmd, shell=True).decode('utf-8')
97 cmd = "helm install test test/chartmuseum --version 3.1.0 --namespace test --set env.open.DISABLE_API=false"
98 check_output(cmd, shell=True).decode('utf-8')
99 wait(lambda: is_chartmuseum_up(), sleep_seconds=10, timeout_seconds=60, waiting_for="chartmuseum to be ready")
102 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"
103 cmd = f"curl -X POST --data-binary @{dname}/resources/cl-test-helm-chart/oru-app-1.0.0.tgz http://{chartmuseum_url}/api/charts"
104 check_output(cmd, shell=True).decode('utf-8')
105 cmd = f"curl -X POST --data-binary @{dname}/resources/cl-test-helm-chart/odu-app-1.0.0.tgz http://{chartmuseum_url}/api/charts"
106 check_output(cmd, shell=True).decode('utf-8')
107 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"
108 check_output(cmd, shell=True).decode('utf-8')
111 def is_chartmuseum_up() -> bool:
112 """Check if the chartmuseum is up."""
113 cmd = "kubectl get pods --field-selector status.phase=Running -n test"
114 result = check_output(cmd, shell=True).decode('utf-8')
115 logger.info("Checking if chartmuseum is UP: %s", result)
117 logger.info("chartmuseum is Down")
119 logger.info("chartmuseum is Up")
123 def is_rapp_up(appname) -> bool:
124 """Check if the rapp is up."""
125 cmd = "kubectl get pods -n nonrtric | grep " + appname + " | wc -l"
126 result = check_output(cmd, shell=True).decode('utf-8')
127 logger.info("Checking if %s is up :%s", appname, result)
129 logger.info("%s is Up", appname.upper())
131 logger.info("%s is Down", appname.upper())
134 def is_rapp_down(appname) -> bool:
135 """Check if the rapp is down."""
136 cmd = "kubectl get pods -n nonrtric | grep " + appname + " | wc -l"
137 result = check_output(cmd, shell=True).decode('utf-8')
138 logger.info("Checking if %s is down :%s", appname, result)
140 logger.info("%s is Down", appname.upper())
142 logger.info("%s is Up", appname.upper())
145 def test_cl_oru_app_deploy():
146 """The Closed Loop O-RU Fronthaul Recovery usecase Apex version."""
147 chart_version = "1.0.0"
148 chart_name = "oru-app"
149 release_name = "nonrtric"
150 pytest.app_name = chart_name
151 logger.info("Upload tosca to commissioning")
152 commissioning_payload = jinja_env().get_template("commission_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartVersion=chart_version, chartName=chart_name, releaseName=release_name)
153 instance_payload = jinja_env().get_template("create_instance_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartVersion=chart_version, chartName=chart_name, releaseName=release_name, instanceName=usecase_name)
154 assert clcommissioning_utils.create_instance(usecase_name, commissioning_payload, instance_payload) is True
156 logger.info("Check if oru-app is up")
157 wait(lambda: is_rapp_up(chart_name), sleep_seconds=5, timeout_seconds=300, waiting_for="Oru app to be up")
159 def test_cl_odu_app_smo_deploy():
160 """The O-DU Slice Assurance SMO Version use case."""
161 chart_version = "1.0.0"
162 chart_name = "odu-app"
163 release_name = "odu-app"
164 pytest.app_name = chart_name
165 logger.info("Upload tosca to commissioning")
166 commissioning_payload = jinja_env().get_template("commission_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartVersion=chart_version, chartName=chart_name, releaseName=release_name)
167 instance_payload = jinja_env().get_template("create_instance_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartVersion=chart_version, chartName=chart_name, releaseName=release_name, instanceName=usecase_name)
168 assert clcommissioning_utils.create_instance(usecase_name, commissioning_payload, instance_payload) is True
170 logger.info("Check if odu-app smo version is up")
171 wait(lambda: is_rapp_up(chart_name), sleep_seconds=5, timeout_seconds=300, waiting_for="Odu app smo version to be up")
173 def test_cl_odu_app_ics_deploy():
174 """The O-DU Slice Assurance ICS Version use case."""
175 chart_version = "1.0.0"
176 chart_name = "odu-app-ics-version"
177 release_name = "odu-app-ics-version"
178 pytest.app_name = chart_name
179 logger.info("Upload tosca to commissioning")
180 commissioning_payload = jinja_env().get_template("commission_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartVersion=chart_version, chartName=chart_name, releaseName=release_name)
181 instance_payload = jinja_env().get_template("create_instance_k8s.json.j2").render(chartmuseumIp=chartmuseum_ip, chartVersion=chart_version, chartName=chart_name, releaseName=release_name, instanceName=usecase_name)
182 assert clcommissioning_utils.create_instance(usecase_name, commissioning_payload, instance_payload) is True
184 logger.info("Check if odu-app ics version is up")
185 wait(lambda: is_rapp_up(chart_name), sleep_seconds=5, timeout_seconds=300, waiting_for="Odu app ics version to be up")