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 \r
  16 import json\r
  17 import uuid\r
  18 \r
  19 import mock\r
  20 from django.test import TestCase, Client\r
  21 from rest_framework import status\r
  22 \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
  31 \r
  32 \r
  33 class TestNFTerminate(TestCase):\r
  34     def setUp(self):\r
  35         self.client = Client()\r
  36         StorageInstModel.objects.create(storageid="1",\r
  37                                         vimid="1",\r
  38                                         resourceid="11",\r
  39                                         insttype=0,\r
  40                                         instid="1111",\r
  41                                         is_predefined=1)\r
  42         NetworkInstModel.objects.create(networkid='1',\r
  43                                         vimid='1',\r
  44                                         resourceid='1',\r
  45                                         name='pnet_network',\r
  46                                         is_predefined=1,\r
  47                                         tenant='admin',\r
  48                                         insttype=0,\r
  49                                         instid='1111')\r
  50         SubNetworkInstModel.objects.create(subnetworkid='1',\r
  51                                            vimid='1',\r
  52                                            resourceid='1',\r
  53                                            networkid='1',\r
  54                                            is_predefined=1,\r
  55                                            name='sub_pnet',\r
  56                                            tenant='admin',\r
  57                                            insttype=0,\r
  58                                            instid='1111')\r
  59         PortInstModel.objects.create(portid='1',\r
  60                                      networkid='1',\r
  61                                      subnetworkid='1',\r
  62                                      vimid='1',\r
  63                                      resourceid='1',\r
  64                                      is_predefined=1,\r
  65                                      name='aaa_pnet_cp',\r
  66                                      tenant='admin',\r
  67                                      insttype=0,\r
  68                                      instid='1111')\r
  69         FlavourInstModel.objects.create(flavourid="1",\r
  70                                         vimid="1",\r
  71                                         resourceid="11",\r
  72                                         instid="1111",\r
  73                                         is_predefined=1)\r
  74         VmInstModel.objects.create(vmid="1",\r
  75                                    vimid="1",\r
  76                                    resourceid="11",\r
  77                                    insttype=0,\r
  78                                    instid="1111",\r
  79                                    vmname="test_01",\r
  80                                    is_predefined=1,\r
  81                                    operationalstate=1)\r
  82 \r
  83     def tearDown(self):\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 \r
  89     def assert_job_result(self, job_id, job_progress, job_detail):\r
  90         jobs = JobStatusModel.objects.filter(jobid=job_id,\r
  91                                              progress=job_progress,\r
  92                                              descp=job_detail)\r
  93         self.assertEqual(1, len(jobs))\r
  94 \r
  95     @mock.patch.object(TerminateVnf, 'run')\r
  96     def test_terminate_vnf(self, mock_run):\r
  97         req_data = {\r
  98             "terminationType": "GRACEFUL",\r
  99             "gracefulTerminationTimeout": 120\r
 100         }\r
 101         NfInstModel(nfinstid='12',\r
 102                     nf_name='VNF1',\r
 103                     nf_desc="VNF DESC",\r
 104                     vnfdid="1",\r
 105                     netype="XGW",\r
 106                     vendor="ZTE",\r
 107                     vnfSoftwareVersion="V1",\r
 108                     version="V1",\r
 109                     package_id="2",\r
 110                     status='INSTANTIATED').save()\r
 111         mock_run.re.return_value = None\r
 112         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/terminate", data=req_data, format='json')\r
 113         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)\r
 114 \r
 115     def test_terminate_vnf_when_inst_id_not_exist(self):\r
 116         data = {\r
 117             "terminationType": "GRACEFUL",\r
 118             "gracefulTerminationTimeout": 120\r
 119         }\r
 120         self.nf_inst_id = str(uuid.uuid4())\r
 121         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 122         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")\r
 123         TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 124         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")\r
 125 \r
 126     @mock.patch.object(restcall, 'call_req')\r
 127     @mock.patch.object(api, 'call')\r
 128     @mock.patch.object(NotificationsUtil, 'post_notification')\r
 129     def test_terminate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):\r
 130         NfInstModel.objects.create(nfinstid='1111',\r
 131                                    nf_name='2222',\r
 132                                    vnfminstid='1',\r
 133                                    package_id='todo',\r
 134                                    version='',\r
 135                                    vendor='',\r
 136                                    netype='',\r
 137                                    vnfd_model='',\r
 138                                    status='VNF_INSTANTIATED',\r
 139                                    nf_desc='',\r
 140                                    vnfdid='',\r
 141                                    vnfSoftwareVersion='',\r
 142                                    vnfConfigurableProperties='todo',\r
 143                                    localizationLanguage='EN_US',\r
 144                                    create_time=now_time())\r
 145 \r
 146         SubscriptionModel.objects.create(\r
 147             subscription_id=str(uuid.uuid4()),\r
 148             callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',\r
 149             auth_info=json.JSONEncoder().encode({\r
 150                 'authType': ['BASIC'],\r
 151                 'paramsBasic': {\r
 152                     'userName': 'username',\r
 153                     'password': 'password'\r
 154                 }\r
 155             }),\r
 156             notification_types=str([\r
 157                 'VnfLcmOperationOccurrenceNotification',\r
 158                 'VnfIdentifierCreationNotification',\r
 159                 'VnfIdentifierDeletionNotification'\r
 160             ]),\r
 161             operation_types=str(['TERMINATE']),\r
 162             operation_states=str(['COMPLETED']),\r
 163             vnf_instance_filter=json.JSONEncoder().encode({\r
 164                 'vnfdIds': ['111'],\r
 165                 'vnfProductsFromProviders': [],\r
 166                 'vnfInstanceIds': ['1111'],\r
 167                 'vnfInstanceNames': [],\r
 168             })\r
 169         )\r
 170 \r
 171         t1_apply_grant_result = [0, json.JSONEncoder().encode(\r
 172             {\r
 173                 "id": "1",\r
 174                 "vnfInstanceId": "1",\r
 175                 "vnfLcmOpOccId": "2",\r
 176                 "vimConnections": [\r
 177                     {\r
 178                         "id": "1",\r
 179                         "vimId": "1"\r
 180                     }\r
 181                 ]\r
 182             }), '200']\r
 183         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 184         t3_delete_flavor = [0, json.JSONEncoder().encode({"vim_id": "vimid_1"}), '200']\r
 185         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_delete_flavor]\r
 186         mock_call.return_value = None\r
 187         mock_post_notification.return_value = None\r
 188         data = {\r
 189             "terminationType": "FORCEFUL",\r
 190             "gracefulTerminationTimeout": 120\r
 191         }\r
 192         self.nf_inst_id = '1111'\r
 193         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 194         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")\r
 195         TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 196         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")\r