1 # Copyright 2017 ZTE Corporation.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
18 from django.test import Client
19 from django.test import TestCase
20 from rest_framework import status
21 from lcm.ns.const import NS_INST_STATUS
22 from lcm.ns.ns_manual_scale import NSManualScaleService
23 from lcm.pub.database.models import NSInstModel, JobModel, NfInstModel
24 from lcm.pub.exceptions import NSLCMException
25 from lcm.pub.utils import restcall
26 from lcm.pub.utils.jobutil import JobUtil, JOB_TYPE, JOB_MODEL_STATUS
27 from lcm.pub.msapi import catalog
28 from lcm.pub.utils.scaleaspect import get_json_data
35 "ns_scale_aspect": "TIC_CORE_IMS",
41 "vnfd_id": "zte_ims_cscf",
42 "vnf_scale_aspect": "mpu",
46 "vnfd_id": "zte_ims_hss",
47 "vnf_scale_aspect": "fpu",
56 "vnfd_id": "zte_ims_cscf",
57 "vnf_scale_aspect": "mpu",
61 "vnfd_id": "zte_ims_hss",
62 "vnf_scale_aspect": "fpu",
71 "ns_scale_aspect": "TIC_EDGE_EPC",
77 "vnfd_id": "zte_epc_spgw",
78 "vnf_scale_aspect": "gpu",
82 "vnfd_id": "zte_epc_tas",
83 "vnf_scale_aspect": "fpu",
92 "vnfd_id": "zte_epc_spgw",
93 "vnf_scale_aspect": "mpu",
97 "vnfd_id": "zte_epc_tas",
98 "vnf_scale_aspect": "fpu",
109 class TestNsManualScale(TestCase):
111 self.ns_inst_id = str(uuid.uuid4())
112 self.job_id = JobUtil.create_job(
113 "NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
114 self.package_id = "7"
115 self.client = Client()
119 nspackage_id=self.package_id,
122 self.init_scaling_map_json()
125 NSInstModel.objects.filter().delete()
126 JobModel.objects.filter().delete()
128 def init_scaling_map_json(self):
129 curdir_path = os.path.dirname(
132 os.path.abspath(__file__))))
133 filename = curdir_path + "/ns/data/scalemapping.json"
134 self.scaling_map_json = get_json_data(filename)
136 def insert_new_ns(self):
137 ns_inst_id = str(uuid.uuid4())
138 job_id = JobUtil.create_job(
139 "NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
144 nspackage_id=package_id,
145 nsd_id=package_id).save()
146 return ns_inst_id, job_id
148 def insert_new_nf(self):
149 # Create a third vnf instance
150 self.nf_name = "name_1"
152 self.vnfm_inst_id = "1"
154 package_id = "nf_zte_hss"
155 nf_uuid = "ab34-3g5j-de13-ab85-ij93"
157 NfInstModel.objects.create(
159 nf_name=self.nf_name,
161 vnfm_inst_id=self.vnfm_inst_id,
162 ns_inst_id=self.ns_inst_id,
170 package_id=package_id,
171 vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
172 '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
173 '"productType": "CN","vnfType": "PGW",'
174 '"description": "PGW VNFD description",'
175 '"isShared":true,"vnfExtendType":"driver"}}')
177 @mock.patch.object(NSManualScaleService, 'run')
178 def test_ns_manual_scale(self, mock_run):
180 "scaleType": "SCALE_NS",
182 "scaleNsByStepsData": [{
185 "scalingDirection": "0"
189 response = self.client.post(
190 "/api/nslcm/v1/ns/%s/scale" %
191 self.ns_inst_id, data=data)
192 self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
194 def test_ns_manual_scale_error_scaletype(self):
196 "scaleType": "SCALE_ERR",
198 "scaleNsByStepsData": [{
199 "aspectId": "sss_zte",
201 "scalingDirection": "0"
205 NSManualScaleService(self.ns_inst_id, data, self.job_id).run()
206 jobs = JobModel.objects.filter(jobid=self.job_id)
207 self.assertEqual(255, jobs[0].progress)
209 def test_ns_manual_scale_error_nsd_id(self):
211 "scaleType": "SCALE_NS",
213 "scaleNsByStepsData": [{
214 "aspectId": "sss_zte",
216 "scalingDirection": "0"
220 NSManualScaleService(self.ns_inst_id, data, self.job_id).run()
221 jobs = JobModel.objects.filter(jobid=self.job_id)
222 self.assertEqual(255, jobs[0].progress)
224 def test_ns_manual_scale_error_aspect(self):
226 "scaleType": "SCALE_NS",
228 "scaleNsByStepsData": [{
229 "aspectId": "sss_zte",
231 "scalingDirection": "0"
235 ns_inst_id, job_id = self.insert_new_ns()
236 job_id = JobUtil.create_job(
237 "NS", JOB_TYPE.MANUAL_SCALE_VNF, ns_inst_id)
238 NSManualScaleService(ns_inst_id, data, job_id).run()
239 jobs = JobModel.objects.filter(jobid=job_id)
240 self.assertEqual(255, jobs[0].progress)
242 @mock.patch.object(catalog, 'get_scalingmap_json_package')
243 @mock.patch.object(NSManualScaleService, 'do_vnfs_scale')
244 def test_ns_manual_scale_success(self, mock_do_vnfs_scale, mock_get_scalingmap_json_package):
246 "scaleType": "SCALE_NS",
248 "scaleNsByStepsData": [{
249 "aspectId": "TIC_EDGE_IMS",
250 "numberOfSteps": "1",
251 "scalingDirection": "0"
255 mock_get_scalingmap_json_package.return_value = self.scaling_map_json
256 mock_do_vnfs_scale.return_value = JOB_MODEL_STATUS.FINISHED
257 ns_inst_id, job_id = self.insert_new_ns()
258 job_id = JobUtil.create_job(
259 "NS", JOB_TYPE.MANUAL_SCALE_VNF, ns_inst_id)
261 NSManualScaleService(ns_inst_id, data, job_id).run()
262 jobs = JobModel.objects.filter(jobid=job_id)
263 self.assertEqual(255, jobs[0].progress)
265 @mock.patch.object(restcall, 'call_req')
266 def test_ns_manual_scale_thread(self, mock_call):
268 "scaleType": "SCALE_NS",
270 "scaleNsByStepsData": [{
273 "scalingDirection": "0"
277 NSManualScaleService(self.ns_inst_id, data, self.job_id).run()
279 NSInstModel.objects.get(
280 id=self.ns_inst_id).status,
281 NS_INST_STATUS.ACTIVE)
283 def test_swagger_ok(self):
284 resp = self.client.get("/api/nslcm/v1/swagger.json", format='json')
285 self.assertEqual(resp.status_code, status.HTTP_200_OK)
287 @mock.patch.object(NSManualScaleService, 'start')
288 def test_ns_manual_scale_empty_data(self, mock_start):
289 mock_start.side_effect = NSLCMException("NS scale failed.")
290 response = self.client.post(
291 "/api/nslcm/v1/ns/%s/scale" %
292 self.ns_inst_id, data={})
294 response.status_code,
295 status.HTTP_500_INTERNAL_SERVER_ERROR)
296 self.assertIn("error", response.data)
298 @mock.patch.object(NSManualScaleService, 'start')
299 def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
300 mock_start.side_effect = NSLCMException("NS scale failed.")
302 "scaleType": "SCALE_NS",
304 "scaleNsByStepsData": [{
307 "scalingDirection": "0"
311 response = self.client.post("/api/nslcm/v1/ns/11/scale", data=data)
313 response.status_code,
314 status.HTTP_500_INTERNAL_SERVER_ERROR)
315 self.assertIn("error", response.data)