lcm/lcm/nf/vnfs/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.vnfs.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
  86     def tearDown(self):
  87         VmInstModel.objects.all().delete()
  88         NetworkInstModel.objects.all().delete()
  89         SubNetworkInstModel.objects.all().delete()
  90         PortInstModel.objects.all().delete()
  91
  92
  93     def assert_job_result(self, job_id, job_progress, job_detail):
  94         jobs = JobStatusModel.objects.filter(jobid=job_id,
  95                                              progress=job_progress,
  96                                              descp=job_detail)
  97         self.assertEqual(1, len(jobs))
  98
  99
 100     @mock.patch.object(restcall, 'call_req')
 101     def test_delete_vnf_identifier(self, mock_call_req):
 102         NfInstModel.objects.create(nfinstid='1111',
 103                                    nf_name='2222',
 104                                    package_id='todo',
 105                                    version='',
 106                                    vendor='',
 107                                    netype='',
 108                                    vnfd_model='',
 109                                    status='NOT_INSTANTIATED',
 110                                    nf_desc='',
 111                                    vnfdid='',
 112                                    vnfSoftwareVersion='',
 113                                    vnfConfigurableProperties='todo',
 114                                    localizationLanguage='EN_US',
 115                                    create_time=now_time())
 116         r1_create_vnf_to_aai = [0, json.JSONEncoder().encode({}), '200']
 117         mock_call_req.side_effect = [r1_create_vnf_to_aai]
 118         response = self.client.delete("/api/vnflcm/v1/vnf_instances/1111")
 119         self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 120         self.assertEqual(None, response.data)
 121
 122
 123     def test_delete_vnf_identifier_when_vnf_not_exist(self):
 124         response = self.client.delete("/api/vnflcm/v1/vnf_instances/1111")
 125         self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 126
 127
 128     @mock.patch.object(TermVnf, 'run')
 129     def test_terminate_vnf(self, mock_run):
 130         mock_run.re.return_value = None
 131         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/terminate", data={}, format='json')
 132         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
 133
 134
 135     def test_terminate_vnf_when_inst_id_not_exist(self):
 136         data = {
 137             "terminationType": "GRACEFUL",
 138             "gracefulTerminationTimeout": 120
 139         }
 140         self.nf_inst_id = str(uuid.uuid4())
 141         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
 142         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
 143         TermVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
 144         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")
 145
 146
 147     @mock.patch.object(restcall, 'call_req')
 148     @mock.patch.object(api, 'call')
 149     def test_terminate_vnf_success(self, mock_call, mock_call_req):
 150         NfInstModel.objects.create(nfinstid='1111',
 151                                    nf_name='2222',
 152                                    package_id='todo',
 153                                    version='',
 154                                    vendor='',
 155                                    netype='',
 156                                    vnfd_model='',
 157                                    status='VNF_INSTANTIATED',
 158                                    nf_desc='',
 159                                    vnfdid='',
 160                                    vnfSoftwareVersion='',
 161                                    vnfConfigurableProperties='todo',
 162                                    localizationLanguage='EN_US',
 163                                    create_time=now_time())
 164         t1_apply_grant_result = [0, json.JSONEncoder().encode(
 165             {
 166                 "vim":{
 167                     "vimid": 'vimid_1',
 168                     "accessinfo":{
 169                         "tenant": 'tenantname_1'
 170                     }
 171                 }
 172             }), '200']
 173         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
 174         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result]
 175         mock_call.return_value = None
 176         data = {
 177             "terminationType": "FORCEFUL",
 178             "gracefulTerminationTimeout": 120
 179         }
 180         self.nf_inst_id = '1111'
 181         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
 182         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
 183         TermVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
 184         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")