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, NfvoRegInfoModel\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.vimapi import api\r
  30 \r
  31 \r
  32 class TestNFTerminate(TestCase):\r
  33     def setUp(self):\r
  34         self.client = Client()\r
  35         StorageInstModel.objects.create(storageid="1",\r
  36                                         vimid="1",\r
  37                                         resouceid="11",\r
  38                                         insttype=0,\r
  39                                         instid="1111",\r
  40                                         is_predefined=1)\r
  41         NetworkInstModel.objects.create(networkid='1',\r
  42                                         vimid='1',\r
  43                                         resouceid='1',\r
  44                                         name='pnet_network',\r
  45                                         is_predefined=1,\r
  46                                         tenant='admin',\r
  47                                         insttype=0,\r
  48                                         instid='1111')\r
  49         SubNetworkInstModel.objects.create(subnetworkid='1',\r
  50                                            vimid='1',\r
  51                                            resouceid='1',\r
  52                                            networkid='1',\r
  53                                            is_predefined=1,\r
  54                                            name='sub_pnet',\r
  55                                            tenant='admin',\r
  56                                            insttype=0,\r
  57                                            instid='1111')\r
  58         PortInstModel.objects.create(portid='1',\r
  59                                      networkid='1',\r
  60                                      subnetworkid='1',\r
  61                                      vimid='1',\r
  62                                      resouceid='1',\r
  63                                      is_predefined=1,\r
  64                                      name='aaa_pnet_cp',\r
  65                                      tenant='admin',\r
  66                                      insttype=0,\r
  67                                      instid='1111')\r
  68         FlavourInstModel.objects.create(flavourid="1",\r
  69                                         vimid="1",\r
  70                                         resouceid="11",\r
  71                                         instid="1111",\r
  72                                         is_predefined=1)\r
  73         VmInstModel.objects.create(vmid="1",\r
  74                                    vimid="1",\r
  75                                    resouceid="11",\r
  76                                    insttype=0,\r
  77                                    instid="1111",\r
  78                                    vmname="test_01",\r
  79                                    is_predefined=1,\r
  80                                    operationalstate=1)\r
  81         NfvoRegInfoModel.objects.create(nfvoid='1111',\r
  82                                         vnfminstid='11111',\r
  83                                         apiurl='1')\r
  84 \r
  85     def tearDown(self):\r
  86         VmInstModel.objects.all().delete()\r
  87         NetworkInstModel.objects.all().delete()\r
  88         SubNetworkInstModel.objects.all().delete()\r
  89         PortInstModel.objects.all().delete()\r
  90 \r
  91     def assert_job_result(self, job_id, job_progress, job_detail):\r
  92         jobs = JobStatusModel.objects.filter(jobid=job_id,\r
  93                                              progress=job_progress,\r
  94                                              descp=job_detail)\r
  95         self.assertEqual(1, len(jobs))\r
  96 \r
  97     @mock.patch.object(TerminateVnf, 'run')\r
  98     def test_terminate_vnf(self, mock_run):\r
  99         req_data = {\r
 100             "terminationType": "GRACEFUL",\r
 101             "gracefulTerminationTimeout": 120\r
 102         }\r
 103         NfInstModel(nfinstid='12', nf_name='VNF1').save()\r
 104         mock_run.re.return_value = None\r
 105         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/terminate", data=req_data, format='json')\r
 106         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)\r
 107 \r
 108     def test_terminate_vnf_when_inst_id_not_exist(self):\r
 109         data = {\r
 110             "terminationType": "GRACEFUL",\r
 111             "gracefulTerminationTimeout": 120\r
 112         }\r
 113         self.nf_inst_id = str(uuid.uuid4())\r
 114         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 115         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")\r
 116         TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 117         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")\r
 118 \r
 119     @mock.patch.object(restcall, 'call_req')\r
 120     @mock.patch.object(api, 'call')\r
 121     def test_terminate_vnf_success(self, mock_call, mock_call_req):\r
 122         NfInstModel.objects.create(nfinstid='1111',\r
 123                                    nf_name='2222',\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         t1_apply_grant_result = [0, json.JSONEncoder().encode(\r
 137             {\r
 138                 "vimid": 'vimid_1',\r
 139                 "tenant": 'tenantname_1'\r
 140             }), '200']\r
 141         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 142         t3_delete_flavor = [0, json.JSONEncoder().encode({"vim_id": "vimid_1"}), '200']\r
 143         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_delete_flavor]\r
 144         mock_call.return_value = None\r
 145         data = {\r
 146             "terminationType": "FORCEFUL",\r
 147             "gracefulTerminationTimeout": 120\r
 148         }\r
 149         self.nf_inst_id = '1111'\r
 150         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 151         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")\r
 152         TerminateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 153         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")\r