lcm/lcm/nf/tests/test_scale_vnf_to_level.py

   1 # Copyright (C) 2019 ZTE. All Rights Reserved.
   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 import json
  15 import mock
  16 from django.test import TestCase
  17 from rest_framework import status
  18 from rest_framework.test import APIClient
  19 from . import const
  20 from lcm.pub.database.models import NfInstModel, JobStatusModel, StorageInstModel, NetworkInstModel, \
  21     SubNetworkInstModel, PortInstModel, FlavourInstModel, VmInstModel, VNFCInstModel
  22 from lcm.pub.utils import restcall
  23 from lcm.pub.vimapi import api
  24 from lcm.pub.utils.jobutil import JobUtil
  25 from lcm.nf.biz.scale_vnf_to_level import ScaleVnfToLevel
  26
  27
  28 class TestNfScaleToLevel(TestCase):
  29     def setUp(self):
  30         self.client = APIClient()
  31         self.req_data = {
  32             "instantiationLevelId": "instantiation_level_2",
  33             "scaleInfo": []
  34         }
  35         NfInstModel(nfinstid='98765',
  36                     nf_name='VNF1',
  37                     nf_desc="VNF DESC",
  38                     vnfdid="1",
  39                     netype="XGW",
  40                     vendor="ZTE",
  41                     vnfSoftwareVersion="V1",
  42                     version="V1",
  43                     package_id="2",
  44                     status='NOT_INSTANTIATED').save()
  45
  46     def tearDown(self):
  47         NfInstModel.objects.filter(nfinstid='98765').delete()
  48
  49     def assert_job_result(self, job_id, job_progress, job_detail):
  50         jobs = JobStatusModel.objects.filter(
  51             jobid=job_id,
  52             progress=job_progress,
  53             descp=job_detail
  54         )
  55         self.assertEqual(1, len(jobs))
  56
  57     def test_scale_vnf_to_level_not_found(self):
  58         url = "/api/vnflcm/v1/vnf_instances/12/scale_to_level"
  59         response = self.client.post(url,
  60                                     data=self.req_data,
  61                                     format='json')
  62         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
  63
  64     def test_scale_vnf_conflict(self):
  65         url = "/api/vnflcm/v1/vnf_instances/98765/scale_to_level"
  66         response = self.client.post(url,
  67                                     data=self.req_data,
  68                                     format='json')
  69         self.assertEqual(status.HTTP_409_CONFLICT, response.status_code)
  70
  71     def test_scale_to_level_badreq(self):
  72         NfInstModel(nfinstid='678',
  73                     nf_name='VNF1',
  74                     nf_desc="VNF DESC",
  75                     vnfdid="1",
  76                     netype="XGW",
  77                     vendor="ZTE",
  78                     vnfSoftwareVersion="V1",
  79                     version="V1",
  80                     package_id="2",
  81                     status='INSTANTIATED').save()
  82         url = "/api/vnflcm/v1/vnf_instances/678/scale_to_level"
  83         response = self.client.post(url,
  84                                     data={},
  85                                     format='json')
  86         NfInstModel.objects.filter(nfinstid='678').delete()
  87         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
  88
  89     @mock.patch.object(restcall, 'call_req')
  90     @mock.patch.object(api, 'call')
  91     def test_scale_out_vnf_success(self, mock_call, mock_call_req):
  92         self.nf_inst_id = '6789'
  93         res_cache = {"volume": {}, "flavor": {}, "port": {}}
  94         res_cache["port"]["ext_cp"] = "port1"
  95         NfInstModel(nfinstid=self.nf_inst_id,
  96                     nf_name='VNF1',
  97                     nf_desc="VNF DESC",
  98                     vnfdid="1",
  99                     netype="XGW",
 100                     vendor="ZTE",
 101                     vnfSoftwareVersion="V1",
 102                     version="V1",
 103                     package_id="2",
 104                     status='INSTANTIATED',
 105                     vnfd_model=json.dumps(const.vnfd_for_scale),
 106                     vimInfo=json.dumps({}),
 107                     resInfo=json.dumps(res_cache)).save()
 108         StorageInstModel.objects.create(
 109             storageid="1",
 110             vimid="1",
 111             resourceid="11",
 112             insttype=0,
 113             instid=self.nf_inst_id,
 114             is_predefined=1
 115         )
 116         NetworkInstModel.objects.create(
 117             networkid='1',
 118             vimid='1',
 119             resourceid='1',
 120             name='pnet_network',
 121             is_predefined=1,
 122             tenant='admin',
 123             insttype=0,
 124             instid=self.nf_inst_id
 125         )
 126         SubNetworkInstModel.objects.create(
 127             subnetworkid='1',
 128             vimid='1',
 129             resourceid='1',
 130             networkid='1',
 131             is_predefined=1,
 132             name='sub_pnet',
 133             tenant='admin',
 134             insttype=0,
 135             instid=self.nf_inst_id
 136         )
 137         PortInstModel.objects.create(
 138             portid='1',
 139             networkid='1',
 140             subnetworkid='1',
 141             vimid='1',
 142             resourceid='1',
 143             is_predefined=1,
 144             name='ext_cp',
 145             tenant='admin',
 146             insttype=0,
 147             instid=self.nf_inst_id
 148         )
 149         FlavourInstModel.objects.create(
 150             flavourid="1",
 151             vimid="1",
 152             resourceid="11",
 153             instid=self.nf_inst_id,
 154             is_predefined=1,
 155             name="Flavor_sunshine"
 156         )
 157         VmInstModel.objects.create(
 158             vmid="1",
 159             vimid="1",
 160             resourceid="11",
 161             insttype=0,
 162             instid=self.nf_inst_id,
 163             vmname="test_01",
 164             is_predefined=1,
 165             operationalstate=1
 166         )
 167         r1_apply_grant_result = [
 168             0,
 169             json.JSONEncoder().encode(const.instantiate_grant_result),
 170             '200'
 171         ]
 172         mock_call_req.side_effect = [
 173             r1_apply_grant_result,
 174         ]
 175         mock_call.side_effect = [
 176             const.c1_data_get_tenant_id,
 177             const.c7_data_create_flavor,
 178             const.c8_data_list_image,
 179             const.c9_data_create_vm,
 180             const.c10_data_get_vm
 181         ]
 182
 183         self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id)
 184         JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", )
 185
 186         ScaleVnfToLevel(
 187             {"instantiationLevelId": "instantiation_level_2"},
 188             nf_inst_id=self.nf_inst_id,
 189             job_id=self.job_id
 190         ).run()
 191
 192         NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
 193         self.assert_job_result(
 194             self.job_id,
 195             100,
 196             'Scale Vnf success.'
 197         )
 198
 199     @mock.patch.object(restcall, 'call_req')
 200     @mock.patch.object(api, 'call')
 201     def test_scale_to_inlevel_vnf_success(self, mock_call, mock_call_req):
 202         self.nf_inst_id = '6789'
 203         res_cache = {"volume": {}, "flavor": {}, "port": {}}
 204         res_cache["port"]["ext_cp"] = "port1"
 205         NfInstModel(nfinstid=self.nf_inst_id,
 206                     nf_name='VNF1',
 207                     nf_desc="VNF DESC",
 208                     vnfdid="1",
 209                     netype="XGW",
 210                     vendor="ZTE",
 211                     vnfSoftwareVersion="V1",
 212                     version="V1",
 213                     package_id="2",
 214                     status='INSTANTIATED',
 215                     vnfd_model=json.dumps(const.vnfd_for_scale),
 216                     vimInfo=json.dumps({}),
 217                     resInfo=json.dumps(res_cache)).save()
 218         StorageInstModel.objects.create(
 219             storageid="1",
 220             vimid="1",
 221             resourceid="11",
 222             insttype=0,
 223             instid=self.nf_inst_id,
 224             is_predefined=1
 225         )
 226         NetworkInstModel.objects.create(
 227             networkid='1',
 228             vimid='1',
 229             resourceid='1',
 230             name='pnet_network',
 231             is_predefined=1,
 232             tenant='admin',
 233             insttype=0,
 234             instid=self.nf_inst_id
 235         )
 236         SubNetworkInstModel.objects.create(
 237             subnetworkid='1',
 238             vimid='1',
 239             resourceid='1',
 240             networkid='1',
 241             is_predefined=1,
 242             name='sub_pnet',
 243             tenant='admin',
 244             insttype=0,
 245             instid=self.nf_inst_id
 246         )
 247         PortInstModel.objects.create(
 248             portid='1',
 249             networkid='1',
 250             subnetworkid='1',
 251             vimid='1',
 252             resourceid='1',
 253             is_predefined=1,
 254             name='ext_cp',
 255             tenant='admin',
 256             insttype=0,
 257             instid=self.nf_inst_id
 258         )
 259         FlavourInstModel.objects.create(
 260             flavourid="1",
 261             vimid="1",
 262             resourceid="11",
 263             instid=self.nf_inst_id,
 264             is_predefined=1,
 265             name="Flavor_sunshine"
 266         )
 267         VmInstModel.objects.create(
 268             vmid="1",
 269             vimid="1",
 270             resourceid="11",
 271             insttype=0,
 272             instid=self.nf_inst_id,
 273             vmname="test_01",
 274             is_predefined=1,
 275             operationalstate=1
 276         )
 277         VmInstModel.objects.create(
 278             vmid="2",
 279             vimid="1",
 280             resourceid="22",
 281             insttype=0,
 282             instid=self.nf_inst_id,
 283             vmname="test_02",
 284             is_predefined=1,
 285             operationalstate=1
 286         )
 287         VNFCInstModel.objects.create(
 288             vnfcinstanceid="1",
 289             instid=self.nf_inst_id,
 290             vmid="1"
 291         )
 292         VNFCInstModel.objects.create(
 293             vnfcinstanceid="2",
 294             instid=self.nf_inst_id,
 295             vmid="2"
 296         )
 297         r1_apply_grant_result = [
 298             0,
 299             json.JSONEncoder().encode(const.instantiate_grant_result),
 300             '200'
 301         ]
 302         mock_call_req.side_effect = [
 303             r1_apply_grant_result,
 304         ]
 305         mock_call.side_effect = [
 306             const.c1_data_get_tenant_id,
 307             const.c7_data_create_flavor,
 308             const.c8_data_list_image,
 309             const.c9_data_create_vm,
 310             const.c10_data_get_vm
 311         ]
 312
 313         self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id)
 314         JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", )
 315
 316         ScaleVnfToLevel(
 317             {"scaleInfo": {"aspectId": "sunshine_aspect",
 318                            "scaleLevel": 0}},
 319             nf_inst_id=self.nf_inst_id,
 320             job_id=self.job_id
 321         ).run()
 322
 323         NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
 324         self.assert_job_result(
 325             self.job_id,
 326             100,
 327             'Scale Vnf success.'
 328         )