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', nf_name='VNF1').save()\r
 102         mock_run.re.return_value = None\r
 103         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/terminate", data=req_data, format='json')\r
 104         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)\r
 105 \r
 106     def test_terminate_vnf_when_inst_id_not_exist(self):\r
 107         data = {\r
 108             "terminationType": "GRACEFUL",\r
 109             "gracefulTerminationTimeout": 120\r
 110         }\r
 111         self.nf_inst_id = str(uuid.uuid4())\r
 112         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 113         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")\r
 114         TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 115         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")\r
 116 \r
 117     @mock.patch.object(restcall, 'call_req')\r
 118     @mock.patch.object(api, 'call')\r
 119     @mock.patch.object(NotificationsUtil, 'post_notification')\r
 120     def test_terminate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):\r
 121         NfInstModel.objects.create(nfinstid='1111',\r
 122                                    nf_name='2222',\r
 123                                    vnfminstid='1',\r
 124                                    package_id='todo',\r
 125                                    version='',\r
 126                                    vendor='',\r
 127                                    netype='',\r
 128                                    vnfd_model='',\r
 129                                    status='VNF_INSTANTIATED',\r
 130                                    nf_desc='',\r
 131                                    vnfdid='',\r
 132                                    vnfSoftwareVersion='',\r
 133                                    vnfConfigurableProperties='todo',\r
 134                                    localizationLanguage='EN_US',\r
 135                                    create_time=now_time())\r
 136 \r
 137         SubscriptionModel.objects.create(\r
 138             subscription_id=str(uuid.uuid4()),\r
 139             callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',\r
 140             auth_info=json.JSONEncoder().encode({\r
 141                 'authType': ['BASIC'],\r
 142                 'paramsBasic': {\r
 143                     'userName': 'username',\r
 144                     'password': 'password'\r
 145                 }\r
 146             }),\r
 147             notification_types=str([\r
 148                 'VnfLcmOperationOccurrenceNotification',\r
 149                 'VnfIdentifierCreationNotification',\r
 150                 'VnfIdentifierDeletionNotification'\r
 151             ]),\r
 152             operation_types=str(['TERMINATE']),\r
 153             operation_states=str(['COMPLETED']),\r
 154             vnf_instance_filter=json.JSONEncoder().encode({\r
 155                 'vnfdIds': ['111'],\r
 156                 'vnfProductsFromProviders': [],\r
 157                 'vnfInstanceIds': ['1111'],\r
 158                 'vnfInstanceNames': [],\r
 159             })\r
 160         )\r
 161 \r
 162         t1_apply_grant_result = [0, json.JSONEncoder().encode(\r
 163             {\r
 164                 "id": "1",\r
 165                 "vnfInstanceId": "1",\r
 166                 "vnfLcmOpOccId": "2",\r
 167                 "vimConnections": [\r
 168                     {\r
 169                         "id": "1",\r
 170                         "vimId": "1"\r
 171                     }\r
 172                 ]\r
 173             }), '200']\r
 174         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 175         t3_delete_flavor = [0, json.JSONEncoder().encode({"vim_id": "vimid_1"}), '200']\r
 176         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_delete_flavor]\r
 177         mock_call.return_value = None\r
 178         mock_post_notification.return_value = None\r
 179         data = {\r
 180             "terminationType": "FORCEFUL",\r
 181             "gracefulTerminationTimeout": 120\r
 182         }\r
 183         self.nf_inst_id = '1111'\r
 184         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 185         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")\r
 186         TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 187         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")\r