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