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
\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
50 class TestNFInstantiate(TestCase):
\r
52 self.client = APIClient()
\r
53 self.grant_result = instantiate_grant_result
\r
56 NfInstModel.objects.all().delete()
\r
58 def assert_job_result(self, job_id, job_progress, job_detail):
\r
59 jobs = JobStatusModel.objects.filter(
\r
61 progress=job_progress,
\r
64 self.assertEqual(1, len(jobs))
\r
66 @mock.patch.object(InstantiateVnf, 'run')
\r
67 def test_instantiate_vnf(self, mock_run):
\r
71 status='NOT_INSTANTIATED'
\r
73 mock_run.re.return_value = None
\r
74 response = self.client.post(
\r
75 '/api/vnflcm/v1/vnf_instances/12/instantiate',
\r
79 self.failUnlessEqual(
\r
80 status.HTTP_202_ACCEPTED,
\r
81 response.status_code
\r
84 @mock.patch.object(InstantiateVnf, 'run')
\r
85 def test_instantiate_vnf_notfound(self, mock_run):
\r
86 mock_run.re.return_value = None
\r
87 response = self.client.post(
\r
88 '/api/vnflcm/v1/vnf_instances/3421/instantiate',
\r
92 self.failUnlessEqual(
\r
93 status.HTTP_404_NOT_FOUND,
\r
94 response.status_code,
\r
98 @mock.patch.object(InstantiateVnf, 'run')
\r
99 def test_instantiate_vnf_conflict(self, mock_run):
\r
103 status='INSTANTIATED'
\r
105 mock_run.re.return_value = None
\r
106 response = self.client.post(
\r
107 '/api/vnflcm/v1/vnf_instances/1123/instantiate',
\r
108 data=inst_req_data,
\r
111 self.failUnlessEqual(
\r
112 status.HTTP_409_CONFLICT,
\r
113 response.status_code,
\r
117 def test_instantiate_vnf_when_inst_id_not_exist(self):
\r
118 self.nf_inst_id = str(uuid.uuid4())
\r
119 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
120 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
121 jobs = JobStatusModel.objects.filter(
\r
124 descp='INST_VNF_READY'
\r
126 self.assertEqual(1, len(jobs))
\r
127 data = inst_req_data
\r
130 nf_inst_id=self.nf_inst_id,
\r
133 self.assert_job_result(
\r
136 'VNF nf_inst_id is not exist.'
\r
139 def test_instantiate_vnf_when_already_instantiated(self):
\r
140 NfInstModel.objects.create(
\r
148 status='INSTANTIATED',
\r
149 nf_desc='vFW in Nanjing TIC Edge',
\r
151 create_time=now_time()
\r
153 self.nf_inst_id = '1111'
\r
154 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
155 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
156 jobs = JobStatusModel.objects.filter(
\r
159 descp='INST_VNF_READY'
\r
161 self.assertEqual(1, len(jobs))
\r
162 data = inst_req_data
\r
165 nf_inst_id=self.nf_inst_id,
\r
168 self.assert_job_result(
\r
171 'VNF instantiationState is not NOT_INSTANTIATED.'
\r
174 @mock.patch.object(restcall, 'call_req')
\r
175 def test_instantiate_vnf_when_get_packageinfo_by_csarid_failed(self, mock_call_req):
\r
176 NfInstModel.objects.create(
\r
184 status='NOT_INSTANTIATED',
\r
185 nf_desc='vFW in Nanjing TIC Edge',
\r
187 create_time=now_time()
\r
189 r1_get_vnfpackage_by_vnfdid = [
\r
191 json.JSONEncoder().encode(vnfpackage_info),
\r
194 mock_call_req.side_effect = [r1_get_vnfpackage_by_vnfdid]
\r
195 self.nf_inst_id = '1111'
\r
196 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
197 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
198 data = inst_req_data
\r
201 nf_inst_id=self.nf_inst_id,
\r
204 self.assert_job_result(
\r
207 'Failed to query vnf CSAR(111) from catalog.'
\r
210 @mock.patch.object(restcall, 'call_req')
\r
211 def test_instantiate_vnf_when_applay_grant_failed(self, mock_call_req):
\r
212 NfInstModel.objects.create(
\r
220 status='NOT_INSTANTIATED',
\r
221 nf_desc='vFW in Nanjing TIC Edge',
\r
223 create_time=now_time()
\r
225 r1_get_vnfpackage_by_vnfdid = [
\r
227 json.JSONEncoder().encode(vnfpackage_info),
\r
230 r2_apply_grant_result = [
\r
232 json.JSONEncoder().encode(self.grant_result),
\r
235 mock_call_req.side_effect = [
\r
236 r1_get_vnfpackage_by_vnfdid,
\r
237 r2_apply_grant_result
\r
239 self.nf_inst_id = '1111'
\r
240 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
241 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
242 data = inst_req_data
\r
245 nf_inst_id=self.nf_inst_id,
\r
248 self.assert_job_result(
\r
251 'Nf instancing apply grant exception'
\r
254 @mock.patch.object(restcall, 'call_req')
\r
255 @mock.patch.object(api, 'call')
\r
256 def test_instantiate_vnf_when_unexpected_exception(self, mock_call, mock_call_req):
\r
257 NfInstModel.objects.create(
\r
265 status='NOT_INSTANTIATED',
\r
266 nf_desc='vFW in Nanjing TIC Edge',
\r
268 create_time=now_time()
\r
270 r1_get_vnfpackage_by_vnfdid = [
\r
272 json.JSONEncoder().encode(vnfpackage_info),
\r
275 r2_apply_grant_result = [
\r
277 json.JSONEncoder().encode(self.grant_result),
\r
280 mock_call_req.side_effect = [
\r
281 r1_get_vnfpackage_by_vnfdid,
\r
282 r2_apply_grant_result
\r
284 mock_call.side_effect = [
\r
285 c1_data_get_tenant_id,
\r
286 c2_data_create_volume,
\r
289 self.nf_inst_id = '1111'
\r
290 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
291 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
292 data = inst_req_data
\r
295 nf_inst_id=self.nf_inst_id,
\r
298 self.assert_job_result(
\r
301 'unexpected exception'
\r
304 @mock.patch.object(restcall, 'call_req')
\r
305 @mock.patch.object(api, 'call')
\r
306 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
307 def test_instantiate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):
\r
308 NfInstModel.objects.create(
\r
316 status='NOT_INSTANTIATED',
\r
317 nf_desc='vFW in Nanjing TIC Edge',
\r
319 create_time=now_time()
\r
321 SubscriptionModel.objects.create(
\r
322 subscription_id=str(uuid.uuid4()),
\r
323 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
324 auth_info=json.JSONEncoder().encode({
\r
325 'authType': ['BASIC'],
\r
327 'userName': 'username',
\r
328 'password': 'password'
\r
331 notification_types=str([
\r
332 'VnfLcmOperationOccurrenceNotification',
\r
333 'VnfIdentifierCreationNotification',
\r
334 'VnfIdentifierDeletionNotification'
\r
336 operation_types=str(['INSTANTIATE']),
\r
337 operation_states=str(['COMPLETED']),
\r
338 vnf_instance_filter=json.JSONEncoder().encode({
\r
339 'vnfdIds': ['111'],
\r
340 'vnfProductsFromProviders': [],
\r
341 'vnfInstanceIds': ['1111'],
\r
342 'vnfInstanceNames': [],
\r
345 r1_get_vnfpackage_by_vnfdid = [
\r
347 json.JSONEncoder().encode(vnfpackage_info),
\r
350 r2_apply_grant_result = [
\r
352 json.JSONEncoder().encode(self.grant_result),
\r
355 r3_all_aai_result = [
\r
357 json.JSONEncoder().encode(''),
\r
360 r4_lcm_notify_result = [
\r
362 json.JSONEncoder().encode(''),
\r
365 mock_call_req.side_effect = [
\r
366 r1_get_vnfpackage_by_vnfdid,
\r
367 r2_apply_grant_result,
\r
369 r4_lcm_notify_result
\r
371 mock_call.side_effect = [
\r
372 c1_data_get_tenant_id,
\r
373 c2_data_create_volume,
\r
374 c3_data_get_volume,
\r
375 c4_data_create_network,
\r
376 c5_data_create_subnet,
\r
377 c6_data_create_port,
\r
378 c7_data_create_flavor,
\r
379 c8_data_list_image,
\r
383 mock_post_notification.side_effect = None
\r
384 self.nf_inst_id = '1111'
\r
385 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
386 JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')
\r
387 data = inst_req_data
\r
390 nf_inst_id=self.nf_inst_id,
\r
394 @mock.patch.object(JobUtil, 'create_job')
\r
395 def test_instantiate_inner_error(self, mock_run):
\r
399 status='NOT_INSTANTIATED'
\r
401 mock_run.return_value = NFLCMException('Boom!')
\r
402 response = self.client.post(
\r
403 '/api/vnflcm/v1/vnf_instances/144/instantiate',
\r
404 data=inst_req_data,
\r
407 NfInstModel.objects.filter(nfinstid='144').delete()
\r
408 self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)
\r
410 def test_instantiate_badreq(self):
\r
414 status='NOT_INSTANTIATED'
\r
416 response = self.client.post(
\r
417 '/api/vnflcm/v1/vnf_instances/144/instantiate',
\r
421 NfInstModel.objects.filter(nfinstid='144').delete()
\r
422 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
\r