lcm/pub/tests/test_scaleaspect.py

   1 from django.test import TestCase
   2 from lcm.pub.utils.scaleaspect import get_json_data
   3 from lcm.pub.utils.scaleaspect import get_nsdId
   4 from lcm.pub.utils.scaleaspect import get_scale_vnf_data_from_json
   5 from lcm.pub.utils.scaleaspect import get_scale_vnf_data_info_list
   6 from lcm.pub.utils.scaleaspect import set_scacle_vnf_instance_id
   7 from lcm.pub.utils.scaleaspect import get_and_check_params
   8 from lcm.pub.utils.scaleaspect import set_scaleVnfData_type
   9 from lcm.pub.database.models import NfInstModel
  10 from lcm.pub.database.models import NSInstModel
  11 from lcm.pub.msapi import catalog
  12 from lcm.pub.utils.timeutil import now_time
  13 import os
  14 import mock
  15
  16
  17 class TestScaleAspect(TestCase):
  18
  19     def setUp(self):
  20         self.init_scaling_map_json()
  21         self.initInstModel()
  22
  23         self.init_scale_ns_data()
  24
  25         self.vnf_scale_info_list = [
  26             {
  27                 "vnfd_id": "nf_zte_cscf",
  28                 "vnf_scaleAspectId": "mpu",
  29                 "numberOfSteps": "1"
  30             },
  31             {
  32                 "vnfd_id": "nf_zte_hss",
  33                 "vnf_scaleAspectId": "gsu",
  34                 "numberOfSteps": "2"
  35             }
  36         ]
  37
  38     def init_scale_ns_data(self):
  39         self.ns_scale_aspect = "TIC_EDGE_IMS"
  40         self.ns_scale_steps = "1"
  41         self.ns_scale_direction = "SCALE_IN"
  42         self.scaleNsData = {
  43             "aspectId": self.ns_scale_aspect,
  44             "numberOfSteps": self.ns_scale_steps,
  45             "scalingDirection": self.ns_scale_direction
  46         }
  47
  48     def init_scaling_map_json(self):
  49         curdir_path = os.path.dirname(
  50             os.path.dirname(
  51                 os.path.dirname(
  52                     os.path.abspath(__file__))))
  53         filename = curdir_path + "/ns/data/scalemapping.json"
  54         self.scaling_map_json = get_json_data(filename)
  55
  56     def initInstModel(self):
  57         self.nsd_id = "23"
  58         self.ns_inst_id = "1"
  59         self.ns_name = "ns_1"
  60         self.ns_package_id = "ns_zte"
  61         self.description = "ns_zte"
  62         self.global_customer_id = "global_customer_id"
  63         self.service_type = "service_role"
  64
  65         NSInstModel(id=self.ns_inst_id,
  66                     name=self.ns_name,
  67                     nspackage_id=self.ns_package_id,
  68                     nsd_id=self.nsd_id,
  69                     description=self.description,
  70                     status='empty',
  71                     lastuptime=now_time(),
  72                     global_customer_id=self.global_customer_id,
  73                     service_type=self.service_type).save()
  74
  75         self.nf_inst_id = "231"
  76         self.ns_inst_id = "1"
  77         self.nf_name = "name_1"
  78         self.vnf_id = "1"
  79         self.vnfm_inst_id = "1"
  80         self.package_id = "nf_zte_cscf"
  81         self.nf_uuid = "abc34-345a-de13-ab85-ijs9"
  82
  83         NfInstModel.objects.create(
  84             nfinstid=self.nf_inst_id,
  85             nf_name=self.nf_name,
  86             vnf_id=self.vnf_id,
  87             vnfm_inst_id=self.vnfm_inst_id,
  88             ns_inst_id=self.ns_inst_id,
  89             max_cpu='14',
  90             max_ram='12296',
  91             max_hd='101',
  92             max_shd="20",
  93             max_net=10,
  94             status='active',
  95             mnfinstid=self.nf_uuid,
  96             package_id=self.package_id,
  97             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
  98             '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
  99             '"productType": "CN","vnfType": "PGW",'
 100             '"description": "PGW VNFD description",'
 101             '"isShared":true,"vnfExtendType":"driver"}}')
 102
 103         # Create a second vnf instance
 104         self.nf_inst_id = "232"
 105         self.package_id = "nf_zte_hss"
 106         self.nf_uuid = "abc34-3g5a-de13-ab85-ijs3"
 107
 108         NfInstModel.objects.create(
 109             nfinstid=self.nf_inst_id,
 110             nf_name=self.nf_name,
 111             vnf_id=self.vnf_id,
 112             vnfm_inst_id=self.vnfm_inst_id,
 113             ns_inst_id=self.ns_inst_id,
 114             max_cpu='14',
 115             max_ram='12296',
 116             max_hd='101',
 117             max_shd="20",
 118             max_net=10,
 119             status='active',
 120             mnfinstid=self.nf_uuid,
 121             package_id=self.package_id,
 122             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 123                        '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 124                        '"productType": "CN","vnfType": "PGW",'
 125                        '"description": "PGW VNFD description",'
 126                        '"isShared":true,"vnfExtendType":"driver"}}')
 127
 128     def add_another_nf_instance(self):
 129         # Create a third vnf instance
 130         nf_inst_id = "233"
 131         package_id = "nf_zte_hss"
 132         nf_uuid = "ab34-3g5j-de13-ab85-ij93"
 133
 134         NfInstModel.objects.create(
 135             nfinstid=nf_inst_id,
 136             nf_name=self.nf_name,
 137             vnf_id=self.vnf_id,
 138             vnfm_inst_id=self.vnfm_inst_id,
 139             ns_inst_id=self.ns_inst_id,
 140             max_cpu='14',
 141             max_ram='12296',
 142             max_hd='101',
 143             max_shd="20",
 144             max_net=10,
 145             status='active',
 146             mnfinstid=nf_uuid,
 147             package_id=package_id,
 148             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 149                        '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 150                        '"productType": "CN","vnfType": "PGW",'
 151                        '"description": "PGW VNFD description",'
 152                        '"isShared":true,"vnfExtendType":"driver"}}')
 153
 154     def tearDown(self):
 155         NSInstModel().clean()
 156         NfInstModel().clean()
 157
 158     def test_get_and_check_params(self):
 159         aspect, numberOfSteps, scale_type = get_and_check_params(
 160             self.scaleNsData, "1")
 161         self.assertEqual(aspect, self.ns_scale_aspect)
 162         self.assertEqual(numberOfSteps, self.ns_scale_steps)
 163         self.assertEqual(scale_type, self.ns_scale_direction)
 164
 165     def test_get_scale_vnf_data_from_json(self):
 166         vnf_data_package = get_scale_vnf_data_from_json(
 167             self.scaling_map_json, "23", "TIC_EDGE_IMS", "1")
 168         self.assertIsNotNone(vnf_data_package)
 169         self.assertEqual(2, vnf_data_package.__len__())
 170         self.assertIsNotNone(vnf_data_package)
 171         self.assertEqual(2, vnf_data_package.__len__())
 172         self.assertEqual("nf_zte_cscf", vnf_data_package[0]["vnfd_id"])
 173         self.assertEqual("1", vnf_data_package[0]["numberOfSteps"])
 174         self.assertEqual("gsu", vnf_data_package[0]["vnf_scaleAspectId"])
 175         self.assertEqual("nf_zte_hss", vnf_data_package[1]["vnfd_id"])
 176         self.assertEqual("3", vnf_data_package[1]["numberOfSteps"])
 177         self.assertEqual("gpu", vnf_data_package[1]["vnf_scaleAspectId"])
 178
 179     def test_get_scale_vnf_data_from_json_2(self):
 180         vnf_data_package = get_scale_vnf_data_from_json(
 181             self.scaling_map_json, "23", "TIC_EDGE_IMS", "2")
 182         self.assertIsNotNone(vnf_data_package)
 183         self.assertEqual(2, vnf_data_package.__len__())
 184         self.assertEqual("nf_zte_cscf", vnf_data_package[0]["vnfd_id"])
 185         self.assertEqual("2", vnf_data_package[0]["numberOfSteps"])
 186         self.assertEqual("mpu", vnf_data_package[0]["vnf_scaleAspectId"])
 187         self.assertEqual("nf_zte_hss", vnf_data_package[1]["vnfd_id"])
 188         self.assertEqual("4", vnf_data_package[1]["numberOfSteps"])
 189         self.assertEqual("mpu", vnf_data_package[1]["vnf_scaleAspectId"])
 190
 191     def test_set_scacle_vnf_instance_id(self):
 192         result = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
 193         self.assertEqual(2, result.__len__())
 194         self.assertEqual(result[0]["numberOfSteps"],
 195                          self.vnf_scale_info_list[0]["numberOfSteps"])
 196         self.assertEqual(
 197             result[0]["vnf_scaleAspectId"],
 198             self.vnf_scale_info_list[0]["vnf_scaleAspectId"])
 199         self.assertEqual(result[1]["numberOfSteps"],
 200                          self.vnf_scale_info_list[1]["numberOfSteps"])
 201         self.assertEqual(
 202             result[1]["vnf_scaleAspectId"],
 203             self.vnf_scale_info_list[1]["vnf_scaleAspectId"])
 204         self.assertEqual("231", result[0]["vnfInstanceId"])
 205         self.assertEqual("232", result[1]["vnfInstanceId"])
 206         self.assertNotIn("vnfd_id", result[0])
 207         self.assertNotIn("vnfd_id", result[1])
 208
 209     def test_set_scacle_vnf_instance_id_2(self):
 210         vnf_scale_info_list = [
 211             {
 212                 "vnfd_id": "error1",
 213                 "vnf_scaleAspectId": "mpu",
 214                 "numberOfSteps": "1"
 215             },
 216             {
 217                 "vnfd_id": "nf_zte_hss",
 218                 "vnf_scaleAspectId": "mpu",
 219                 "numberOfSteps": "1"
 220             }
 221         ]
 222         result = set_scacle_vnf_instance_id(vnf_scale_info_list)
 223         self.assertEqual(1, result.__len__())
 224         self.assertEqual(
 225             result[0]["numberOfSteps"],
 226             vnf_scale_info_list[0]["numberOfSteps"])
 227         self.assertEqual(
 228             result[0]["vnf_scaleAspectId"],
 229             vnf_scale_info_list[0]["vnf_scaleAspectId"])
 230         self.assertEqual("232", result[0]["vnfInstanceId"])
 231         self.assertNotIn("vnfd_id", result[0])
 232
 233     def test_set_scacle_vnf_instance_id_3(self):
 234         vnf_scale_info_list = [
 235             {
 236                 "vnfd_id": "error1",
 237                 "vnf_scaleAspectId": "mpu",
 238                 "numberOfSteps": "1"
 239             },
 240             {
 241                 "vnfd_id": "error2",
 242                 "vnf_scaleAspectId": "gsu",
 243                 "numberOfSteps": "1"
 244             }
 245         ]
 246         result = set_scacle_vnf_instance_id(vnf_scale_info_list)
 247         self.assertEqual(0, result.__len__())
 248
 249     def test_set_scacle_vnf_instance_id_4(self):
 250         self.add_another_nf_instance()
 251         result = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
 252         self.assertEqual(3, result.__len__())
 253         self.assertEqual("231", result[0]["vnfInstanceId"])
 254         self.assertEqual("232", result[1]["vnfInstanceId"])
 255         self.assertEqual("233", result[2]["vnfInstanceId"])
 256
 257     def test_set_scaleVnfData_type(self):
 258         vnf_scale_list = set_scacle_vnf_instance_id(self.vnf_scale_info_list)
 259         result = set_scaleVnfData_type(vnf_scale_list, self.ns_scale_direction)
 260         self.assertEqual(2, result.__len__())
 261         self.assertNotIn("scaleByStepData", result)
 262         self.assertEqual(
 263             self.ns_scale_direction,
 264             result[0]["scaleByStepData"]["type"])
 265         self.assertEqual("mpu", result[0]["scaleByStepData"]["aspectId"])
 266         self.assertNotIn("vnf_scaleAspectId", result[0]["scaleByStepData"])
 267         self.assertEqual("1", result[0]["scaleByStepData"]["numberOfSteps"])
 268         self.assertEqual(
 269             self.ns_scale_direction,
 270             result[1]["scaleByStepData"]["type"])
 271         self.assertEqual("gsu", result[1]["scaleByStepData"]["aspectId"])
 272         self.assertNotIn("vnf_scaleAspectId", result[1]["scaleByStepData"])
 273         self.assertEqual("2", result[1]["scaleByStepData"]["numberOfSteps"])
 274
 275     def test_get_nsdId(self):
 276         nsd_id = get_nsdId("1")
 277         self.assertEqual("23", nsd_id)
 278
 279     @mock.patch.object(catalog, 'get_scalingmap_json_package')
 280     def test_get_scale_vnf_data_info_list(
 281             self, mock_get_scalingmap_json_package):
 282         mock_get_scalingmap_json_package.return_value = self.scaling_map_json
 283
 284         scale_vnf_data = get_scale_vnf_data_info_list(self.scaleNsData, "1")
 285         self.assertIsNotNone(scale_vnf_data)
 286         self.assertEqual(2, scale_vnf_data.__len__())