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 @mock.patch.object(OperateVnf, 'run')
\r
61 def test_operate_vnf_success(self, mock_run):
\r
63 "changeStateTo": "STARTED"
\r
65 NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').save()
\r
66 response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json')
\r
67 mock_run.re.return_value = None
\r
68 self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
\r
69 NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').delete()
\r
71 @mock.patch.object(restcall, 'call_req')
\r
72 @mock.patch.object(api, 'call')
\r
73 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
74 def test_operate_vnf_success_start(self, mock_post_notification, mock_call, mock_call_req):
\r
75 NfInstModel.objects.create(nfinstid='1111',
\r
83 status='INSTANTIATED',
\r
86 vnfSoftwareVersion='',
\r
87 vnfConfigurableProperties='todo',
\r
88 localizationLanguage='EN_US',
\r
89 create_time=now_time())
\r
91 VmInstModel.objects.create(vmid="1",
\r
100 SubscriptionModel.objects.create(
\r
101 subscription_id=str(uuid.uuid4()),
\r
102 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
103 auth_info=json.JSONEncoder().encode({
\r
104 'authType': ['BASIC'],
\r
106 'userName': 'username',
\r
107 'password': 'password'
\r
110 notification_types=str([
\r
111 'VnfLcmOperationOccurrenceNotification',
\r
112 'VnfIdentifierCreationNotification',
\r
113 'VnfIdentifierDeletionNotification'
\r
115 operation_types=str(['OPERATE']),
\r
116 operation_states=str(['COMPLETED']),
\r
117 vnf_instance_filter=json.JSONEncoder().encode({
\r
119 'vnfProductsFromProviders': [],
\r
120 'vnfInstanceIds': ['1111'],
\r
121 'vnfInstanceNames': [],
\r
125 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
126 # t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
127 t3_action_vm_start_result = [0, json.JSONEncoder().encode(''), '202']
\r
128 # mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_start_result]
\r
129 mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_start_result]
\r
130 mock_call.return_value = None
\r
131 mock_post_notification.return_value = None
\r
133 "changeStateTo": "STARTED"
\r
135 self.nf_inst_id = '1111'
\r
136 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
137 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
138 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
139 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
141 @mock.patch.object(restcall, 'call_req')
\r
142 @mock.patch.object(api, 'call')
\r
143 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
144 def test_operate_vnf_success_stop(self, mock_post_notification, mock_call, mock_call_req):
\r
145 NfInstModel.objects.create(nfinstid='1111',
\r
153 status='INSTANTIATED',
\r
156 vnfSoftwareVersion='',
\r
157 vnfConfigurableProperties='todo',
\r
158 localizationLanguage='EN_US',
\r
159 create_time=now_time())
\r
161 VmInstModel.objects.create(vmid="1",
\r
168 operationalstate=1)
\r
170 SubscriptionModel.objects.create(
\r
171 subscription_id=str(uuid.uuid4()),
\r
172 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
173 auth_info=json.JSONEncoder().encode({
\r
174 'authType': ['BASIC'],
\r
176 'userName': 'username',
\r
177 'password': 'password'
\r
180 notification_types=str([
\r
181 'VnfLcmOperationOccurrenceNotification',
\r
182 'VnfIdentifierCreationNotification',
\r
183 'VnfIdentifierDeletionNotification'
\r
185 operation_types=str(['OPERATE']),
\r
186 operation_states=str(['COMPLETED']),
\r
187 vnf_instance_filter=json.JSONEncoder().encode({
\r
189 'vnfProductsFromProviders': [],
\r
190 'vnfInstanceIds': ['1111'],
\r
191 'vnfInstanceNames': [],
\r
195 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
196 # t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
197 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
198 # mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
199 mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_stop_result]
\r
200 mock_call.return_value = None
\r
201 mock_post_notification.return_value = None
\r
203 "changeStateTo": "STOPPED"
\r
205 self.nf_inst_id = '1111'
\r
206 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
207 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
208 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
209 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
211 @mock.patch.object(restcall, 'call_req')
\r
212 @mock.patch.object(api, 'call')
\r
213 def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req):
\r
214 NfInstModel.objects.create(nfinstid='1111',
\r
222 status='INSTANTIATED',
\r
225 vnfSoftwareVersion='',
\r
226 vnfConfigurableProperties='todo',
\r
227 localizationLanguage='EN_US',
\r
228 create_time=now_time())
\r
230 VmInstModel.objects.create(vmid="1",
\r
237 operationalstate=1)
\r
238 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
239 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
240 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
241 mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
242 mock_call.return_value = None
\r
244 "changeStateTo": "STOPPED",
\r
245 "stopType": "GRACEFUL",
\r
246 "gracefulStopTimeout": 2
\r
248 self.nf_inst_id = '1111'
\r
249 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
250 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
251 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
252 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
254 @mock.patch.object(restcall, 'call_req')
\r
255 @mock.patch.object(api, 'call')
\r
256 def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req):
\r
257 NfInstModel.objects.create(nfinstid='1111',
\r
265 status='INSTANTIATED',
\r
268 vnfSoftwareVersion='',
\r
269 vnfConfigurableProperties='todo',
\r
270 localizationLanguage='EN_US',
\r
271 create_time=now_time())
\r
273 VmInstModel.objects.create(vmid="1",
\r
280 operationalstate=1)
\r
281 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
282 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
283 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
284 mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
285 mock_call.return_value = None
\r
287 "changeStateTo": "STOPPED",
\r
288 "stopType": "FORCEFUL",
\r
290 self.nf_inst_id = '1111'
\r
291 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
292 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
293 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
294 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r