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.c6_data_create_port,
 179             const.c7_data_create_flavor,
 180             const.c8_data_list_image,
 181             const.c9_data_create_vm,
 182             const.c10_data_get_vm
 183         ]
 184
 185         self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id)
 186         JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", )
 187
 188         ScaleVnfToLevel(
 189             {"instantiationLevelId": "instantiation_level_2"},
 190             nf_inst_id=self.nf_inst_id,
 191             job_id=self.job_id
 192         ).run()
 193
 194         NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
 195         self.assert_job_result(
 196             self.job_id,
 197             100,
 198             'Scale Vnf success.'
 199         )
 200
 201     @mock.patch.object(restcall, 'call_req')
 202     @mock.patch.object(api, 'call')
 203     def test_scale_to_inlevel_vnf_success(self, mock_call, mock_call_req):
 204         self.nf_inst_id = '6789'
 205         res_cache = {"volume": {}, "flavor": {}, "port": {}}
 206         res_cache["port"]["ext_cp"] = "port1"
 207         NfInstModel(nfinstid=self.nf_inst_id,
 208                     nf_name='VNF1',
 209                     nf_desc="VNF DESC",
 210                     vnfdid="1",
 211                     netype="XGW",
 212                     vendor="ZTE",
 213                     vnfSoftwareVersion="V1",
 214                     version="V1",
 215                     package_id="2",
 216                     status='INSTANTIATED',
 217                     vnfd_model=json.dumps(const.vnfd_for_scale),
 218                     vimInfo=json.dumps({}),
 219                     resInfo=json.dumps(res_cache)).save()
 220         StorageInstModel.objects.create(
 221             storageid="1",
 222             vimid="1",
 223             resourceid="11",
 224             insttype=0,
 225             instid=self.nf_inst_id,
 226             is_predefined=1
 227         )
 228         NetworkInstModel.objects.create(
 229             networkid='1',
 230             vimid='1',
 231             resourceid='1',
 232             name='pnet_network',
 233             is_predefined=1,
 234             tenant='admin',
 235             insttype=0,
 236             instid=self.nf_inst_id
 237         )
 238         SubNetworkInstModel.objects.create(
 239             subnetworkid='1',
 240             vimid='1',
 241             resourceid='1',
 242             networkid='1',
 243             is_predefined=1,
 244             name='sub_pnet',
 245             tenant='admin',
 246             insttype=0,
 247             instid=self.nf_inst_id
 248         )
 249         PortInstModel.objects.create(
 250             portid='1',
 251             networkid='1',
 252             subnetworkid='1',
 253             vimid='1',
 254             resourceid='1',
 255             is_predefined=1,
 256             name='ext_cp',
 257             tenant='admin',
 258             insttype=0,
 259             instid=self.nf_inst_id
 260         )
 261         FlavourInstModel.objects.create(
 262             flavourid="1",
 263             vimid="1",
 264             resourceid="11",
 265             instid=self.nf_inst_id,
 266             is_predefined=1,
 267             name="Flavor_sunshine"
 268         )
 269         VmInstModel.objects.create(
 270             vmid="1",
 271             vimid="1",
 272             resourceid="11",
 273             insttype=0,
 274             instid=self.nf_inst_id,
 275             vmname="test_01",
 276             is_predefined=1,
 277             operationalstate=1
 278         )
 279         VmInstModel.objects.create(
 280             vmid="2",
 281             vimid="1",
 282             resourceid="22",
 283             insttype=0,
 284             instid=self.nf_inst_id,
 285             vmname="test_02",
 286             is_predefined=1,
 287             operationalstate=1
 288         )
 289         VNFCInstModel.objects.create(
 290             vnfcinstanceid="1",
 291             instid=self.nf_inst_id,
 292             vmid="1"
 293         )
 294         VNFCInstModel.objects.create(
 295             vnfcinstanceid="2",
 296             instid=self.nf_inst_id,
 297             vmid="2"
 298         )
 299         r1_apply_grant_result = [
 300             0,
 301             json.JSONEncoder().encode(const.instantiate_grant_result),
 302             '200'
 303         ]
 304         mock_call_req.side_effect = [
 305             r1_apply_grant_result,
 306         ]
 307         mock_call.side_effect = [
 308             const.c1_data_get_tenant_id,
 309             const.c7_data_create_flavor,
 310             const.c8_data_list_image,
 311             const.c9_data_create_vm,
 312             const.c10_data_get_vm
 313         ]
 314
 315         self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id)
 316         JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", )
 317
 318         ScaleVnfToLevel(
 319             {"scaleInfo": {"aspectId": "sunshine_aspect",
 320                            "scaleLevel": 0}},
 321             nf_inst_id=self.nf_inst_id,
 322             job_id=self.job_id
 323         ).run()
 324
 325         NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
 326         self.assert_job_result(
 327             self.job_id,
 328             100,
 329             'Scale Vnf success.'
 330         )