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.
19 from rest_framework.test import APIClient
20 from django.test import TestCase
21 from rest_framework import status
23 from lcm.ns.biz.ns_manual_scale import NSManualScaleService
24 from lcm.ns.enum import NS_INST_STATUS
25 from lcm.pub.database.models import NSInstModel, JobModel, NfInstModel
26 from lcm.pub.exceptions import NSLCMException
27 from lcm.pub.msapi import catalog
28 from lcm.pub.utils import restcall
29 from lcm.pub.utils.jobutil import JobUtil
30 from lcm.jobs.enum import JOB_MODEL_STATUS, JOB_TYPE, JOB_ACTION
31 from lcm.ns.tests import SCALING_MAP_DICT, VNFD_MODEL_DICT, SCALE_NS_DICT
34 class TestNsManualScale(TestCase):
37 self.scaling_map_json = SCALING_MAP_DICT
38 self.ns_inst_id = str(uuid.uuid4())
39 self.job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, self.ns_inst_id)
40 self.client = APIClient()
45 nspackage_id=self.package_id,
49 NSInstModel.objects.filter().delete()
50 JobModel.objects.filter().delete()
52 def insert_new_ns(self):
53 ns_inst_id = str(uuid.uuid4())
54 job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, self.ns_inst_id)
59 nspackage_id=package_id,
60 nsd_id=package_id).save()
61 return ns_inst_id, job_id
63 def insert_new_nf(self):
64 self.vnfm_inst_id = "1"
65 NfInstModel.objects.create(
70 ns_inst_id=self.ns_inst_id,
77 mnfinstid="ab34-3g5j-de13-ab85-ij93",
78 package_id="nf_zte_hss",
79 vnfd_model=json.dumps(VNFD_MODEL_DICT))
81 @mock.patch.object(NSManualScaleService, 'run')
82 def test_ns_manual_scale(self, mock_run):
83 response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data=SCALE_NS_DICT, format='json')
84 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
86 def test_ns_manual_scale_error_scaletype(self):
87 scale_ns_json = SCALE_NS_DICT.copy()
88 scale_ns_json["scaleType"] = "SCALE_ERR"
89 NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
90 jobs = JobModel.objects.filter(jobid=self.job_id)
91 self.assertEqual(255, jobs[0].progress)
93 def test_ns_manual_scale_error_nsd_id(self):
94 scale_ns_json = SCALE_NS_DICT.copy()
95 scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
96 NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
97 jobs = JobModel.objects.filter(jobid=self.job_id)
98 self.assertEqual(255, jobs[0].progress)
100 def test_ns_manual_scale_error_aspect(self):
101 scale_ns_json = SCALE_NS_DICT.copy()
102 scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
103 ns_inst_id, job_id = self.insert_new_ns()
104 job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, ns_inst_id)
105 NSManualScaleService(ns_inst_id, scale_ns_json, job_id).run()
106 jobs = JobModel.objects.filter(jobid=job_id)
107 self.assertEqual(255, jobs[0].progress)
109 @mock.patch.object(catalog, 'get_scalingmap_json_package')
110 @mock.patch.object(NSManualScaleService, 'do_vnfs_scale')
111 def test_ns_manual_scale_success(self, mock_do_vnfs_scale, mock_get_scalingmap_json_package):
112 scale_ns_json = SCALE_NS_DICT.copy()
113 scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "TIC_EDGE_IMS"
114 mock_get_scalingmap_json_package.return_value = self.scaling_map_json
115 mock_do_vnfs_scale.return_value = JOB_MODEL_STATUS.FINISHED
116 ns_inst_id, job_id = self.insert_new_ns()
117 job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, ns_inst_id)
119 NSManualScaleService(ns_inst_id, scale_ns_json, job_id).run()
120 jobs = JobModel.objects.filter(jobid=job_id)
121 self.assertEqual(255, jobs[0].progress)
123 @mock.patch.object(restcall, 'call_req')
124 def test_ns_manual_scale_thread(self, mock_call):
125 scale_ns_json = SCALE_NS_DICT.copy()
126 NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
127 self.assertTrue(NSInstModel.objects.get(id=self.ns_inst_id).status, NS_INST_STATUS.ACTIVE)
129 @mock.patch.object(NSManualScaleService, 'start')
130 def test_ns_manual_scale_empty_data(self, mock_start):
131 mock_start.side_effect = NSLCMException("NS scale failed.")
132 response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data={}, format='json')
133 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
134 self.assertIn("error", response.data)
136 @mock.patch.object(NSManualScaleService, 'start')
137 def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
138 mock_start.side_effect = NSLCMException("NS scale failed.")
139 scale_ns_json = SCALE_NS_DICT.copy()
140 response = self.client.post("/api/nslcm/v1/ns/11/scale", data=scale_ns_json, format='json')
141 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
142 self.assertIn("error", response.data)