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, Client
\r
20 from rest_framework import status
\r
22 from lcm.pub.database.models import NfInstModel
\r
23 from lcm.pub.database.models import JobStatusModel
\r
24 from lcm.pub.database.models import VmInstModel
\r
25 from lcm.pub.database.models import NetworkInstModel
\r
26 from lcm.pub.database.models import SubNetworkInstModel
\r
27 from lcm.pub.database.models import PortInstModel
\r
28 from lcm.pub.database.models import FlavourInstModel
\r
29 from lcm.pub.database.models import StorageInstModel
\r
30 from lcm.pub.database.models import SubscriptionModel
\r
31 from lcm.pub.utils import restcall
\r
32 from lcm.pub.utils.jobutil import JobUtil
\r
33 from lcm.pub.utils.timeutil import now_time
\r
34 from lcm.pub.utils.notificationsutil import NotificationsUtil
\r
35 from lcm.pub.vimapi import api
\r
36 from lcm.nf.biz.terminate_vnf import TerminateVnf
\r
39 class TestNFTerminate(TestCase):
\r
41 self.client = Client()
\r
42 StorageInstModel.objects.create(
\r
50 NetworkInstModel.objects.create(
\r
54 name='pnet_network',
\r
60 SubNetworkInstModel.objects.create(
\r
71 PortInstModel.objects.create(
\r
83 FlavourInstModel.objects.create(
\r
90 VmInstModel.objects.create(
\r
101 def tearDown(self):
\r
102 VmInstModel.objects.all().delete()
\r
103 NetworkInstModel.objects.all().delete()
\r
104 SubNetworkInstModel.objects.all().delete()
\r
105 PortInstModel.objects.all().delete()
\r
106 NfInstModel.objects.all().delete()
\r
108 def assert_job_result(self, job_id, job_progress, job_detail):
\r
109 jobs = JobStatusModel.objects.filter(
\r
111 progress=job_progress,
\r
114 self.assertEqual(1, len(jobs))
\r
116 @mock.patch.object(TerminateVnf, 'run')
\r
117 def test_terminate_vnf(self, mock_run):
\r
119 "terminationType": "GRACEFUL",
\r
120 "gracefulTerminationTimeout": 120
\r
125 nf_desc="VNF DESC",
\r
129 vnfSoftwareVersion="V1",
\r
132 status='INSTANTIATED'
\r
134 mock_run.re.return_value = None
\r
135 response = self.client.post(
\r
136 "/api/vnflcm/v1/vnf_instances/12/terminate",
\r
141 status.HTTP_202_ACCEPTED,
\r
142 response.status_code,
\r
146 @mock.patch.object(TerminateVnf, 'run')
\r
147 def test_terminate_vnf_not_found(self, mock_run):
\r
149 "terminationType": "GRACEFUL",
\r
150 "gracefulTerminationTimeout": 120
\r
152 mock_run.re.return_value = None
\r
153 response = self.client.post(
\r
154 "/api/vnflcm/v1/vnf_instances/567/terminate",
\r
159 status.HTTP_404_NOT_FOUND,
\r
160 response.status_code,
\r
164 @mock.patch.object(TerminateVnf, 'run')
\r
165 def test_terminate_vnf_conflict(self, mock_run):
\r
167 "terminationType": "GRACEFUL",
\r
168 "gracefulTerminationTimeout": 120
\r
173 nf_desc="VNF DESC",
\r
177 vnfSoftwareVersion="V1",
\r
180 status='NOT_INSTANTIATED'
\r
182 mock_run.re.return_value = None
\r
183 response = self.client.post(
\r
184 "/api/vnflcm/v1/vnf_instances/123/terminate",
\r
189 status.HTTP_409_CONFLICT,
\r
190 response.status_code,
\r
194 def test_terminate_vnf_when_inst_id_not_exist(self):
\r
196 "terminationType": "GRACEFUL",
\r
197 "gracefulTerminationTimeout": 120
\r
199 self.nf_inst_id = str(uuid.uuid4())
\r
200 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
201 JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
\r
202 TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
203 self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")
\r
205 @mock.patch.object(restcall, 'call_req')
\r
206 @mock.patch.object(api, 'call')
\r
207 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
208 def test_terminate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):
\r
209 NfInstModel.objects.create(
\r
218 status='VNF_INSTANTIATED',
\r
221 vnfSoftwareVersion='',
\r
222 vnfConfigurableProperties='todo',
\r
223 localizationLanguage='EN_US',
\r
224 create_time=now_time()
\r
227 SubscriptionModel.objects.create(
\r
228 subscription_id=str(uuid.uuid4()),
\r
229 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
230 auth_info=json.JSONEncoder().encode({
\r
231 'authType': ['BASIC'],
\r
233 'userName': 'username',
\r
234 'password': 'password'
\r
237 notification_types=str([
\r
238 'VnfLcmOperationOccurrenceNotification',
\r
239 'VnfIdentifierCreationNotification',
\r
240 'VnfIdentifierDeletionNotification'
\r
242 operation_types=str(['TERMINATE']),
\r
243 operation_states=str(['COMPLETED']),
\r
244 vnf_instance_filter=json.JSONEncoder().encode({
\r
245 'vnfdIds': ['111'],
\r
246 'vnfProductsFromProviders': [],
\r
247 'vnfInstanceIds': ['1111'],
\r
248 'vnfInstanceNames': [],
\r
252 t1_apply_grant_result = [0, json.JSONEncoder().encode(
\r
255 "vnfInstanceId": "1",
\r
256 "vnfLcmOpOccId": "2",
\r
257 "vimConnections": [
\r
264 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
265 t3_delete_flavor = [0, json.JSONEncoder().encode({"vim_id": "vimid_1"}), '200']
\r
266 mock_call_req.side_effect = [
\r
267 t1_apply_grant_result,
\r
268 t2_lcm_notify_result,
\r
271 mock_call.return_value = None
\r
272 mock_post_notification.return_value = None
\r
274 "terminationType": "FORCEFUL",
\r
275 "gracefulTerminationTimeout": 120
\r
277 self.nf_inst_id = '1111'
\r
278 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
279 JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
\r
280 TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
281 self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")
\r