lcm/lcm/nf/tests/test_terminate_vnf.py

   1 # Copyright 2017 ZTE Corporation.\r
   2 #\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
   6 #\r
   7 #         http://www.apache.org/licenses/LICENSE-2.0\r
   8 #\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
  14 \r
  15 import json\r
  16 import uuid\r
  17 import mock\r
  18 \r
  19 from django.test import TestCase, Client\r
  20 from rest_framework import status\r
  21 \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
  37 \r
  38 \r
  39 class TestNFTerminate(TestCase):\r
  40     def setUp(self):\r
  41         self.client = Client()\r
  42         StorageInstModel.objects.create(\r
  43             storageid="1",\r
  44             vimid="1",\r
  45             resourceid="11",\r
  46             insttype=0,\r
  47             instid="1111",\r
  48             is_predefined=1\r
  49         )\r
  50         NetworkInstModel.objects.create(\r
  51             networkid='1',\r
  52             vimid='1',\r
  53             resourceid='1',\r
  54             name='pnet_network',\r
  55             is_predefined=1,\r
  56             tenant='admin',\r
  57             insttype=0,\r
  58             instid='1111'\r
  59         )\r
  60         SubNetworkInstModel.objects.create(\r
  61             subnetworkid='1',\r
  62             vimid='1',\r
  63             resourceid='1',\r
  64             networkid='1',\r
  65             is_predefined=1,\r
  66             name='sub_pnet',\r
  67             tenant='admin',\r
  68             insttype=0,\r
  69             instid='1111'\r
  70         )\r
  71         PortInstModel.objects.create(\r
  72             portid='1',\r
  73             networkid='1',\r
  74             subnetworkid='1',\r
  75             vimid='1',\r
  76             resourceid='1',\r
  77             is_predefined=1,\r
  78             name='aaa_pnet_cp',\r
  79             tenant='admin',\r
  80             insttype=0,\r
  81             instid='1111'\r
  82         )\r
  83         FlavourInstModel.objects.create(\r
  84             flavourid="1",\r
  85             vimid="1",\r
  86             resourceid="11",\r
  87             instid="1111",\r
  88             is_predefined=1\r
  89         )\r
  90         VmInstModel.objects.create(\r
  91             vmid="1",\r
  92             vimid="1",\r
  93             resourceid="11",\r
  94             insttype=0,\r
  95             instid="1111",\r
  96             vmname="test_01",\r
  97             is_predefined=1,\r
  98             operationalstate=1\r
  99         )\r
 100 \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
 107 \r
 108     def assert_job_result(self, job_id, job_progress, job_detail):\r
 109         jobs = JobStatusModel.objects.filter(\r
 110             jobid=job_id,\r
 111             progress=job_progress,\r
 112             descp=job_detail\r
 113         )\r
 114         self.assertEqual(1, len(jobs))\r
 115 \r
 116     @mock.patch.object(TerminateVnf, 'run')\r
 117     def test_terminate_vnf(self, mock_run):\r
 118         req_data = {\r
 119             "terminationType": "GRACEFUL",\r
 120             "gracefulTerminationTimeout": 120\r
 121         }\r
 122         NfInstModel(\r
 123             nfinstid='12',\r
 124             nf_name='VNF1',\r
 125             nf_desc="VNF DESC",\r
 126             vnfdid="1",\r
 127             netype="XGW",\r
 128             vendor="ZTE",\r
 129             vnfSoftwareVersion="V1",\r
 130             version="V1",\r
 131             package_id="2",\r
 132             status='INSTANTIATED'\r
 133         ).save()\r
 134         mock_run.re.return_value = None\r
 135         response = self.client.post(\r
 136             "/api/vnflcm/v1/vnf_instances/12/terminate",\r
 137             data=req_data,\r
 138             format='json'\r
 139         )\r
 140         self.failUnlessEqual(\r
 141             status.HTTP_202_ACCEPTED,\r
 142             response.status_code,\r
 143             response.content\r
 144         )\r
 145 \r
 146     @mock.patch.object(TerminateVnf, 'run')\r
 147     def test_terminate_vnf_not_found(self, mock_run):\r
 148         req_data = {\r
 149             "terminationType": "GRACEFUL",\r
 150             "gracefulTerminationTimeout": 120\r
 151         }\r
 152         mock_run.re.return_value = None\r
 153         response = self.client.post(\r
 154             "/api/vnflcm/v1/vnf_instances/567/terminate",\r
 155             data=req_data,\r
 156             format='json'\r
 157         )\r
 158         self.failUnlessEqual(\r
 159             status.HTTP_404_NOT_FOUND,\r
 160             response.status_code,\r
 161             response.content\r
 162         )\r
 163 \r
 164     @mock.patch.object(TerminateVnf, 'run')\r
 165     def test_terminate_vnf_conflict(self, mock_run):\r
 166         req_data = {\r
 167             "terminationType": "GRACEFUL",\r
 168             "gracefulTerminationTimeout": 120\r
 169         }\r
 170         NfInstModel(\r
 171             nfinstid='123',\r
 172             nf_name='VNF1',\r
 173             nf_desc="VNF DESC",\r
 174             vnfdid="1",\r
 175             netype="XGW",\r
 176             vendor="ZTE",\r
 177             vnfSoftwareVersion="V1",\r
 178             version="V1",\r
 179             package_id="2",\r
 180             status='NOT_INSTANTIATED'\r
 181         ).save()\r
 182         mock_run.re.return_value = None\r
 183         response = self.client.post(\r
 184             "/api/vnflcm/v1/vnf_instances/123/terminate",\r
 185             data=req_data,\r
 186             format='json'\r
 187         )\r
 188         self.failUnlessEqual(\r
 189             status.HTTP_409_CONFLICT,\r
 190             response.status_code,\r
 191             response.content\r
 192         )\r
 193 \r
 194     def test_terminate_vnf_when_inst_id_not_exist(self):\r
 195         data = {\r
 196             "terminationType": "GRACEFUL",\r
 197             "gracefulTerminationTimeout": 120\r
 198         }\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
 204 \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
 210             nfinstid='1111',\r
 211             nf_name='2222',\r
 212             vnfminstid='1',\r
 213             package_id='todo',\r
 214             version='',\r
 215             vendor='',\r
 216             netype='',\r
 217             vnfd_model='',\r
 218             status='VNF_INSTANTIATED',\r
 219             nf_desc='',\r
 220             vnfdid='',\r
 221             vnfSoftwareVersion='',\r
 222             vnfConfigurableProperties='todo',\r
 223             localizationLanguage='EN_US',\r
 224             create_time=now_time()\r
 225         )\r
 226 \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
 232                 'paramsBasic': {\r
 233                     'userName': 'username',\r
 234                     'password': 'password'\r
 235                 }\r
 236             }),\r
 237             notification_types=str([\r
 238                 'VnfLcmOperationOccurrenceNotification',\r
 239                 'VnfIdentifierCreationNotification',\r
 240                 'VnfIdentifierDeletionNotification'\r
 241             ]),\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
 249             })\r
 250         )\r
 251 \r
 252         t1_apply_grant_result = [0, json.JSONEncoder().encode(\r
 253             {\r
 254                 "id": "1",\r
 255                 "vnfInstanceId": "1",\r
 256                 "vnfLcmOpOccId": "2",\r
 257                 "vimConnections": [\r
 258                     {\r
 259                         "id": "1",\r
 260                         "vimId": "1"\r
 261                     }\r
 262                 ]\r
 263             }), '200']\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
 269             t3_delete_flavor\r
 270         ]\r
 271         mock_call.return_value = None\r
 272         mock_post_notification.return_value = None\r
 273         data = {\r
 274             "terminationType": "FORCEFUL",\r
 275             "gracefulTerminationTimeout": 120\r
 276         }\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