1 # Copyright (C) 2019 ZTE. All Rights Reserved.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
17 from django.test import TestCase
18 from rest_framework.test import APIClient
19 from lcm.pub.database.models import NfInstModel, JobStatusModel, StorageInstModel, NetworkInstModel, \
20 SubNetworkInstModel, PortInstModel, FlavourInstModel, VmInstModel
21 from lcm.pub.utils import restcall
22 from lcm.pub.vimapi import api
23 from lcm.pub.utils.jobutil import JobUtil
24 from lcm.nf.biz.scale_vnf import ScaleVnf
27 class TestNfScale(TestCase):
29 self.client = APIClient()
30 NfInstModel(nfinstid='12345',
36 vnfSoftwareVersion="V1",
39 status='NOT_INSTANTIATED').save()
42 "aspectId": "sunshine_aspect"
46 NfInstModel.objects.filter(nfinstid='12345').delete()
48 def assert_job_result(self, job_id, job_progress, job_detail):
49 jobs = JobStatusModel.objects.filter(
51 progress=job_progress,
54 self.assertEqual(1, len(jobs))
56 # def test_scale_vnf_not_found(self):
57 # url = "/api/vnflcm/v1/vnf_instances/12/scale"
58 # response = self.client.post(url,
61 # self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
63 # def test_scale_vnf_conflict(self):
64 # url = "/api/vnflcm/v1/vnf_instances/12345/scale"
65 # response = self.client.post(url,
68 # self.assertEqual(status.HTTP_409_CONFLICT, response.status_code)
70 # def test_scale_vnf_badreq(self):
71 # NfInstModel(nfinstid='678',
77 # vnfSoftwareVersion="V1",
80 # status='INSTANTIATED').save()
81 # url = "/api/vnflcm/v1/vnf_instances/678/scale"
82 # response = self.client.post(url,
85 # NfInstModel.objects.filter(nfinstid='678').delete()
86 # self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
88 @mock.patch.object(restcall, 'call_req')
89 @mock.patch.object(api, 'call')
90 def test_scale_out_vnf_success(self, mock_call, mock_call_req):
91 self.nf_inst_id = '6789'
92 res_cache = {"volume": {}, "flavor": {}, "port": {}}
93 res_cache["volume"]["test"] = "test"
94 # res_cache["flavor"]["vdu1Id"] = "flavor1"
95 # res_cache["port"]["ext_cp"] = "port1"
96 NfInstModel(nfinstid=self.nf_inst_id,
102 vnfSoftwareVersion="V1",
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(
114 instid=self.nf_inst_id,
117 NetworkInstModel.objects.create(
125 instid=self.nf_inst_id
127 SubNetworkInstModel.objects.create(
136 instid=self.nf_inst_id
138 PortInstModel.objects.create(
148 instid=self.nf_inst_id
150 FlavourInstModel.objects.create(
154 instid=self.nf_inst_id,
156 name="Flavor_sunshine"
158 VmInstModel.objects.create(
163 instid=self.nf_inst_id,
168 r1_apply_grant_result = [
170 json.JSONEncoder().encode(const.instantiate_grant_result),
173 mock_call_req.side_effect = [
174 r1_apply_grant_result,
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
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", )
189 {"type": "SCALE_OUT",
190 "aspectId": "sunshine_aspect"},
191 nf_inst_id=self.nf_inst_id,
195 NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
196 # print([{job.progress: job.descp} for job in JobStatusModel.objects.filter( jobid=self.job_id)])
197 self.assert_job_result(
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,
213 # nf_desc="VNF DESC",
217 # vnfSoftwareVersion="V1",
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(
229 # instid=self.nf_inst_id,
232 # NetworkInstModel.objects.create(
236 # name='pnet_network',
240 # instid=self.nf_inst_id
242 # SubNetworkInstModel.objects.create(
251 # instid=self.nf_inst_id
253 # PortInstModel.objects.create(
263 # instid=self.nf_inst_id
265 # FlavourInstModel.objects.create(
269 # instid=self.nf_inst_id,
271 # name="Flavor_sunshine"
273 # VmInstModel.objects.create(
278 # instid=self.nf_inst_id,
283 # VmInstModel.objects.create(
288 # instid=self.nf_inst_id,
293 # VNFCInstModel.objects.create(
294 # vnfcinstanceid="1",
295 # instid=self.nf_inst_id,
298 # VNFCInstModel.objects.create(
299 # vnfcinstanceid="2",
300 # instid=self.nf_inst_id,
303 # r1_apply_grant_result = [
305 # json.JSONEncoder().encode(const.instantiate_grant_result),
308 # mock_call_req.side_effect = [
309 # r1_apply_grant_result,
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
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", )
323 # {"type": "SCALE_IN",
324 # "aspectId": "sunshine_aspect"},
325 # nf_inst_id=self.nf_inst_id,
329 # NfInstModel.objects.filter(nfinstid=self.nf_inst_id).delete()
330 # self.assert_job_result(
333 # 'Scale Vnf success.'