lcm/pub/tests/test_scaleaspect.py

   1 from django.test import TestCase
   2 from lcm.pub.utils.scaleaspect import get_scale_vnf_data_package
   3 from lcm.pub.utils.scaleaspect import get_vnf_scale_info_package
   4 from lcm.pub.utils.scaleaspect import get_vnf_data_package
   5 from lcm.pub.utils.scaleaspect import get_json_data
   6 from lcm.pub.utils.scaleaspect import get_nsdId
   7 from lcm.pub.utils.scaleaspect import deal_vnf_scale_info
   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.scaleNsData = {
  29             "aspectId": "TIC_EDGE_IMS",
  30             "numberOfSteps": "1",
  31             "scalingDirection": "UP"
  32         }
  33
  34     def initInstModel(self):
  35         self.nsd_id = "23"
  36         self.ns_inst_id = "1"
  37         self.ns_name = "ns_1"
  38         self.ns_package_id = "ns_zte"
  39         self.description = "ns_zte"
  40         self.global_customer_id = "global_customer_id"
  41         self.service_type = "service_role"
  42
  43         NSInstModel(id=self.ns_inst_id,
  44                     name=self.ns_name,
  45                     nspackage_id=self.ns_package_id,
  46                     nsd_id=self.nsd_id,
  47                     description=self.description,
  48                     status='empty',
  49                     lastuptime=now_time(),
  50                     global_customer_id=self.global_customer_id,
  51                     service_type=self.service_type).save()
  52
  53         self.nf_inst_id = "231"
  54         self.ns_inst_id = "1"
  55         self.nf_name = "name_1"
  56         self.vnf_id = "1"
  57         self.vnfm_inst_id = "1"
  58         self.package_id = "nf_zte_cscf"
  59         self.nf_uuid = "abc34-345a-de13-ab85-ijs9"
  60
  61         NfInstModel.objects.create(
  62             nfinstid=self.nf_inst_id,
  63             nf_name=self.nf_name,
  64             vnf_id=self.vnf_id,
  65             vnfm_inst_id=self.vnfm_inst_id,
  66             ns_inst_id=self.ns_inst_id,
  67             max_cpu='14',
  68             max_ram='12296',
  69             max_hd='101',
  70             max_shd="20",
  71             max_net=10,
  72             status='active',
  73             mnfinstid=self.nf_uuid,
  74             package_id=self.package_id,
  75             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
  76             '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
  77             '"productType": "CN","vnfType": "PGW",'
  78             '"description": "PGW VNFD description",'
  79             '"isShared":true,"vnfExtendType":"driver"}}')
  80
  81         # Create a second vnf instance
  82         self.nf_inst_id = "232"
  83         self.package_id = "nf_zte_hss"
  84         self.nf_uuid = "abc34-3g5a-de13-ab85-ijs3"
  85
  86         NfInstModel.objects.create(
  87             nfinstid=self.nf_inst_id,
  88             nf_name=self.nf_name,
  89             vnf_id=self.vnf_id,
  90             vnfm_inst_id=self.vnfm_inst_id,
  91             ns_inst_id=self.ns_inst_id,
  92             max_cpu='14',
  93             max_ram='12296',
  94             max_hd='101',
  95             max_shd="20",
  96             max_net=10,
  97             status='active',
  98             mnfinstid=self.nf_uuid,
  99             package_id=self.package_id,
 100             vnfd_model='{"metadata": {"vnfdId": "1","vnfdName": "PGW001",'
 101                        '"vnfProvider": "zte","vnfdVersion": "V00001","vnfVersion": "V5.10.20",'
 102                        '"productType": "CN","vnfType": "PGW",'
 103                        '"description": "PGW VNFD description",'
 104                        '"isShared":true,"vnfExtendType":"driver"}}')
 105
 106     def tearDown(self):
 107         NSInstModel().clean()
 108         NfInstModel().clean()
 109
 110     def test_get_vnf_data_package(self):
 111         vnf_data_package = get_vnf_data_package(
 112             self.scaling_map_json, "1", "TIC_EDGE_IMS", "1", "IN")
 113         self.assertIsNotNone(vnf_data_package)
 114         self.assertEqual(2, vnf_data_package.__len__())
 115
 116     def test_get_vnf_scale_info_package(self):
 117         scale_vnf_info_list = get_vnf_scale_info_package(
 118             self.scaling_map_json, "23", "TIC_EDGE_IMS", "1")
 119         self.assertIsNotNone(scale_vnf_info_list)
 120         self.assertEqual(2, scale_vnf_info_list.__len__())
 121
 122     @mock.patch.object(catalog, 'get_scalingmap_json_package')
 123     def test_get_scale_vnf_data_package(
 124             self, mock_get_scalingmap_json_package):
 125         mock_get_scalingmap_json_package.return_value = self.scaling_map_json
 126
 127         scale_vnf_data = get_scale_vnf_data_package(self.scaleNsData, "1")
 128         self.assertIsNotNone(scale_vnf_data)
 129         self.assertEqual(2, scale_vnf_data.__len__())
 130
 131     def test_deal_vnf_scale_info(self):
 132         vnf_scale_info_list = [
 133             {
 134                 "vnfd_id": "nf_zte_cscf",
 135                 "vnf_scaleAspectId": "mpu",
 136                 "numberOfSteps": "1"
 137             },
 138             {
 139                 "vnfd_id": "nf_zte_hss",
 140                 "vnf_scaleAspectId": "mpu",
 141                 "numberOfSteps": "1"
 142             }
 143         ]
 144         result = deal_vnf_scale_info(vnf_scale_info_list)
 145         self.assertEqual(result[0]["numberOfSteps"], vnf_scale_info_list[0]["numberOfSteps"])
 146         self.assertEqual(result[0]["vnf_scaleAspectId"], vnf_scale_info_list[0]["vnf_scaleAspectId"])
 147         self.assertEqual(result[1]["numberOfSteps"], vnf_scale_info_list[0]["numberOfSteps"])
 148         self.assertEqual(result[1]["vnf_scaleAspectId"], vnf_scale_info_list[0]["vnf_scaleAspectId"])
 149         self.assertEqual("231", result[0]["vnfInstanceId"])
 150         self.assertEqual("232", result[1]["vnfInstanceId"])
 151         self.assertNotIn("vnfd_id", result[0])
 152         self.assertNotIn("vnfd_id", result[1])
 153
 154     def test_get_nsdId(self):
 155         nsd_id = get_nsdId("1")
 156         self.assertEqual("23", nsd_id)