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["volume"]["test"] = "test"
  95         res_cache["port"]["ext_cp"] = "port1"
  96         NfInstModel(nfinstid=self.nf_inst_id,
  97                     nf_name='VNF1',
  98                     nf_desc="VNF DESC",
  99                     vnfdid="1",
 100                     netype="XGW",
 101                     vendor="ZTE",
 102                     vnfSoftwareVersion="V1",
 103                     version="V1",
 104                     package_id="2",
 105                     status='INSTANTIATED',
 106                     vnfd_model=json.dumps(const.vnfd_for_scale),
 107                     vimInfo=json.dumps({}),
 108                     resInfo=json.dumps(res_cache)).save()
 109         StorageInstModel.objects.create(
 110             storageid="1",
 111             vimid="1",
 112             resourceid="11",
 113             insttype=0,
 114             instid=self.nf_inst_id,
 115             is_predefined=1
 116         )
 117         NetworkInstModel.objects.create(
 118             networkid='1',
 119             vimid='1',
 120             resourceid='1',
 121             name='pnet_network',
 122             is_predefined=1,
 123             tenant='admin',
 124             insttype=0,
 125             instid=self.nf_inst_id
 126         )
 127         SubNetworkInstModel.objects.create(
 128             subnetworkid='1',
 129             vimid='1',
 130             resourceid='1',
 131             networkid='1',
 132             is_predefined=1,
 133             name='sub_pnet',
 134             tenant='admin',
 135             insttype=0,
 136             instid=self.nf_inst_id
 137         )
 138         PortInstModel.objects.create(
 139             portid='1',
 140             networkid='1',
 141             subnetworkid='1',
 142             vimid='1',
 143             resourceid='1',
 144             is_predefined=1,
 145             name='ext_cp',
 146             tenant='admin',
 147             insttype=0,
 148             instid=self.nf_inst_id
 149         )
 150         FlavourInstModel.objects.create(
 151             flavourid="1",
 152             vimid="1",
 153             resourceid="11",
 154             instid=self.nf_inst_id,
 155             is_predefined=1,
 156             name="Flavor_sunshine"
 157         )
 158         VmInstModel.objects.create(
 159             vmid="1",
 160             vimid="1",
 161             resourceid="11",
 162             insttype=0,
 163             instid=self.nf_inst_id,
 164             vmname="test_01",
 165             is_predefined=1,
 166             operationalstate=1
 167         )
 168         r1_apply_grant_result = [
 169             0,
 170             json.JSONEncoder().encode(const.instantiate_grant_result),
 171             '200'
 172         ]
 173         mock_call_req.side_effect = [
 174             r1_apply_grant_result,
 175         ]
 176         mock_call.side_effect = [
 177             const.c1_data_get_tenant_id,
 178             const.c7_data_create_flavor,
 179             const.c8_data_list_image,
 180             const.c9_data_create_vm,
 181             const.c10_data_get_vm
 182         ]
 183
 184         self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id)
 185         JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", )
 186
 187         ScaleVnfToLevel(
 188             {"instantiationLevelId": "instantiation_level_2"},
 189             nf_inst_id=self.nf_inst_id,
 190             job_id=self.job_id
 191         ).run()
 192
 193         NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
 194         self.assert_job_result(
 195             self.job_id,
 196             100,
 197             'Scale Vnf success.'
 198         )
 199
 200     @mock.patch.object(restcall, 'call_req')
 201     @mock.patch.object(api, 'call')
 202     def test_scale_to_inlevel_vnf_success(self, mock_call, mock_call_req):
 203         self.nf_inst_id = '6789'
 204         res_cache = {"volume": {}, "flavor": {}, "port": {}}
 205         res_cache["port"]["ext_cp"] = "port1"
 206         NfInstModel(nfinstid=self.nf_inst_id,
 207                     nf_name='VNF1',
 208                     nf_desc="VNF DESC",
 209                     vnfdid="1",
 210                     netype="XGW",
 211                     vendor="ZTE",
 212                     vnfSoftwareVersion="V1",
 213                     version="V1",
 214                     package_id="2",
 215                     status='INSTANTIATED',
 216                     vnfd_model=json.dumps(const.vnfd_for_scale),
 217                     vimInfo=json.dumps({}),
 218                     resInfo=json.dumps(res_cache)).save()
 219         StorageInstModel.objects.create(
 220             storageid="1",
 221             vimid="1",
 222             resourceid="11",
 223             insttype=0,
 224             instid=self.nf_inst_id,
 225             is_predefined=1
 226         )
 227         NetworkInstModel.objects.create(
 228             networkid='1',
 229             vimid='1',
 230             resourceid='1',
 231             name='pnet_network',
 232             is_predefined=1,
 233             tenant='admin',
 234             insttype=0,
 235             instid=self.nf_inst_id
 236         )
 237         SubNetworkInstModel.objects.create(
 238             subnetworkid='1',
 239             vimid='1',
 240             resourceid='1',
 241             networkid='1',
 242             is_predefined=1,
 243             name='sub_pnet',
 244             tenant='admin',
 245             insttype=0,
 246             instid=self.nf_inst_id
 247         )
 248         PortInstModel.objects.create(
 249             portid='1',
 250             networkid='1',
 251             subnetworkid='1',
 252             vimid='1',
 253             resourceid='1',
 254             is_predefined=1,
 255             name='ext_cp',
 256             tenant='admin',
 257             insttype=0,
 258             instid=self.nf_inst_id
 259         )
 260         FlavourInstModel.objects.create(
 261             flavourid="1",
 262             vimid="1",
 263             resourceid="11",
 264             instid=self.nf_inst_id,
 265             is_predefined=1,
 266             name="Flavor_sunshine"
 267         )
 268         VmInstModel.objects.create(
 269             vmid="1",
 270             vimid="1",
 271             resourceid="11",
 272             insttype=0,
 273             instid=self.nf_inst_id,
 274             vmname="test_01",
 275             is_predefined=1,
 276             operationalstate=1
 277         )
 278         VmInstModel.objects.create(
 279             vmid="2",
 280             vimid="1",
 281             resourceid="22",
 282             insttype=0,
 283             instid=self.nf_inst_id,
 284             vmname="test_02",
 285             is_predefined=1,
 286             operationalstate=1
 287         )
 288         VNFCInstModel.objects.create(
 289             vnfcinstanceid="1",
 290             instid=self.nf_inst_id,
 291             vmid="1"
 292         )
 293         VNFCInstModel.objects.create(
 294             vnfcinstanceid="2",
 295             instid=self.nf_inst_id,
 296             vmid="2"
 297         )
 298         r1_apply_grant_result = [
 299             0,
 300             json.JSONEncoder().encode(const.instantiate_grant_result),
 301             '200'
 302         ]
 303         mock_call_req.side_effect = [
 304             r1_apply_grant_result,
 305         ]
 306         mock_call.side_effect = [
 307             const.c1_data_get_tenant_id,
 308             const.c7_data_create_flavor,
 309             const.c8_data_list_image,
 310             const.c9_data_create_vm,
 311             const.c10_data_get_vm
 312         ]
 313
 314         self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id)
 315         JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", )
 316
 317         ScaleVnfToLevel(
 318             {"scaleInfo": {"aspectId": "sunshine_aspect",
 319                            "scaleLevel": 0}},
 320             nf_inst_id=self.nf_inst_id,
 321             job_id=self.job_id
 322         ).run()
 323
 324         NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
 325         self.assert_job_result(
 326             self.job_id,
 327             100,
 328             'Scale Vnf success.'
 329         )