lcm/pub/tests/test_scaleaspect.py

   1 # Copyright 2016-2018 ZTE Corporation.
   2 #
   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
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  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
  27 import os
  28 import mock
  29
  30
  31 class TestScaleAspect(TestCase):
  32
  33     def setUp(self):
  34         self.init_scaling_map_json()
  35         self.initInstModel()
  36
  37         self.init_scale_ns_data()
  38
  39         self.vnf_scale_info_list = [
  40             {
  41                 "vnfd_id": "nf_zte_cscf",
  42                 "vnf_scaleAspectId": "mpu",
  43                 "numberOfSteps": "1"
  44             },
  45             {
  46                 "vnfd_id": "nf_zte_hss",
  47                 "vnf_scaleAspectId": "gsu",
  48                 "numberOfSteps": "2"
  49             }
  50         ]
  51
  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"
  56         self.scaleNsData = {
  57             "aspectId": self.ns_scale_aspect,
  58             "numberOfSteps": self.ns_scale_steps,
  59             "scalingDirection": self.ns_scale_direction
  60         }
  61
  62         self.ns_scale_aspect2 = "TIC_EDGE_HW"
  63         self.ns_scale_steps2 = "4"
  64         self.scaleNsData2 = {
  65             "aspectId": self.ns_scale_aspect2,
  66             "numberOfSteps": self.ns_scale_steps2,
  67             "scalingDirection": self.ns_scale_direction
  68         }
  69
  70     def init_scaling_map_json(self):
  71         curdir_path = os.path.dirname(
  72             os.path.dirname(
  73                 os.path.dirname(
  74                     os.path.abspath(__file__))))
  75         filename = curdir_path + "/ns/data/scalemapping.json"
  76         self.scaling_map_json = get_json_data(filename)
  77
  78     def initInstModel(self):
  79         self.nsd_id = "23"
  80         self.ns_inst_id = "1"
  81         self.ns_name = "ns_1"
  82         self.ns_package_id = "ns_zte"
  83         self.description = "ns_zte"
  84         self.global_customer_id = "global_customer_id"
  85         self.service_type = "service_role"
  86
  87         NSInstModel(id=self.ns_inst_id,
  88                     name=self.ns_name,
  89                     nspackage_id=self.ns_package_id,
  90                     nsd_id=self.nsd_id,
  91                     description=self.description,
  92                     status='empty',
  93                     lastuptime=now_time(),
  94                     global_customer_id=self.global_customer_id,
  95                     service_type=self.service_type).save()
  96
  97         self.nf_inst_id = "231"
  98         self.ns_inst_id = "1"
  99         self.nf_name = "name_1"
 100         self.vnf_id = "1"
 101         self.vnfm_inst_id = "1"
 102         self.package_id = "nf_zte_cscf"
 103         self.nf_uuid = "abc34-345a-de13-ab85-ijs9"
 104
 105         NfInstModel.objects.create(
 106             nfinstid=self.nf_inst_id,
 107             nf_name=self.nf_name,
 108             vnf_id=self.vnf_id,
 109             vnfm_inst_id=self.vnfm_inst_id,
 110             ns_inst_id=self.ns_inst_id,
 111             max_cpu='14',
 112             max_ram='12296',
 113             max_hd='101',
 114             max_shd="20",
 115             max_net=10,
 116             status='active',
 117             mnfinstid=self.nf_uuid,
 118             package_id=self.package_id,
 119             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 120             '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 121             '"productType": "CN","vnfType": "PGW",'
 122             '"description": "PGW VNFD description",'
 123             '"isShared":true,"vnfExtendType":"driver"}}')
 124
 125         # Create a second vnf instance
 126         self.nf_inst_id = "232"
 127         self.package_id = "nf_zte_hss"
 128         self.nf_uuid = "abc34-3g5a-de13-ab85-ijs3"
 129
 130         NfInstModel.objects.create(
 131             nfinstid=self.nf_inst_id,
 132             nf_name=self.nf_name,
 133             vnf_id=self.vnf_id,
 134             vnfm_inst_id=self.vnfm_inst_id,
 135             ns_inst_id=self.ns_inst_id,
 136             max_cpu='14',
 137             max_ram='12296',
 138             max_hd='101',
 139             max_shd="20",
 140             max_net=10,
 141             status='active',
 142             mnfinstid=self.nf_uuid,
 143             package_id=self.package_id,
 144             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 145                        '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 146                        '"productType": "CN","vnfType": "PGW",'
 147                        '"description": "PGW VNFD description",'
 148                        '"isShared":true,"vnfExtendType":"driver"}}')
 149
 150     def add_another_nf_instance(self):
 151         # Create a third vnf instance
 152         nf_inst_id = "233"
 153         package_id = "nf_zte_hss"
 154         nf_uuid = "ab34-3g5j-de13-ab85-ij93"
 155
 156         NfInstModel.objects.create(
 157             nfinstid=nf_inst_id,
 158             nf_name=self.nf_name,
 159             vnf_id=self.vnf_id,
 160             vnfm_inst_id=self.vnfm_inst_id,
 161             ns_inst_id=self.ns_inst_id,
 162             max_cpu='14',
 163             max_ram='12296',
 164             max_hd='101',
 165             max_shd="20",
 166             max_net=10,
 167             status='active',
 168             mnfinstid=nf_uuid,
 169             package_id=package_id,
 170             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 171                        '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 172                        '"productType": "CN","vnfType": "PGW",'
 173                        '"description": "PGW VNFD description",'
 174                        '"isShared":true,"vnfExtendType":"driver"}}')
 175
 176     def add_new_vnf_instance(self):
 177         # Create a third vnf instance
 178         nf_inst_id = "241"
 179         package_id = "nf_hw_cscf"
 180         nf_uuid = "ab34-3g5j-de13-aa85-ij93"
 181
 182         NfInstModel.objects.create(
 183             nfinstid=nf_inst_id,
 184             nf_name=self.nf_name,
 185             vnf_id=self.vnf_id,
 186             vnfm_inst_id=self.vnfm_inst_id,
 187             ns_inst_id=self.ns_inst_id,
 188             max_cpu='14',
 189             max_ram='12296',
 190             max_hd='101',
 191             max_shd="20",
 192             max_net=10,
 193             status='active',
 194             mnfinstid=nf_uuid,
 195             package_id=package_id,
 196             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 197                        '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 198                        '"productType": "CN","vnfType": "PGW",'
 199                        '"description": "PGW VNFD description",'
 200                        '"isShared":true,"vnfExtendType":"driver"}}')
 201
 202         # Create a third vnf instance
 203         nf_inst_id = "242"
 204         package_id = "nf_hw_hss"
 205         nf_uuid = "ab34-3g5j-de13-aa85-id93"
 206
 207         NfInstModel.objects.create(
 208             nfinstid=nf_inst_id,
 209             nf_name=self.nf_name,
 210             vnf_id=self.vnf_id,
 211             vnfm_inst_id=self.vnfm_inst_id,
 212             ns_inst_id=self.ns_inst_id,
 213             max_cpu='14',
 214             max_ram='12296',
 215             max_hd='101',
 216             max_shd="20",
 217             max_net=10,
 218             status='active',
 219             mnfinstid=nf_uuid,
 220             package_id=package_id,
 221             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 222                        '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 223                        '"productType": "CN","vnfType": "PGW",'
 224                        '"description": "PGW VNFD description",'
 225                        '"isShared":true,"vnfExtendType":"driver"}}')
 226
 227     def tearDown(self):
 228         NSInstModel().clean()
 229         NfInstModel().clean()
 230
 231     def test_get_and_check_params(self):
 232         aspect, numberOfSteps, scale_type = get_and_check_params(
 233             self.scaleNsData, "1")
 234         self.assertEqual(aspect, self.ns_scale_aspect)
 235         self.assertEqual(numberOfSteps, self.ns_scale_steps)
 236         self.assertEqual(scale_type, self.ns_scale_direction)
 237
 238     def test_get_scale_vnf_data_from_json(self):
 239         vnf_data_package = get_scale_vnf_data_from_json(
 240             self.scaling_map_json, "23", "TIC_EDGE_IMS", "1")
 241         self.assertIsNotNone(vnf_data_package)
 242         self.assertEqual(2, vnf_data_package.__len__())
 243         self.assertIsNotNone(vnf_data_package)
 244         self.assertEqual(2, vnf_data_package.__len__())
 245         self.assertEqual("nf_zte_cscf", vnf_data_package[0]["vnfd_id"])
 246         self.assertEqual("1", vnf_data_package[0]["numberOfSteps"])
 247         self.assertEqual("gsu", vnf_data_package[0]["vnf_scaleAspectId"])
 248         self.assertEqual("nf_zte_hss", vnf_data_package[1]["vnfd_id"])
 249         self.assertEqual("3", vnf_data_package[1]["numberOfSteps"])
 250         self.assertEqual("gpu", vnf_data_package[1]["vnf_scaleAspectId"])
 251
 252     def test_get_scale_vnf_data_from_json_2(self):
 253         vnf_data_package = get_scale_vnf_data_from_json(
 254             self.scaling_map_json, "23", "TIC_EDGE_IMS", "2")
 255         self.assertIsNotNone(vnf_data_package)
 256         self.assertEqual(2, vnf_data_package.__len__())
 257         self.assertEqual("nf_zte_cscf", vnf_data_package[0]["vnfd_id"])
 258         self.assertEqual("2", vnf_data_package[0]["numberOfSteps"])
 259         self.assertEqual("mpu", vnf_data_package[0]["vnf_scaleAspectId"])
 260         self.assertEqual("nf_zte_hss", vnf_data_package[1]["vnfd_id"])
 261         self.assertEqual("4", vnf_data_package[1]["numberOfSteps"])
 262         self.assertEqual("mpu", vnf_data_package[1]["vnf_scaleAspectId"])
 263
 264     def test_set_scacle_vnf_instance_id(self):
 265         result = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
 266         self.assertEqual(2, result.__len__())
 267         self.assertEqual(result[0]["numberOfSteps"],
 268                          self.vnf_scale_info_list[0]["numberOfSteps"])
 269         self.assertEqual(
 270             result[0]["vnf_scaleAspectId"],
 271             self.vnf_scale_info_list[0]["vnf_scaleAspectId"])
 272         self.assertEqual(result[1]["numberOfSteps"],
 273                          self.vnf_scale_info_list[1]["numberOfSteps"])
 274         self.assertEqual(
 275             result[1]["vnf_scaleAspectId"],
 276             self.vnf_scale_info_list[1]["vnf_scaleAspectId"])
 277         self.assertEqual("231", result[0]["vnfInstanceId"])
 278         self.assertEqual("232", result[1]["vnfInstanceId"])
 279         self.assertNotIn("vnfd_id", result[0])
 280         self.assertNotIn("vnfd_id", result[1])
 281
 282     def test_set_scacle_vnf_instance_id_2(self):
 283         vnf_scale_info_list = [
 284             {
 285                 "vnfd_id": "error1",
 286                 "vnf_scaleAspectId": "mpu",
 287                 "numberOfSteps": "1"
 288             },
 289             {
 290                 "vnfd_id": "nf_zte_hss",
 291                 "vnf_scaleAspectId": "mpu",
 292                 "numberOfSteps": "1"
 293             }
 294         ]
 295         result = set_scacle_vnf_instance_id(vnf_scale_info_list)
 296         self.assertEqual(1, result.__len__())
 297         self.assertEqual(
 298             result[0]["numberOfSteps"],
 299             vnf_scale_info_list[0]["numberOfSteps"])
 300         self.assertEqual(
 301             result[0]["vnf_scaleAspectId"],
 302             vnf_scale_info_list[0]["vnf_scaleAspectId"])
 303         self.assertEqual("232", result[0]["vnfInstanceId"])
 304         self.assertNotIn("vnfd_id", result[0])
 305
 306     def test_set_scacle_vnf_instance_id_3(self):
 307         vnf_scale_info_list = [
 308             {
 309                 "vnfd_id": "error1",
 310                 "vnf_scaleAspectId": "mpu",
 311                 "numberOfSteps": "1"
 312             },
 313             {
 314                 "vnfd_id": "error2",
 315                 "vnf_scaleAspectId": "gsu",
 316                 "numberOfSteps": "1"
 317             }
 318         ]
 319         result = set_scacle_vnf_instance_id(vnf_scale_info_list)
 320         self.assertEqual(0, result.__len__())
 321
 322     def test_set_scacle_vnf_instance_id_4(self):
 323         self.add_another_nf_instance()
 324         result = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
 325         self.assertEqual(3, result.__len__())
 326         self.assertEqual("231", result[0]["vnfInstanceId"])
 327         self.assertEqual("232", result[1]["vnfInstanceId"])
 328         self.assertEqual("233", result[2]["vnfInstanceId"])
 329
 330     def test_set_scaleVnfData_type(self):
 331         vnf_scale_list = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
 332         result = set_scaleVnfData_type(vnf_scale_list, self.ns_scale_direction)
 333         self.assertEqual(2, result.__len__())
 334         self.assertNotIn("scaleByStepData", result)
 335         self.assertEqual(
 336             self.ns_scale_direction,
 337             result[0]["scaleByStepData"]["type"])
 338         self.assertEqual("mpu", result[0]["scaleByStepData"]["aspectId"])
 339         self.assertNotIn("vnf_scaleAspectId", result[0]["scaleByStepData"])
 340         self.assertEqual("1", result[0]["scaleByStepData"]["numberOfSteps"])
 341         self.assertEqual(
 342             self.ns_scale_direction,
 343             result[1]["scaleByStepData"]["type"])
 344         self.assertEqual("gsu", result[1]["scaleByStepData"]["aspectId"])
 345         self.assertNotIn("vnf_scaleAspectId", result[1]["scaleByStepData"])
 346         self.assertEqual("2", result[1]["scaleByStepData"]["numberOfSteps"])
 347
 348     def test_get_nsdId(self):
 349         nsd_id = get_nsdId("1")
 350         self.assertEqual("23", nsd_id)
 351
 352     @mock.patch.object(catalog, 'get_scalingmap_json_package')
 353     def test_get_scale_vnf_data_info_list(
 354             self, mock_get_scalingmap_json_package):
 355         mock_get_scalingmap_json_package.return_value = self.scaling_map_json
 356
 357         scale_vnf_data = get_scale_vnf_data_info_list(self.scaleNsData, "1")
 358         self.assertIsNotNone(scale_vnf_data)
 359         self.assertEqual(2, scale_vnf_data.__len__())
 360
 361     @mock.patch.object(catalog, 'get_scalingmap_json_package')
 362     def test_get_scale_vnf_data_info_list_2(
 363             self, mock_get_scalingmap_json_package):
 364         mock_get_scalingmap_json_package.return_value = self.scaling_map_json
 365
 366         scale_vnf_data = None
 367         is_exception_caught = False
 368         try:
 369             scale_vnf_data = get_scale_vnf_data_info_list(
 370                 self.scaleNsData2, "1")
 371         except Exception:
 372             is_exception_caught = True
 373         self.assertTrue(is_exception_caught)
 374         self.assertIsNone(scale_vnf_data)
 375
 376     @mock.patch.object(catalog, 'get_scalingmap_json_package')
 377     def test_get_scale_vnf_data_info_list_3(
 378             self, mock_get_scalingmap_json_package):
 379         mock_get_scalingmap_json_package.return_value = self.scaling_map_json
 380         self.add_new_vnf_instance()
 381
 382         scale_vnf_data = None
 383         is_exception_caught = False
 384         try:
 385             scale_vnf_data = get_scale_vnf_data_info_list(
 386                 self.scaleNsData2, "1")
 387         except Exception:
 388             is_exception_caught = True
 389         self.assertFalse(is_exception_caught)
 390         self.assertEqual(2, scale_vnf_data.__len__())