1 # Copyright 2017 ZTE Corporation.
\r
3 # Licensed under the Apache License, Version 2.0 (the "License");
\r
4 # you may not use this file except in compliance with the License.
\r
5 # You may obtain a copy of the License at
\r
7 # http://www.apache.org/licenses/LICENSE-2.0
\r
9 # Unless required by applicable law or agreed to in writing, software
\r
10 # distributed under the License is distributed on an "AS IS" BASIS,
\r
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
\r
12 # See the License for the specific language governing permissions and
\r
13 # limitations under the License.
\r
19 from django.test import TestCase
\r
20 from rest_framework import status
\r
21 from rest_framework.test import APIClient
\r
23 from .const import c1_data_get_tenant_id
\r
24 from .const import c4_data_create_network
\r
25 from .const import c2_data_create_volume
\r
26 from .const import c5_data_create_subnet
\r
27 from .const import c3_data_get_volume
\r
28 from .const import c6_data_create_port
\r
29 from .const import c7_data_create_flavor
\r
30 from .const import c8_data_list_image
\r
31 from .const import c9_data_create_vm
\r
32 from .const import c10_data_get_vm
\r
33 from .const import inst_req_data
\r
34 from .const import vnfpackage_info
\r
35 from .const import instantiate_grant_result
\r
37 from lcm.pub.database.models import NfInstModel, VNFLcmOpOccModel
\r
38 from lcm.pub.database.models import JobStatusModel
\r
39 from lcm.pub.database.models import SubscriptionModel
\r
40 from lcm.pub.utils import restcall
\r
41 from lcm.pub.utils.jobutil import JobUtil
\r
42 from lcm.pub.utils.timeutil import now_time
\r
43 from lcm.pub.utils.notificationsutil import NotificationsUtil
\r
44 from lcm.pub.vimapi import api
\r
45 from lcm.pub.exceptions import NFLCMException
\r
47 from lcm.nf.biz.instantiate_vnf import InstantiateVnf
\r
48 from lcm.nf import const
\r
51 class TestNFInstantiate(TestCase):
\r
53 self.client = APIClient()
\r
54 self.grant_result = instantiate_grant_result
\r
57 NfInstModel.objects.all().delete()
\r
59 def assert_job_result(self, job_id, job_progress, job_detail):
\r
60 jobs = JobStatusModel.objects.filter(
\r
62 progress=job_progress,
\r
65 self.assertEqual(1, len(jobs))
\r
67 @mock.patch.object(InstantiateVnf, 'run')
\r
68 def test_instantiate_vnf(self, mock_run):
\r
72 status='NOT_INSTANTIATED'
\r
74 mock_run.re.return_value = None
\r
75 response = self.client.post(
\r
76 '/api/vnflcm/v1/vnf_instances/12/instantiate',
\r
81 status.HTTP_202_ACCEPTED,
\r
82 response.status_code
\r
85 @mock.patch.object(InstantiateVnf, 'run')
\r
86 def test_instantiate_vnf_notfound(self, mock_run):
\r
87 mock_run.re.return_value = None
\r
88 response = self.client.post(
\r
89 '/api/vnflcm/v1/vnf_instances/3421/instantiate',
\r
94 status.HTTP_404_NOT_FOUND,
\r
95 response.status_code,
\r
99 @mock.patch.object(InstantiateVnf, 'run')
\r
100 def test_instantiate_vnf_conflict(self, mock_run):
\r
104 status='INSTANTIATED'
\r
106 mock_run.re.return_value = None
\r
107 response = self.client.post(
\r
108 '/api/vnflcm/v1/vnf_instances/1123/instantiate',
\r
109 data=inst_req_data,
\r
113 status.HTTP_409_CONFLICT,
\r
114 response.status_code,
\r
118 def test_instantiate_vnf_when_inst_id_not_exist(self):
\r
119 self.nf_inst_id = str(uuid.uuid4())
\r
120 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
121 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
122 jobs = JobStatusModel.objects.filter(
\r
125 descp='INST_VNF_READY'
\r
127 self.assertEqual(1, len(jobs))
\r
128 data = inst_req_data
\r
131 nf_inst_id=self.nf_inst_id,
\r
134 self.assert_job_result(
\r
137 'VNF nf_inst_id is not exist.'
\r
140 def test_instantiate_vnf_when_already_instantiated(self):
\r
141 NfInstModel.objects.create(
\r
149 status='INSTANTIATED',
\r
150 nf_desc='vFW in Nanjing TIC Edge',
\r
152 create_time=now_time()
\r
154 self.nf_inst_id = '1111'
\r
155 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
156 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
157 jobs = JobStatusModel.objects.filter(
\r
160 descp='INST_VNF_READY'
\r
162 self.assertEqual(1, len(jobs))
\r
163 data = inst_req_data
\r
166 nf_inst_id=self.nf_inst_id,
\r
169 self.assert_job_result(
\r
172 'VNF instantiationState is not NOT_INSTANTIATED.'
\r
175 @mock.patch.object(restcall, 'call_req')
\r
176 def test_instantiate_vnf_when_get_packageinfo_by_csarid_failed(self, mock_call_req):
\r
177 NfInstModel.objects.create(
\r
185 status='NOT_INSTANTIATED',
\r
186 nf_desc='vFW in Nanjing TIC Edge',
\r
188 create_time=now_time()
\r
190 r1_get_vnfpackage_by_vnfdid = [
\r
192 json.JSONEncoder().encode(vnfpackage_info),
\r
195 mock_call_req.side_effect = [r1_get_vnfpackage_by_vnfdid]
\r
196 self.nf_inst_id = '1111'
\r
197 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
198 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
199 data = inst_req_data
\r
202 nf_inst_id=self.nf_inst_id,
\r
205 self.assert_job_result(
\r
208 'Failed to query vnf CSAR(111) from catalog.'
\r
211 @mock.patch.object(restcall, 'call_req')
\r
212 def test_instantiate_vnf_when_applay_grant_failed(self, mock_call_req):
\r
213 NfInstModel.objects.create(
\r
221 status='NOT_INSTANTIATED',
\r
222 nf_desc='vFW in Nanjing TIC Edge',
\r
224 create_time=now_time()
\r
226 r1_get_vnfpackage_by_vnfdid = [
\r
228 json.JSONEncoder().encode(vnfpackage_info),
\r
231 r2_apply_grant_result = [
\r
233 json.JSONEncoder().encode(self.grant_result),
\r
236 mock_call_req.side_effect = [
\r
237 r1_get_vnfpackage_by_vnfdid,
\r
238 r2_apply_grant_result
\r
240 self.nf_inst_id = '1111'
\r
241 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
242 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
243 data = inst_req_data
\r
246 nf_inst_id=self.nf_inst_id,
\r
249 self.assert_job_result(
\r
252 'Nf instancing apply grant exception'
\r
255 @mock.patch.object(restcall, 'call_req')
\r
256 @mock.patch.object(api, 'call')
\r
257 def test_instantiate_vnf_when_unexpected_exception(self, mock_call, mock_call_req):
\r
258 NfInstModel.objects.create(
\r
266 status='NOT_INSTANTIATED',
\r
267 nf_desc='vFW in Nanjing TIC Edge',
\r
269 create_time=now_time()
\r
271 r1_get_vnfpackage_by_vnfdid = [
\r
273 json.JSONEncoder().encode(vnfpackage_info),
\r
276 r2_apply_grant_result = [
\r
278 json.JSONEncoder().encode(self.grant_result),
\r
281 mock_call_req.side_effect = [
\r
282 r1_get_vnfpackage_by_vnfdid,
\r
283 r2_apply_grant_result
\r
285 mock_call.side_effect = [
\r
286 c1_data_get_tenant_id,
\r
287 c2_data_create_volume,
\r
290 self.nf_inst_id = '1111'
\r
291 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
292 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
293 data = inst_req_data
\r
296 nf_inst_id=self.nf_inst_id,
\r
299 self.assert_job_result(
\r
302 'unexpected exception'
\r
305 @mock.patch.object(restcall, 'call_req')
\r
306 @mock.patch.object(api, 'call')
\r
307 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
308 def test_instantiate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):
\r
309 NfInstModel.objects.create(
\r
317 status='NOT_INSTANTIATED',
\r
318 nf_desc='vFW in Nanjing TIC Edge',
\r
320 create_time=now_time()
\r
322 SubscriptionModel.objects.create(
\r
323 subscription_id=str(uuid.uuid4()),
\r
324 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
325 auth_info=json.JSONEncoder().encode({
\r
326 'authType': ['BASIC'],
\r
328 'userName': 'username',
\r
329 'password': 'password'
\r
332 notification_types=str([
\r
333 'VnfLcmOperationOccurrenceNotification',
\r
334 'VnfIdentifierCreationNotification',
\r
335 'VnfIdentifierDeletionNotification'
\r
337 operation_types=str(['INSTANTIATE']),
\r
338 operation_states=str(['COMPLETED']),
\r
339 vnf_instance_filter=json.JSONEncoder().encode({
\r
340 'vnfdIds': ['111'],
\r
341 'vnfProductsFromProviders': [],
\r
342 'vnfInstanceIds': ['1111'],
\r
343 'vnfInstanceNames': [],
\r
346 r1_get_vnfpackage_by_vnfdid = [
\r
348 json.JSONEncoder().encode(vnfpackage_info),
\r
351 r2_apply_grant_result = [
\r
353 json.JSONEncoder().encode(self.grant_result),
\r
356 r3_all_aai_result = [
\r
358 json.JSONEncoder().encode(''),
\r
361 r4_lcm_notify_result = [
\r
363 json.JSONEncoder().encode(''),
\r
366 mock_call_req.side_effect = [
\r
367 r1_get_vnfpackage_by_vnfdid,
\r
368 r2_apply_grant_result,
\r
370 r4_lcm_notify_result
\r
372 mock_call.side_effect = [
\r
373 c1_data_get_tenant_id,
\r
374 c2_data_create_volume,
\r
375 c3_data_get_volume,
\r
376 c4_data_create_network,
\r
377 c5_data_create_subnet,
\r
378 c6_data_create_port,
\r
379 c7_data_create_flavor,
\r
380 c8_data_list_image,
\r
384 mock_post_notification.side_effect = None
\r
385 self.nf_inst_id = '1111'
\r
386 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
387 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
388 data = inst_req_data
\r
391 nf_inst_id=self.nf_inst_id,
\r
394 self.assert_job_result(
\r
397 'Instantiate Vnf success.'
\r
400 @mock.patch.object(JobUtil, 'create_job')
\r
401 def test_instantiate_inner_error(self, mock_run):
\r
405 status='NOT_INSTANTIATED'
\r
407 mock_run.return_value = NFLCMException('Boom!')
\r
408 response = self.client.post(
\r
409 '/api/vnflcm/v1/vnf_instances/144/instantiate',
\r
410 data=inst_req_data,
\r
413 NfInstModel.objects.filter(nfinstid='144').delete()
\r
414 self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)
\r
416 def test_instantiate_badreq(self):
\r
420 status='NOT_INSTANTIATED'
\r
422 response = self.client.post(
\r
423 '/api/vnflcm/v1/vnf_instances/144/instantiate',
\r
427 NfInstModel.objects.filter(nfinstid='144').delete()
\r
428 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
\r
430 @mock.patch.object(restcall, 'call_req')
\r
431 def test_instantiate_operating_conflict(self, mock_call_req):
\r
432 NfInstModel.objects.create(
\r
440 status='NOT_INSTANTIATED',
\r
441 nf_desc='vFW in Nanjing TIC Edge',
\r
443 create_time=now_time()
\r
445 r1_get_vnfpackage_by_vnfdid = [
\r
447 json.JSONEncoder().encode(vnfpackage_info),
\r
450 mock_call_req.side_effect = [
\r
451 r1_get_vnfpackage_by_vnfdid
\r
453 self.nf_inst_id = '1111'
\r
454 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
455 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
456 VNFLcmOpOccModel.objects.create(vnf_instance_id=self.nf_inst_id,
\r
458 operation=const.OPERATION_TYPE.INSTANTIATE,
\r
459 operation_state=const.OPERATION_STATE_TYPE.PROCESSING)
\r
460 response = self.client.post(
\r
461 '/api/vnflcm/v1/vnf_instances/1111/instantiate',
\r
462 data=inst_req_data,
\r
466 status.HTTP_409_CONFLICT,
\r
467 response.status_code
\r