1 # Copyright 2016-2018 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.
15 from django.test import TestCase
16 from lcm.pub.utils.scaleaspect import get_json_data
17 from lcm.pub.utils.scaleaspect import get_nsdId
18 from lcm.pub.utils.scaleaspect import get_scale_vnf_data_from_json
19 from lcm.pub.utils.scaleaspect import get_scale_vnf_data_info_list
20 from lcm.pub.utils.scaleaspect import set_scacle_vnf_instance_id
21 from lcm.pub.utils.scaleaspect import get_and_check_params
22 from lcm.pub.utils.scaleaspect import set_scaleVnfData_type
23 from lcm.pub.database.models import NfInstModel
24 from lcm.pub.database.models import NSInstModel
25 from lcm.pub.msapi import catalog
26 from lcm.pub.utils.timeutil import now_time
31 class TestScaleAspect(TestCase):
34 self.init_scaling_map_json()
37 self.init_scale_ns_data()
39 self.vnf_scale_info_list = [
41 "vnfd_id": "nf_zte_cscf",
42 "vnf_scaleAspectId": "mpu",
46 "vnfd_id": "nf_zte_hss",
47 "vnf_scaleAspectId": "gsu",
52 def init_scale_ns_data(self):
53 self.ns_scale_aspect = "TIC_EDGE_IMS"
54 self.ns_scale_steps = "1"
55 self.ns_scale_direction = "SCALE_IN"
57 "scaleNsByStepsData": [
59 "aspectId": self.ns_scale_aspect,
60 "numberOfSteps": self.ns_scale_steps,
61 "scalingDirection": self.ns_scale_direction
66 self.ns_scale_aspect2 = "TIC_EDGE_HW"
67 self.ns_scale_steps2 = "4"
69 "scaleNsByStepsData": [
71 "aspectId": self.ns_scale_aspect2,
72 "numberOfSteps": self.ns_scale_steps2,
73 "scalingDirection": self.ns_scale_direction
78 def init_scaling_map_json(self):
79 curdir_path = os.path.dirname(
82 os.path.abspath(__file__))))
83 filename = curdir_path + "/ns/data/scalemapping.json"
84 self.scaling_map_json = get_json_data(filename)
86 def initInstModel(self):
90 self.ns_package_id = "ns_zte"
91 self.description = "ns_zte"
92 self.global_customer_id = "global_customer_id"
93 self.service_type = "service_role"
95 NSInstModel(id=self.ns_inst_id,
97 nspackage_id=self.ns_package_id,
99 description=self.description,
101 lastuptime=now_time(),
102 global_customer_id=self.global_customer_id,
103 service_type=self.service_type).save()
105 self.nf_inst_id = "231"
106 self.ns_inst_id = "1"
107 self.nf_name = "name_1"
109 self.vnfm_inst_id = "1"
110 self.package_id = "nf_zte_cscf"
111 self.nf_uuid = "abc34-345a-de13-ab85-ijs9"
113 NfInstModel.objects.create(
114 nfinstid=self.nf_inst_id,
115 nf_name=self.nf_name,
117 vnfm_inst_id=self.vnfm_inst_id,
118 ns_inst_id=self.ns_inst_id,
125 mnfinstid=self.nf_uuid,
126 package_id=self.package_id,
127 vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
128 '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
129 '"productType": "CN","vnfType": "PGW",'
130 '"description": "PGW VNFD description",'
131 '"isShared":true,"vnfExtendType":"driver"}}')
133 # Create a second vnf instance
134 self.nf_inst_id = "232"
135 self.package_id = "nf_zte_hss"
136 self.nf_uuid = "abc34-3g5a-de13-ab85-ijs3"
138 NfInstModel.objects.create(
139 nfinstid=self.nf_inst_id,
140 nf_name=self.nf_name,
142 vnfm_inst_id=self.vnfm_inst_id,
143 ns_inst_id=self.ns_inst_id,
150 mnfinstid=self.nf_uuid,
151 package_id=self.package_id,
152 vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
153 '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
154 '"productType": "CN","vnfType": "PGW",'
155 '"description": "PGW VNFD description",'
156 '"isShared":true,"vnfExtendType":"driver"}}')
158 def add_another_nf_instance(self):
159 # Create a third vnf instance
161 package_id = "nf_zte_hss"
162 nf_uuid = "ab34-3g5j-de13-ab85-ij93"
164 NfInstModel.objects.create(
166 nf_name=self.nf_name,
168 vnfm_inst_id=self.vnfm_inst_id,
169 ns_inst_id=self.ns_inst_id,
177 package_id=package_id,
178 vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
179 '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
180 '"productType": "CN","vnfType": "PGW",'
181 '"description": "PGW VNFD description",'
182 '"isShared":true,"vnfExtendType":"driver"}}')
184 def add_new_vnf_instance(self):
185 # Create a third vnf instance
187 package_id = "nf_hw_cscf"
188 nf_uuid = "ab34-3g5j-de13-aa85-ij93"
190 NfInstModel.objects.create(
192 nf_name=self.nf_name,
194 vnfm_inst_id=self.vnfm_inst_id,
195 ns_inst_id=self.ns_inst_id,
203 package_id=package_id,
204 vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
205 '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
206 '"productType": "CN","vnfType": "PGW",'
207 '"description": "PGW VNFD description",'
208 '"isShared":true,"vnfExtendType":"driver"}}')
210 # Create a third vnf instance
212 package_id = "nf_hw_hss"
213 nf_uuid = "ab34-3g5j-de13-aa85-id93"
215 NfInstModel.objects.create(
217 nf_name=self.nf_name,
219 vnfm_inst_id=self.vnfm_inst_id,
220 ns_inst_id=self.ns_inst_id,
228 package_id=package_id,
229 vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
230 '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
231 '"productType": "CN","vnfType": "PGW",'
232 '"description": "PGW VNFD description",'
233 '"isShared":true,"vnfExtendType":"driver"}}')
236 NSInstModel().clean()
237 NfInstModel().clean()
239 def test_get_and_check_params(self):
240 aspect, numberOfSteps, scale_type = get_and_check_params(
241 self.scaleNsData, "1")
242 self.assertEqual(aspect, self.ns_scale_aspect)
243 self.assertEqual(numberOfSteps, self.ns_scale_steps)
244 self.assertEqual(scale_type, self.ns_scale_direction)
246 def test_get_scale_vnf_data_from_json(self):
247 vnf_data_package = get_scale_vnf_data_from_json(
248 self.scaling_map_json, "23", "TIC_EDGE_IMS", "1")
249 self.assertIsNotNone(vnf_data_package)
250 self.assertEqual(2, vnf_data_package.__len__())
251 self.assertIsNotNone(vnf_data_package)
252 self.assertEqual(2, vnf_data_package.__len__())
253 self.assertEqual("nf_zte_cscf", vnf_data_package[0]["vnfd_id"])
254 self.assertEqual("1", vnf_data_package[0]["numberOfSteps"])
255 self.assertEqual("gsu", vnf_data_package[0]["vnf_scaleAspectId"])
256 self.assertEqual("nf_zte_hss", vnf_data_package[1]["vnfd_id"])
257 self.assertEqual("3", vnf_data_package[1]["numberOfSteps"])
258 self.assertEqual("gpu", vnf_data_package[1]["vnf_scaleAspectId"])
260 def test_get_scale_vnf_data_from_json_2(self):
261 vnf_data_package = get_scale_vnf_data_from_json(
262 self.scaling_map_json, "23", "TIC_EDGE_IMS", "2")
263 self.assertIsNotNone(vnf_data_package)
264 self.assertEqual(2, vnf_data_package.__len__())
265 self.assertEqual("nf_zte_cscf", vnf_data_package[0]["vnfd_id"])
266 self.assertEqual("2", vnf_data_package[0]["numberOfSteps"])
267 self.assertEqual("mpu", vnf_data_package[0]["vnf_scaleAspectId"])
268 self.assertEqual("nf_zte_hss", vnf_data_package[1]["vnfd_id"])
269 self.assertEqual("4", vnf_data_package[1]["numberOfSteps"])
270 self.assertEqual("mpu", vnf_data_package[1]["vnf_scaleAspectId"])
272 def test_set_scacle_vnf_instance_id(self):
273 result = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
274 self.assertEqual(2, result.__len__())
275 self.assertEqual(result[0]["numberOfSteps"],
276 self.vnf_scale_info_list[0]["numberOfSteps"])
278 result[0]["vnf_scaleAspectId"],
279 self.vnf_scale_info_list[0]["vnf_scaleAspectId"])
280 self.assertEqual(result[1]["numberOfSteps"],
281 self.vnf_scale_info_list[1]["numberOfSteps"])
283 result[1]["vnf_scaleAspectId"],
284 self.vnf_scale_info_list[1]["vnf_scaleAspectId"])
285 self.assertEqual("231", result[0]["vnfInstanceId"])
286 self.assertEqual("232", result[1]["vnfInstanceId"])
287 self.assertNotIn("vnfd_id", result[0])
288 self.assertNotIn("vnfd_id", result[1])
290 def test_set_scacle_vnf_instance_id_2(self):
291 vnf_scale_info_list = [
294 "vnf_scaleAspectId": "mpu",
298 "vnfd_id": "nf_zte_hss",
299 "vnf_scaleAspectId": "mpu",
303 result = set_scacle_vnf_instance_id(vnf_scale_info_list)
304 self.assertEqual(1, result.__len__())
306 result[0]["numberOfSteps"],
307 vnf_scale_info_list[0]["numberOfSteps"])
309 result[0]["vnf_scaleAspectId"],
310 vnf_scale_info_list[0]["vnf_scaleAspectId"])
311 self.assertEqual("232", result[0]["vnfInstanceId"])
312 self.assertNotIn("vnfd_id", result[0])
314 def test_set_scacle_vnf_instance_id_3(self):
315 vnf_scale_info_list = [
318 "vnf_scaleAspectId": "mpu",
323 "vnf_scaleAspectId": "gsu",
327 result = set_scacle_vnf_instance_id(vnf_scale_info_list)
328 self.assertEqual(0, result.__len__())
330 def test_set_scacle_vnf_instance_id_4(self):
331 self.add_another_nf_instance()
332 result = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
333 self.assertEqual(3, result.__len__())
334 self.assertEqual("231", result[0]["vnfInstanceId"])
335 self.assertEqual("232", result[1]["vnfInstanceId"])
336 self.assertEqual("233", result[2]["vnfInstanceId"])
338 def test_set_scaleVnfData_type(self):
339 vnf_scale_list = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
340 result = set_scaleVnfData_type(vnf_scale_list, self.ns_scale_direction)
341 self.assertEqual(2, result.__len__())
342 self.assertNotIn("scaleByStepData", result)
344 self.ns_scale_direction,
345 result[0]["scaleByStepData"]["type"])
346 self.assertEqual("mpu", result[0]["scaleByStepData"]["aspectId"])
347 self.assertNotIn("vnf_scaleAspectId", result[0]["scaleByStepData"])
348 self.assertEqual("1", result[0]["scaleByStepData"]["numberOfSteps"])
350 self.ns_scale_direction,
351 result[1]["scaleByStepData"]["type"])
352 self.assertEqual("gsu", result[1]["scaleByStepData"]["aspectId"])
353 self.assertNotIn("vnf_scaleAspectId", result[1]["scaleByStepData"])
354 self.assertEqual("2", result[1]["scaleByStepData"]["numberOfSteps"])
356 def test_get_nsdId(self):
357 nsd_id = get_nsdId("1")
358 self.assertEqual("23", nsd_id)
360 @mock.patch.object(catalog, 'get_scalingmap_json_package')
361 def test_get_scale_vnf_data_info_list(
362 self, mock_get_scalingmap_json_package):
363 mock_get_scalingmap_json_package.return_value = self.scaling_map_json
365 scale_vnf_data = get_scale_vnf_data_info_list(self.scaleNsData, "1")
366 self.assertIsNotNone(scale_vnf_data)
367 self.assertEqual(2, scale_vnf_data.__len__())
369 @mock.patch.object(catalog, 'get_scalingmap_json_package')
370 def test_get_scale_vnf_data_info_list_2(
371 self, mock_get_scalingmap_json_package):
372 mock_get_scalingmap_json_package.return_value = self.scaling_map_json
374 scale_vnf_data = None
375 is_exception_caught = False
377 scale_vnf_data = get_scale_vnf_data_info_list(
378 self.scaleNsData2, "1")
380 is_exception_caught = True
381 self.assertTrue(is_exception_caught)
382 self.assertIsNone(scale_vnf_data)
384 @mock.patch.object(catalog, 'get_scalingmap_json_package')
385 def test_get_scale_vnf_data_info_list_3(
386 self, mock_get_scalingmap_json_package):
387 mock_get_scalingmap_json_package.return_value = self.scaling_map_json
388 self.add_new_vnf_instance()
390 scale_vnf_data = None
391 is_exception_caught = False
393 scale_vnf_data = get_scale_vnf_data_info_list(
394 self.scaleNsData2, "1")
396 is_exception_caught = True
397 self.assertFalse(is_exception_caught)
398 self.assertEqual(2, scale_vnf_data.__len__())