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