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, JobStatusModel, VmInstModel, SubscriptionModel
\r
24 from lcm.pub.utils import restcall
\r
25 from lcm.pub.utils.jobutil import JobUtil
\r
26 from lcm.pub.utils.notificationsutil import NotificationsUtil
\r
27 from lcm.pub.utils.timeutil import now_time
\r
28 from lcm.pub.vimapi import api
\r
31 class TestNFOperate(TestCase):
\r
33 self.client = Client()
\r
36 VmInstModel.objects.all().delete()
\r
38 def assert_job_result(self, job_id, job_progress, job_detail):
\r
39 jobs = JobStatusModel.objects.filter(jobid=job_id,
\r
40 progress=job_progress,
\r
42 self.assertEqual(1, len(jobs))
\r
44 def test_operate_vnf_not_found(self):
\r
46 "changeStateTo": "STARTED"
\r
48 response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json')
\r
49 self.failUnlessEqual(status.HTTP_404_NOT_FOUND, response.status_code)
\r
51 def test_operate_vnf_conflict(self):
\r
53 "changeStateTo": "STARTED"
\r
55 NfInstModel(nfinstid='12', nf_name='VNF1', status='NOT_INSTANTIATED').save()
\r
56 response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json')
\r
57 self.failUnlessEqual(status.HTTP_409_CONFLICT, response.status_code)
\r
58 NfInstModel(nfinstid='12', nf_name='VNF1', status='NOT_INSTANTIATED').delete()
\r
60 def test_operate_vnf_inner_error(self):
\r
61 NfInstModel(nfinstid='345',
\r
67 vnfSoftwareVersion="V1",
\r
70 status='INSTANTIATED').save()
\r
71 url = "/api/vnflcm/v1/vnf_instances/345/operate"
\r
72 response = self.client.post(url,
\r
75 NfInstModel.objects.filter(nfinstid='345').delete()
\r
76 self.failUnlessEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)
\r
78 @mock.patch.object(OperateVnf, 'run')
\r
79 def test_operate_vnf_success(self, mock_run):
\r
81 "changeStateTo": "STARTED"
\r
83 NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').save()
\r
84 response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json')
\r
85 mock_run.re.return_value = None
\r
86 self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
\r
87 NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').delete()
\r
89 @mock.patch.object(restcall, 'call_req')
\r
90 @mock.patch.object(api, 'call')
\r
91 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
92 def test_operate_vnf_success_start(self, mock_post_notification, mock_call, mock_call_req):
\r
93 NfInstModel.objects.create(nfinstid='1111',
\r
101 status='INSTANTIATED',
\r
104 vnfSoftwareVersion='',
\r
105 vnfConfigurableProperties='todo',
\r
106 localizationLanguage='EN_US',
\r
107 create_time=now_time())
\r
109 VmInstModel.objects.create(vmid="1",
\r
116 operationalstate=1)
\r
118 SubscriptionModel.objects.create(
\r
119 subscription_id=str(uuid.uuid4()),
\r
120 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
121 auth_info=json.JSONEncoder().encode({
\r
122 'authType': ['BASIC'],
\r
124 'userName': 'username',
\r
125 'password': 'password'
\r
128 notification_types=str([
\r
129 'VnfLcmOperationOccurrenceNotification',
\r
130 'VnfIdentifierCreationNotification',
\r
131 'VnfIdentifierDeletionNotification'
\r
133 operation_types=str(['OPERATE']),
\r
134 operation_states=str(['COMPLETED']),
\r
135 vnf_instance_filter=json.JSONEncoder().encode({
\r
137 'vnfProductsFromProviders': [],
\r
138 'vnfInstanceIds': ['1111'],
\r
139 'vnfInstanceNames': [],
\r
143 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
144 # t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
145 t3_action_vm_start_result = [0, json.JSONEncoder().encode(''), '202']
\r
146 # mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_start_result]
\r
147 mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_start_result]
\r
148 mock_call.return_value = None
\r
149 mock_post_notification.return_value = None
\r
151 "changeStateTo": "STARTED"
\r
153 self.nf_inst_id = '1111'
\r
154 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
155 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
156 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
157 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
159 @mock.patch.object(restcall, 'call_req')
\r
160 @mock.patch.object(api, 'call')
\r
161 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
162 def test_operate_vnf_success_stop(self, mock_post_notification, mock_call, mock_call_req):
\r
163 NfInstModel.objects.create(nfinstid='1111',
\r
171 status='INSTANTIATED',
\r
174 vnfSoftwareVersion='',
\r
175 vnfConfigurableProperties='todo',
\r
176 localizationLanguage='EN_US',
\r
177 create_time=now_time())
\r
179 VmInstModel.objects.create(vmid="1",
\r
186 operationalstate=1)
\r
188 SubscriptionModel.objects.create(
\r
189 subscription_id=str(uuid.uuid4()),
\r
190 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
191 auth_info=json.JSONEncoder().encode({
\r
192 'authType': ['BASIC'],
\r
194 'userName': 'username',
\r
195 'password': 'password'
\r
198 notification_types=str([
\r
199 'VnfLcmOperationOccurrenceNotification',
\r
200 'VnfIdentifierCreationNotification',
\r
201 'VnfIdentifierDeletionNotification'
\r
203 operation_types=str(['OPERATE']),
\r
204 operation_states=str(['COMPLETED']),
\r
205 vnf_instance_filter=json.JSONEncoder().encode({
\r
207 'vnfProductsFromProviders': [],
\r
208 'vnfInstanceIds': ['1111'],
\r
209 'vnfInstanceNames': [],
\r
213 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
214 # t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
215 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
216 # mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
217 mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_stop_result]
\r
218 mock_call.return_value = None
\r
219 mock_post_notification.return_value = None
\r
221 "changeStateTo": "STOPPED"
\r
223 self.nf_inst_id = '1111'
\r
224 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
225 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
226 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
227 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
229 @mock.patch.object(restcall, 'call_req')
\r
230 @mock.patch.object(api, 'call')
\r
231 def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req):
\r
232 NfInstModel.objects.create(nfinstid='1111',
\r
240 status='INSTANTIATED',
\r
243 vnfSoftwareVersion='',
\r
244 vnfConfigurableProperties='todo',
\r
245 localizationLanguage='EN_US',
\r
246 create_time=now_time())
\r
248 VmInstModel.objects.create(vmid="1",
\r
255 operationalstate=1)
\r
256 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
257 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
258 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
259 mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
260 mock_call.return_value = None
\r
262 "changeStateTo": "STOPPED",
\r
263 "stopType": "GRACEFUL",
\r
264 "gracefulStopTimeout": 2
\r
266 self.nf_inst_id = '1111'
\r
267 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
268 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
269 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
270 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
272 @mock.patch.object(restcall, 'call_req')
\r
273 @mock.patch.object(api, 'call')
\r
274 def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req):
\r
275 NfInstModel.objects.create(nfinstid='1111',
\r
283 status='INSTANTIATED',
\r
286 vnfSoftwareVersion='',
\r
287 vnfConfigurableProperties='todo',
\r
288 localizationLanguage='EN_US',
\r
289 create_time=now_time())
\r
291 VmInstModel.objects.create(vmid="1",
\r
298 operationalstate=1)
\r
299 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
300 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
301 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
302 mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
303 mock_call.return_value = None
\r
305 "changeStateTo": "STOPPED",
\r
306 "stopType": "FORCEFUL",
\r
308 self.nf_inst_id = '1111'
\r
309 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
310 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
311 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
312 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r