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 vm = VmInstModel.objects.filter(vmid="1", vimid="1", resourceid="11")
\r
140 self.assertEqual("ACTIVE", vm[0].operationalstate)
\r
141 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
143 @mock.patch.object(restcall, 'call_req')
\r
144 @mock.patch.object(api, 'call')
\r
145 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
146 def test_operate_vnf_success_stop(self, mock_post_notification, mock_call, mock_call_req):
\r
147 NfInstModel.objects.create(nfinstid='1111',
\r
155 status='INSTANTIATED',
\r
158 vnfSoftwareVersion='',
\r
159 vnfConfigurableProperties='todo',
\r
160 localizationLanguage='EN_US',
\r
161 create_time=now_time())
\r
163 VmInstModel.objects.create(vmid="1",
\r
170 operationalstate=1)
\r
172 SubscriptionModel.objects.create(
\r
173 subscription_id=str(uuid.uuid4()),
\r
174 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
175 auth_info=json.JSONEncoder().encode({
\r
176 'authType': ['BASIC'],
\r
178 'userName': 'username',
\r
179 'password': 'password'
\r
182 notification_types=str([
\r
183 'VnfLcmOperationOccurrenceNotification',
\r
184 'VnfIdentifierCreationNotification',
\r
185 'VnfIdentifierDeletionNotification'
\r
187 operation_types=str(['OPERATE']),
\r
188 operation_states=str(['COMPLETED']),
\r
189 vnf_instance_filter=json.JSONEncoder().encode({
\r
191 'vnfProductsFromProviders': [],
\r
192 'vnfInstanceIds': ['1111'],
\r
193 'vnfInstanceNames': [],
\r
197 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
198 # t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
199 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
200 # mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
201 mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_stop_result]
\r
202 mock_call.return_value = None
\r
203 mock_post_notification.return_value = None
\r
205 "changeStateTo": "STOPPED"
\r
207 self.nf_inst_id = '1111'
\r
208 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
209 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
210 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
211 vm = VmInstModel.objects.filter(vmid="1", vimid="1", resourceid="11")
\r
212 self.assertEqual("INACTIVE", vm[0].operationalstate)
\r
213 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
215 @mock.patch.object(restcall, 'call_req')
\r
216 @mock.patch.object(api, 'call')
\r
217 def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req):
\r
218 NfInstModel.objects.create(nfinstid='1111',
\r
226 status='INSTANTIATED',
\r
229 vnfSoftwareVersion='',
\r
230 vnfConfigurableProperties='todo',
\r
231 localizationLanguage='EN_US',
\r
232 create_time=now_time())
\r
234 VmInstModel.objects.create(vmid="1",
\r
241 operationalstate=1)
\r
242 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
243 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
244 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
245 mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
246 mock_call.return_value = None
\r
248 "changeStateTo": "STOPPED",
\r
249 "stopType": "GRACEFUL",
\r
250 "gracefulStopTimeout": 2
\r
252 self.nf_inst_id = '1111'
\r
253 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
254 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
255 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
256 vm = VmInstModel.objects.filter(vmid="1", vimid="1", resourceid="11")
\r
257 self.assertEqual("INACTIVE", vm[0].operationalstate)
\r
258 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r
260 @mock.patch.object(restcall, 'call_req')
\r
261 @mock.patch.object(api, 'call')
\r
262 def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req):
\r
263 NfInstModel.objects.create(nfinstid='1111',
\r
271 status='INSTANTIATED',
\r
274 vnfSoftwareVersion='',
\r
275 vnfConfigurableProperties='todo',
\r
276 localizationLanguage='EN_US',
\r
277 create_time=now_time())
\r
279 VmInstModel.objects.create(vmid="1",
\r
286 operationalstate=1)
\r
287 t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']
\r
288 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
289 t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']
\r
290 mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]
\r
291 mock_call.return_value = None
\r
293 "changeStateTo": "STOPPED",
\r
294 "stopType": "FORCEFUL",
\r
296 self.nf_inst_id = '1111'
\r
297 self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)
\r
298 JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")
\r
299 OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
300 vm = VmInstModel.objects.filter(vmid="1", vimid="1", resourceid="11")
\r
301 self.assertEqual("INACTIVE", vm[0].operationalstate)
\r
302 self.assert_job_result(self.job_id, 100, "Operate Vnf success.")
\r