# Copyright (C) 2019 ZTE. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import json import mock from . import const from django.test import TestCase from rest_framework.test import APIClient from lcm.pub.database.models import NfInstModel, JobStatusModel, StorageInstModel, NetworkInstModel, \ SubNetworkInstModel, PortInstModel, FlavourInstModel, VmInstModel from lcm.pub.utils import restcall from lcm.pub.vimapi import api from lcm.pub.utils.jobutil import JobUtil from lcm.nf.biz.scale_vnf import ScaleVnf class TestNfScale(TestCase): def setUp(self): self.client = APIClient() NfInstModel(nfinstid='12345', nf_name='VNF1', nf_desc="VNF DESC", vnfdid="1", netype="XGW", vendor="ZTE", vnfSoftwareVersion="V1", version="V1", package_id="2", status='NOT_INSTANTIATED').save() self.req_data = { "type": "SCALE_IN", "aspectId": "sunshine_aspect" } def tearDown(self): NfInstModel.objects.filter(nfinstid='12345').delete() def assert_job_result(self, job_id, job_progress, job_detail): jobs = JobStatusModel.objects.filter( jobid=job_id, progress=job_progress, descp=job_detail ) self.assertEqual(1, len(jobs)) # def test_scale_vnf_not_found(self): # url = "/api/vnflcm/v1/vnf_instances/12/scale" # response = self.client.post(url, # data=self.req_data, # format='json') # self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code) # # def test_scale_vnf_conflict(self): # url = "/api/vnflcm/v1/vnf_instances/12345/scale" # response = self.client.post(url, # data=self.req_data, # format='json') # self.assertEqual(status.HTTP_409_CONFLICT, response.status_code) # # def test_scale_vnf_badreq(self): # NfInstModel(nfinstid='678', # nf_name='VNF1', # nf_desc="VNF DESC", # vnfdid="1", # netype="XGW", # vendor="ZTE", # vnfSoftwareVersion="V1", # version="V1", # package_id="2", # status='INSTANTIATED').save() # url = "/api/vnflcm/v1/vnf_instances/678/scale" # response = self.client.post(url, # data={}, # format='json') # NfInstModel.objects.filter(nfinstid='678').delete() # self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) @mock.patch.object(restcall, 'call_req') @mock.patch.object(api, 'call') def test_scale_out_vnf_success(self, mock_call, mock_call_req): self.nf_inst_id = '6789' res_cache = {"volume": {}, "flavor": {}, "port": {}} res_cache["volume"]["test"] = "test" # res_cache["flavor"]["vdu1Id"] = "flavor1" # res_cache["port"]["ext_cp"] = "port1" NfInstModel(nfinstid=self.nf_inst_id, nf_name='VNF1', nf_desc="VNF DESC", vnfdid="1", netype="XGW", vendor="ZTE", vnfSoftwareVersion="V1", version="V1", package_id="2", status='INSTANTIATED', vnfd_model=json.dumps(const.vnfd_for_scale), vimInfo=json.dumps({}), resInfo=json.dumps(res_cache)).save() StorageInstModel.objects.create( storageid="1", vimid="1", resourceid="11", insttype=0, instid=self.nf_inst_id, is_predefined=1 ) NetworkInstModel.objects.create( networkid='1', vimid='1', resourceid='1', name='pnet_network', is_predefined=1, tenant='admin', insttype=0, instid=self.nf_inst_id ) SubNetworkInstModel.objects.create( subnetworkid='1', vimid='1', resourceid='1', networkid='1', is_predefined=1, name='sub_pnet', tenant='admin', insttype=0, instid=self.nf_inst_id ) PortInstModel.objects.create( portid='1', networkid='1', subnetworkid='1', vimid='1', resourceid='1', is_predefined=1, name='ext_cp', tenant='admin', insttype=0, instid=self.nf_inst_id ) FlavourInstModel.objects.create( flavourid="1", vimid="1", resourceid="11", instid=self.nf_inst_id, is_predefined=1, name="Flavor_sunshine" ) VmInstModel.objects.create( vmid="1", vimid="1", resourceid="11", insttype=0, instid=self.nf_inst_id, vmname="test_01", is_predefined=1, operationalstate=1 ) r1_apply_grant_result = [ 0, json.JSONEncoder().encode(const.instantiate_grant_result), '200' ] mock_call_req.side_effect = [ r1_apply_grant_result, ] mock_call.side_effect = [ const.c1_data_get_tenant_id, const.c6_data_create_port, const.c7_data_create_flavor, const.c8_data_list_image, const.c9_data_create_vm, const.c10_data_get_vm ] self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id) JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", ) ScaleVnf( {"type": "SCALE_OUT", "aspectId": "sunshine_aspect"}, nf_inst_id=self.nf_inst_id, job_id=self.job_id ).run() NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete() # print([{job.progress: job.descp} for job in JobStatusModel.objects.filter( jobid=self.job_id)]) self.assert_job_result( self.job_id, 100, 'Scale Vnf success.' ) # @mock.patch.object(restcall, 'call_req') # @mock.patch.object(api, 'call') # def test_scale_in_vnf_success(self, mock_call, mock_call_req): # self.nf_inst_id = '6789' # res_cache = {"volume": {}, "flavor": {}, "port": {}} # # res_cache["volume"]["volume_storage1"] = "vol1" # # res_cache["flavor"]["vdu1Id"] = "flavor1" # res_cache["port"]["ext_cp"] = "port1" # NfInstModel(nfinstid=self.nf_inst_id, # nf_name='VNF1', # nf_desc="VNF DESC", # vnfdid="1", # netype="XGW", # vendor="ZTE", # vnfSoftwareVersion="V1", # version="V1", # package_id="2", # status='INSTANTIATED', # vnfd_model=json.dumps(const.vnfd_for_scale), # vimInfo=json.dumps({}), # resInfo=json.dumps(res_cache)).save() # StorageInstModel.objects.create( # storageid="1", # vimid="1", # resourceid="11", # insttype=0, # instid=self.nf_inst_id, # is_predefined=1 # ) # NetworkInstModel.objects.create( # networkid='1', # vimid='1', # resourceid='1', # name='pnet_network', # is_predefined=1, # tenant='admin', # insttype=0, # instid=self.nf_inst_id # ) # SubNetworkInstModel.objects.create( # subnetworkid='1', # vimid='1', # resourceid='1', # networkid='1', # is_predefined=1, # name='sub_pnet', # tenant='admin', # insttype=0, # instid=self.nf_inst_id # ) # PortInstModel.objects.create( # portid='1', # networkid='1', # subnetworkid='1', # vimid='1', # resourceid='1', # is_predefined=1, # name='ext_cp', # tenant='admin', # insttype=0, # instid=self.nf_inst_id # ) # FlavourInstModel.objects.create( # flavourid="1", # vimid="1", # resourceid="11", # instid=self.nf_inst_id, # is_predefined=1, # name="Flavor_sunshine" # ) # VmInstModel.objects.create( # vmid="1", # vimid="1", # resourceid="11", # insttype=0, # instid=self.nf_inst_id, # vmname="test_01", # is_predefined=1, # operationalstate=1 # ) # VmInstModel.objects.create( # vmid="2", # vimid="1", # resourceid="22", # insttype=0, # instid=self.nf_inst_id, # vmname="test_02", # is_predefined=1, # operationalstate=1 # ) # VNFCInstModel.objects.create( # vnfcinstanceid="1", # instid=self.nf_inst_id, # vmid="1" # ) # VNFCInstModel.objects.create( # vnfcinstanceid="2", # instid=self.nf_inst_id, # vmid="2" # ) # r1_apply_grant_result = [ # 0, # json.JSONEncoder().encode(const.instantiate_grant_result), # '200' # ] # mock_call_req.side_effect = [ # r1_apply_grant_result, # ] # mock_call.side_effect = [ # const.c1_data_get_tenant_id, # const.c7_data_create_flavor, # const.c8_data_list_image, # const.c9_data_create_vm, # const.c10_data_get_vm # ] # # self.job_id = JobUtil.create_job('NF', 'SCALE', self.nf_inst_id) # JobUtil.add_job_status(self.job_id, 0, "VNF_SCALE_READY", ) # # ScaleVnf( # {"type": "SCALE_IN", # "aspectId": "sunshine_aspect"}, # nf_inst_id=self.nf_inst_id, # job_id=self.job_id # ).run() # # NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete() # self.assert_job_result( # self.job_id, # 100, # 'Scale Vnf success.' # )