1 # Copyright (C) 2018 Verizon. All Rights Reserved.
\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, Client
\r
20 from rest_framework import status
\r
22 from lcm.nf.biz.operate_vnf import OperateVnf
\r
23 from lcm.pub.database.models import NfInstModel
\r
24 from lcm.pub.database.models import JobStatusModel
\r
25 from lcm.pub.database.models import VmInstModel
\r
26 from lcm.pub.database.models import SubscriptionModel
\r
27 from lcm.pub.utils import restcall
\r
28 from lcm.pub.utils.jobutil import JobUtil
\r
29 from lcm.pub.utils.notificationsutil import NotificationsUtil
\r
30 from lcm.pub.utils.timeutil import now_time
\r
31 from lcm.pub.vimapi import api
\r
34 class TestNFOperate(TestCase):
\r
36 self.client = Client()
\r
39 VmInstModel.objects.all().delete()
\r
41 def assert_job_result(self, job_id, job_progress, job_detail):
\r
42 jobs = JobStatusModel.objects.filter(
\r
44 progress=job_progress,
\r
47 self.assertEqual(1, len(jobs))
\r
49 def test_operate_vnf_not_found(self):
\r
51 "changeStateTo": "STARTED"
\r
53 response = self.client.post(
\r
54 "/api/vnflcm/v1/vnf_instances/12/operate",
\r
58 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
\r
60 def test_operate_vnf_conflict(self):
\r
62 "changeStateTo": "STARTED"
\r
67 status='NOT_INSTANTIATED'
\r
69 response = self.client.post(
\r
70 "/api/vnflcm/v1/vnf_instances/12/operate",
\r
74 self.assertEqual(status.HTTP_409_CONFLICT, response.status_code)
\r
78 status='NOT_INSTANTIATED'
\r
81 def test_operate_vnf_badreq(self):
\r
89 vnfSoftwareVersion="V1",
\r
92 status='INSTANTIATED'
\r
94 url = "/api/vnflcm/v1/vnf_instances/345/operate"
\r
95 response = self.client.post(
\r
100 NfInstModel.objects.filter(nfinstid='345').delete()
\r
102 status.HTTP_400_BAD_REQUEST,
\r
103 response.status_code
\r
106 @mock.patch.object(OperateVnf, 'run')
\r
107 def test_operate_vnf_success(self, mock_run):
\r
109 "changeStateTo": "STARTED"
\r
114 status='INSTANTIATED'
\r
116 response = self.client.post(
\r
117 "/api/vnflcm/v1/vnf_instances/12/operate",
\r
121 mock_run.re.return_value = None
\r
122 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
\r
126 status='INSTANTIATED'
\r
129 @mock.patch.object(restcall, 'call_req')
\r
130 @mock.patch.object(api, 'call')
\r
131 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
132 def test_operate_vnf_success_start(self, mock_post_notification, mock_call, mock_call_req):
\r
133 NfInstModel.objects.create(
\r
142 status='INSTANTIATED',
\r
145 vnfSoftwareVersion='',
\r
146 vnfConfigurableProperties='todo',
\r
147 localizationLanguage='EN_US',
\r
148 create_time=now_time()
\r
151 VmInstModel.objects.create(
\r
162 SubscriptionModel.objects.create(
\r
163 subscription_id=str(uuid.uuid4()),
\r
164 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
165 auth_info=json.JSONEncoder().encode({
\r
166 'authType': ['BASIC'],
\r
168 'userName': 'username',
\r
169 'password': 'password'
\r
172 notification_types=str([
\r
173 'VnfLcmOperationOccurrenceNotification',
\r
174 'VnfIdentifierCreationNotification',
\r
175 'VnfIdentifierDeletionNotification'
\r
177 operation_types=str(['OPERATE']),
\r
178 operation_states=str(['COMPLETED']),
\r
179 vnf_instance_filter=json.JSONEncoder().encode({
\r
181 'vnfProductsFromProviders': [],
\r
182 'vnfInstanceIds': ['1111'],
\r
183 'vnfInstanceNames': [],
\r
187 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
188 t3_action_vm_start_result = [0, json.JSONEncoder().encode(''), '202']
\r
189 mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_start_result]
\r
190 mock_call.return_value = None
\r
191 mock_post_notification.return_value = None
\r
193 "changeStateTo": "STARTED"
\r
195 self.nf_inst_id = '1111'
\r
196 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
197 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
198 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
199 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
201 @mock.patch.object(restcall, 'call_req')
\r
202 @mock.patch.object(api, 'call')
\r
203 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
204 def test_operate_vnf_success_stop(self, mock_post_notification, mock_call, mock_call_req):
\r
205 NfInstModel.objects.create(
\r
214 status='INSTANTIATED',
\r
217 vnfSoftwareVersion='',
\r
218 vnfConfigurableProperties='todo',
\r
219 localizationLanguage='EN_US',
\r
220 create_time=now_time()
\r
223 VmInstModel.objects.create(
\r
234 SubscriptionModel.objects.create(
\r
235 subscription_id=str(uuid.uuid4()),
\r
236 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
237 auth_info=json.JSONEncoder().encode({
\r
238 'authType': ['BASIC'],
\r
240 'userName': 'username',
\r
241 'password': 'password'
\r
244 notification_types=str([
\r
245 'VnfLcmOperationOccurrenceNotification',
\r
246 'VnfIdentifierCreationNotification',
\r
247 'VnfIdentifierDeletionNotification'
\r
249 operation_types=str(['OPERATE']),
\r
250 operation_states=str(['COMPLETED']),
\r
251 vnf_instance_filter=json.JSONEncoder().encode({
\r
253 'vnfProductsFromProviders': [],
\r
254 'vnfInstanceIds': ['1111'],
\r
255 'vnfInstanceNames': [],
\r
259 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
260 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
261 mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_stop_result]
\r
262 mock_call.return_value = None
\r
263 mock_post_notification.return_value = None
\r
265 "changeStateTo": "STOPPED"
\r
267 self.nf_inst_id = '1111'
\r
268 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
269 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
270 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
271 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
273 @mock.patch.object(restcall, 'call_req')
\r
274 @mock.patch.object(api, 'call')
\r
275 def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req):
\r
276 NfInstModel.objects.create(
\r
285 status='INSTANTIATED',
\r
288 vnfSoftwareVersion='',
\r
289 vnfConfigurableProperties='todo',
\r
290 localizationLanguage='EN_US',
\r
291 create_time=now_time()
\r
294 VmInstModel.objects.create(
\r
304 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
305 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
306 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
307 mock_call_req.side_effect = [
\r
308 t1_apply_grant_result,
\r
309 t2_lcm_notify_result,
\r
310 t3_action_vm_stop_result
\r
312 mock_call.return_value = None
\r
314 "changeStateTo": "STOPPED",
\r
315 "stopType": "GRACEFUL",
\r
316 "gracefulStopTimeout": 2
\r
318 self.nf_inst_id = '1111'
\r
319 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
320 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
321 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
322 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
324 @mock.patch.object(restcall, 'call_req')
\r
325 @mock.patch.object(api, 'call')
\r
326 def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req):
\r
327 NfInstModel.objects.create(
\r
336 status='INSTANTIATED',
\r
339 vnfSoftwareVersion='',
\r
340 vnfConfigurableProperties='todo',
\r
341 localizationLanguage='EN_US',
\r
342 create_time=now_time()
\r
345 VmInstModel.objects.create(
\r
355 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
356 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
357 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
358 mock_call_req.side_effect = [
\r
359 t1_apply_grant_result,
\r
360 t2_lcm_notify_result,
\r
361 t3_action_vm_stop_result
\r
363 mock_call.return_value = None
\r
365 "changeStateTo": "STOPPED",
\r
366 "stopType": "FORCEFUL",
\r
368 self.nf_inst_id = '1111'
\r
369 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
370 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
371 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
372 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r