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
20 from django.test import TestCase, Client
\r
21 from rest_framework import status
\r
23 from lcm.nf.biz.terminate_vnf import TerminateVnf
\r
24 from lcm.pub.database.models import NfInstModel, JobStatusModel, VmInstModel, NetworkInstModel, SubNetworkInstModel, \
\r
25 PortInstModel, FlavourInstModel, StorageInstModel, SubscriptionModel
\r
26 from lcm.pub.utils import restcall
\r
27 from lcm.pub.utils.jobutil import JobUtil
\r
28 from lcm.pub.utils.timeutil import now_time
\r
29 from lcm.pub.utils.notificationsutil import NotificationsUtil
\r
30 from lcm.pub.vimapi import api
\r
33 class TestNFTerminate(TestCase):
\r
35 self.client = Client()
\r
36 StorageInstModel.objects.create(storageid="1",
\r
42 NetworkInstModel.objects.create(networkid='1',
\r
45 name='pnet_network',
\r
50 SubNetworkInstModel.objects.create(subnetworkid='1',
\r
59 PortInstModel.objects.create(portid='1',
\r
69 FlavourInstModel.objects.create(flavourid="1",
\r
74 VmInstModel.objects.create(vmid="1",
\r
84 VmInstModel.objects.all().delete()
\r
85 NetworkInstModel.objects.all().delete()
\r
86 SubNetworkInstModel.objects.all().delete()
\r
87 PortInstModel.objects.all().delete()
\r
88 NfInstModel.objects.all().delete()
\r
90 def assert_job_result(self, job_id, job_progress, job_detail):
\r
91 jobs = JobStatusModel.objects.filter(jobid=job_id,
\r
92 progress=job_progress,
\r
94 self.assertEqual(1, len(jobs))
\r
96 @mock.patch.object(TerminateVnf, 'run')
\r
97 def test_terminate_vnf(self, mock_run):
\r
99 "terminationType": "GRACEFUL",
\r
100 "gracefulTerminationTimeout": 120
\r
102 NfInstModel(nfinstid='12',
\r
104 nf_desc="VNF DESC",
\r
108 vnfSoftwareVersion="V1",
\r
111 status='INSTANTIATED').save()
\r
112 mock_run.re.return_value = None
\r
113 response = self.client.post("/api/vnflcm/v1/vnf_instances/12/terminate", data=req_data, format='json')
\r
114 self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
\r
116 @mock.patch.object(TerminateVnf, 'run')
\r
117 def test_terminate_vnf_not_found(self, mock_run):
\r
119 "terminationType": "GRACEFUL",
\r
120 "gracefulTerminationTimeout": 120
\r
122 mock_run.re.return_value = None
\r
123 response = self.client.post("/api/vnflcm/v1/vnf_instances/567/terminate", data=req_data, format='json')
\r
124 self.failUnlessEqual(status.HTTP_404_NOT_FOUND, response.status_code, response.content)
\r
126 @mock.patch.object(TerminateVnf, 'run')
\r
127 def test_terminate_vnf_conflict(self, mock_run):
\r
129 "terminationType": "GRACEFUL",
\r
130 "gracefulTerminationTimeout": 120
\r
132 NfInstModel(nfinstid='123',
\r
134 nf_desc="VNF DESC",
\r
138 vnfSoftwareVersion="V1",
\r
141 status='NOT_INSTANTIATED').save()
\r
142 mock_run.re.return_value = None
\r
143 response = self.client.post("/api/vnflcm/v1/vnf_instances/123/terminate", data=req_data, format='json')
\r
144 self.failUnlessEqual(status.HTTP_409_CONFLICT, response.status_code, response.content)
\r
146 def test_terminate_vnf_when_inst_id_not_exist(self):
\r
148 "terminationType": "GRACEFUL",
\r
149 "gracefulTerminationTimeout": 120
\r
151 self.nf_inst_id = str(uuid.uuid4())
\r
152 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
153 JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
\r
154 TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
155 self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")
\r
157 @mock.patch.object(restcall, 'call_req')
\r
158 @mock.patch.object(api, 'call')
\r
159 @mock.patch.object(NotificationsUtil, 'post_notification')
\r
160 def test_terminate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):
\r
161 NfInstModel.objects.create(nfinstid='1111',
\r
169 status='VNF_INSTANTIATED',
\r
172 vnfSoftwareVersion='',
\r
173 vnfConfigurableProperties='todo',
\r
174 localizationLanguage='EN_US',
\r
175 create_time=now_time())
\r
177 SubscriptionModel.objects.create(
\r
178 subscription_id=str(uuid.uuid4()),
\r
179 callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',
\r
180 auth_info=json.JSONEncoder().encode({
\r
181 'authType': ['BASIC'],
\r
183 'userName': 'username',
\r
184 'password': 'password'
\r
187 notification_types=str([
\r
188 'VnfLcmOperationOccurrenceNotification',
\r
189 'VnfIdentifierCreationNotification',
\r
190 'VnfIdentifierDeletionNotification'
\r
192 operation_types=str(['TERMINATE']),
\r
193 operation_states=str(['COMPLETED']),
\r
194 vnf_instance_filter=json.JSONEncoder().encode({
\r
195 'vnfdIds': ['111'],
\r
196 'vnfProductsFromProviders': [],
\r
197 'vnfInstanceIds': ['1111'],
\r
198 'vnfInstanceNames': [],
\r
202 t1_apply_grant_result = [0, json.JSONEncoder().encode(
\r
205 "vnfInstanceId": "1",
\r
206 "vnfLcmOpOccId": "2",
\r
207 "vimConnections": [
\r
214 t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
\r
215 t3_delete_flavor = [0, json.JSONEncoder().encode({"vim_id": "vimid_1"}), '200']
\r
216 mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_delete_flavor]
\r
217 mock_call.return_value = None
\r
218 mock_post_notification.return_value = None
\r
220 "terminationType": "FORCEFUL",
\r
221 "gracefulTerminationTimeout": 120
\r
223 self.nf_inst_id = '1111'
\r
224 self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
\r
225 JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
\r
226 TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
\r
227 self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")
\r