lcm/lcm/nf/tests/test_vnf_cancel.py

   1 # Copyright 2017 ZTE Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15
  16 import json
  17 import uuid
  18
  19 import mock
  20 from django.test import TestCase, Client
  21 from rest_framework import status
  22
  23 from lcm.nf.vnf_cancel.term_vnf import TermVnf
  24 from lcm.pub.database.models import NfInstModel, JobStatusModel, VmInstModel, NetworkInstModel, SubNetworkInstModel, \
  25     PortInstModel, FlavourInstModel, StorageInstModel, NfvoRegInfoModel
  26 from lcm.pub.utils import restcall
  27 from lcm.pub.utils.jobutil import JobUtil
  28 from lcm.pub.utils.timeutil import now_time
  29 from lcm.pub.vimapi import api
  30
  31
  32 class TestNFTerminate(TestCase):
  33     def setUp(self):
  34         self.client = Client()
  35         StorageInstModel.objects.create(storageid="1",
  36                                         vimid="1",
  37                                         resouceid="11",
  38                                         insttype=0,
  39                                         instid="1111",
  40                                         is_predefined=1)
  41         NetworkInstModel.objects.create(networkid='1',
  42                                         vimid='1',
  43                                         resouceid='1',
  44                                         name='pnet_network',
  45                                         is_predefined=1,
  46                                         tenant='admin',
  47                                         insttype=0,
  48                                         instid='1111')
  49         SubNetworkInstModel.objects.create(subnetworkid='1',
  50                                            vimid='1',
  51                                            resouceid='1',
  52                                            networkid='1',
  53                                            is_predefined=1,
  54                                            name='sub_pnet',
  55                                            tenant='admin',
  56                                            insttype=0,
  57                                            instid='1111')
  58         PortInstModel.objects.create(portid='1',
  59                                      networkid='1',
  60                                      subnetworkid='1',
  61                                      vimid='1',
  62                                      resouceid='1',
  63                                      is_predefined=1,
  64                                      name='aaa_pnet_cp',
  65                                      tenant='admin',
  66                                      insttype=0,
  67                                      instid='1111')
  68         FlavourInstModel.objects.create(flavourid="1",
  69                                         vimid="1",
  70                                         resouceid="11",
  71                                         instid="1111",
  72                                         is_predefined=1)
  73         VmInstModel.objects.create(vmid="1",
  74                                    vimid="1",
  75                                    resouceid="11",
  76                                    insttype=0,
  77                                    instid="1111",
  78                                    vmname="test_01",
  79                                    is_predefined=1,
  80                                    operationalstate=1)
  81         NfvoRegInfoModel.objects.create(nfvoid='1111',
  82                                         vnfminstid='11111',
  83                                         apiurl='1')
  84
  85     def tearDown(self):
  86         VmInstModel.objects.all().delete()
  87         NetworkInstModel.objects.all().delete()
  88         SubNetworkInstModel.objects.all().delete()
  89         PortInstModel.objects.all().delete()
  90
  91     def assert_job_result(self, job_id, job_progress, job_detail):
  92         jobs = JobStatusModel.objects.filter(jobid=job_id,
  93                                              progress=job_progress,
  94                                              descp=job_detail)
  95         self.assertEqual(1, len(jobs))
  96
  97     def test_delete_vnf_identifier(self):
  98         NfInstModel.objects.create(nfinstid='1111',
  99                                    nf_name='2222',
 100                                    package_id='todo',
 101                                    version='',
 102                                    vendor='',
 103                                    netype='',
 104                                    vnfd_model='',
 105                                    status='NOT_INSTANTIATED',
 106                                    nf_desc='',
 107                                    vnfdid='',
 108                                    vnfSoftwareVersion='',
 109                                    vnfConfigurableProperties='todo',
 110                                    localizationLanguage='EN_US',
 111                                    create_time=now_time())
 112         response = self.client.delete("/api/vnflcm/v1/vnf_instances/1111")
 113         self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 114         self.assertEqual(None, response.data)
 115
 116     def test_delete_vnf_identifier_when_vnf_not_exist(self):
 117         response = self.client.delete("/api/vnflcm/v1/vnf_instances/1111")
 118         self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 119
 120     @mock.patch.object(TermVnf, 'run')
 121     def test_terminate_vnf(self, mock_run):
 122         req_data = {
 123             "terminationType": "GRACEFUL",
 124             "gracefulTerminationTimeout": 120
 125         }
 126         mock_run.re.return_value = None
 127         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/terminate", data=req_data, format='json')
 128         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
 129
 130     def test_terminate_vnf_when_inst_id_not_exist(self):
 131         data = {
 132             "terminationType": "GRACEFUL",
 133             "gracefulTerminationTimeout": 120
 134         }
 135         self.nf_inst_id = str(uuid.uuid4())
 136         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
 137         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
 138         TermVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
 139         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")
 140
 141     @mock.patch.object(restcall, 'call_req')
 142     @mock.patch.object(api, 'call')
 143     def test_terminate_vnf_success(self, mock_call, mock_call_req):
 144         NfInstModel.objects.create(nfinstid='1111',
 145                                    nf_name='2222',
 146                                    package_id='todo',
 147                                    version='',
 148                                    vendor='',
 149                                    netype='',
 150                                    vnfd_model='',
 151                                    status='VNF_INSTANTIATED',
 152                                    nf_desc='',
 153                                    vnfdid='',
 154                                    vnfSoftwareVersion='',
 155                                    vnfConfigurableProperties='todo',
 156                                    localizationLanguage='EN_US',
 157                                    create_time=now_time())
 158         t1_apply_grant_result = [0, json.JSONEncoder().encode(
 159             {
 160                 "vimid": 'vimid_1',
 161                 "tenant": 'tenantname_1'
 162             }), '200']
 163         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
 164         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result]
 165         mock_call.return_value = None
 166         data = {
 167             "terminationType": "FORCEFUL",
 168             "gracefulTerminationTimeout": 120
 169         }
 170         self.nf_inst_id = '1111'
 171         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
 172         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
 173         TermVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
 174         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")