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