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