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 check_and_set_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             "scaleNsByStepsData": [
  58                 {
  59                     "aspectId": self.ns_scale_aspect,
  60                     "numberOfSteps": self.ns_scale_steps,
  61                     "scalingDirection": self.ns_scale_direction
  62                 }
  63             ]
  64         }]
  65
  66         self.ns_scale_aspect2 = "TIC_EDGE_HW"
  67         self.ns_scale_steps2 = "4"
  68         self.scaleNsData2 = [{
  69             "scaleNsByStepsData": [
  70                 {
  71                     "aspectId": self.ns_scale_aspect2,
  72                     "numberOfSteps": self.ns_scale_steps2,
  73                     "scalingDirection": self.ns_scale_direction
  74                 }
  75             ]
  76         }]
  77
  78     def init_scaling_map_json(self):
  79         curdir_path = os.path.dirname(
  80             os.path.dirname(
  81                 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)
  85
  86     def initInstModel(self):
  87         self.nsd_id = "23"
  88         self.ns_inst_id = "1"
  89         self.ns_name = "ns_1"
  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"
  94
  95         NSInstModel(id=self.ns_inst_id,
  96                     name=self.ns_name,
  97                     nspackage_id=self.ns_package_id,
  98                     nsd_id=self.nsd_id,
  99                     description=self.description,
 100                     status='empty',
 101                     lastuptime=now_time(),
 102                     global_customer_id=self.global_customer_id,
 103                     service_type=self.service_type).save()
 104
 105         self.nf_inst_id = "231"
 106         self.ns_inst_id = "1"
 107         self.nf_name = "name_1"
 108         self.vnf_id = "1"
 109         self.vnfm_inst_id = "1"
 110         self.package_id = "nf_zte_cscf"
 111         self.nf_uuid = "abc34-345a-de13-ab85-ijs9"
 112
 113         NfInstModel.objects.create(
 114             nfinstid=self.nf_inst_id,
 115             nf_name=self.nf_name,
 116             vnf_id=self.vnf_id,
 117             vnfm_inst_id=self.vnfm_inst_id,
 118             ns_inst_id=self.ns_inst_id,
 119             max_cpu='14',
 120             max_ram='12296',
 121             max_hd='101',
 122             max_shd="20",
 123             max_net=10,
 124             status='active',
 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"}}')
 132
 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"
 137
 138         NfInstModel.objects.create(
 139             nfinstid=self.nf_inst_id,
 140             nf_name=self.nf_name,
 141             vnf_id=self.vnf_id,
 142             vnfm_inst_id=self.vnfm_inst_id,
 143             ns_inst_id=self.ns_inst_id,
 144             max_cpu='14',
 145             max_ram='12296',
 146             max_hd='101',
 147             max_shd="20",
 148             max_net=10,
 149             status='active',
 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"}}')
 157
 158     def add_another_nf_instance(self):
 159         # Create a third vnf instance
 160         nf_inst_id = "233"
 161         package_id = "nf_zte_hss"
 162         nf_uuid = "ab34-3g5j-de13-ab85-ij93"
 163
 164         NfInstModel.objects.create(
 165             nfinstid=nf_inst_id,
 166             nf_name=self.nf_name,
 167             vnf_id=self.vnf_id,
 168             vnfm_inst_id=self.vnfm_inst_id,
 169             ns_inst_id=self.ns_inst_id,
 170             max_cpu='14',
 171             max_ram='12296',
 172             max_hd='101',
 173             max_shd="20",
 174             max_net=10,
 175             status='active',
 176             mnfinstid=nf_uuid,
 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"}}')
 183
 184     def add_new_vnf_instance(self):
 185         # Create a third vnf instance
 186         nf_inst_id = "241"
 187         package_id = "nf_hw_cscf"
 188         nf_uuid = "ab34-3g5j-de13-aa85-ij93"
 189
 190         NfInstModel.objects.create(
 191             nfinstid=nf_inst_id,
 192             nf_name=self.nf_name,
 193             vnf_id=self.vnf_id,
 194             vnfm_inst_id=self.vnfm_inst_id,
 195             ns_inst_id=self.ns_inst_id,
 196             max_cpu='14',
 197             max_ram='12296',
 198             max_hd='101',
 199             max_shd="20",
 200             max_net=10,
 201             status='active',
 202             mnfinstid=nf_uuid,
 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"}}')
 209
 210         # Create a third vnf instance
 211         nf_inst_id = "242"
 212         package_id = "nf_hw_hss"
 213         nf_uuid = "ab34-3g5j-de13-aa85-id93"
 214
 215         NfInstModel.objects.create(
 216             nfinstid=nf_inst_id,
 217             nf_name=self.nf_name,
 218             vnf_id=self.vnf_id,
 219             vnfm_inst_id=self.vnfm_inst_id,
 220             ns_inst_id=self.ns_inst_id,
 221             max_cpu='14',
 222             max_ram='12296',
 223             max_hd='101',
 224             max_shd="20",
 225             max_net=10,
 226             status='active',
 227             mnfinstid=nf_uuid,
 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"}}')
 234
 235     def tearDown(self):
 236         NSInstModel().clean()
 237         NfInstModel().clean()
 238
 239     def test_get_and_check_params(self):
 240         aspect, numberOfSteps, scale_type = check_and_set_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)
 245
 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"])
 259
 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"])
 271
 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"])
 277         self.assertEqual(
 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"])
 282         self.assertEqual(
 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])
 289
 290     def test_set_scacle_vnf_instance_id_2(self):
 291         vnf_scale_info_list = [
 292             {
 293                 "vnfd_id": "error1",
 294                 "vnf_scaleAspectId": "mpu",
 295                 "numberOfSteps": "1"
 296             },
 297             {
 298                 "vnfd_id": "nf_zte_hss",
 299                 "vnf_scaleAspectId": "mpu",
 300                 "numberOfSteps": "1"
 301             }
 302         ]
 303         result = set_scacle_vnf_instance_id(vnf_scale_info_list)
 304         self.assertEqual(1, result.__len__())
 305         self.assertEqual(
 306             result[0]["numberOfSteps"],
 307             vnf_scale_info_list[0]["numberOfSteps"])
 308         self.assertEqual(
 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])
 313
 314     def test_set_scacle_vnf_instance_id_3(self):
 315         vnf_scale_info_list = [
 316             {
 317                 "vnfd_id": "error1",
 318                 "vnf_scaleAspectId": "mpu",
 319                 "numberOfSteps": "1"
 320             },
 321             {
 322                 "vnfd_id": "error2",
 323                 "vnf_scaleAspectId": "gsu",
 324                 "numberOfSteps": "1"
 325             }
 326         ]
 327         result = set_scacle_vnf_instance_id(vnf_scale_info_list)
 328         self.assertEqual(0, result.__len__())
 329
 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"])
 337
 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)
 343         self.assertEqual(
 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"])
 349         self.assertEqual(
 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"])
 355
 356     def test_get_nsdId(self):
 357         nsd_id = get_nsdId("1")
 358         self.assertEqual("23", nsd_id)
 359
 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
 364
 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__())
 368
 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
 373
 374         scale_vnf_data = None
 375         is_exception_caught = False
 376         try:
 377             scale_vnf_data = get_scale_vnf_data_info_list(
 378                 self.scaleNsData2, "1")
 379         except Exception:
 380             is_exception_caught = True
 381         self.assertTrue(is_exception_caught)
 382         self.assertIsNone(scale_vnf_data)
 383
 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()
 389
 390         scale_vnf_data = None
 391         is_exception_caught = False
 392         try:
 393             scale_vnf_data = get_scale_vnf_data_info_list(
 394                 self.scaleNsData2, "1")
 395         except Exception:
 396             is_exception_caught = True
 397         self.assertFalse(is_exception_caught)
 398         self.assertEqual(2, scale_vnf_data.__len__())