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     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     @mock.patch.object(restcall, 'call_req')
  98     def test_delete_vnf_identifier(self, mock_call_req):
  99         NfInstModel.objects.create(nfinstid='1111',
 100                                    nf_name='2222',
 101                                    package_id='todo',
 102                                    version='',
 103                                    vendor='',
 104                                    netype='',
 105                                    vnfd_model='',
 106                                    status='NOT_INSTANTIATED',
 107                                    nf_desc='',
 108                                    vnfdid='',
 109                                    vnfSoftwareVersion='',
 110                                    vnfConfigurableProperties='todo',
 111                                    localizationLanguage='EN_US',
 112                                    create_time=now_time())
 113         vnf_info = {
 114             "vnf-id": "vnf-id-test111",
 115             "vnf-name": "vnf-name-test111",
 116             "vnf-type": "vnf-type-test111",
 117             "in-maint": True,
 118             "is-closed-loop-disabled": False,
 119             "resource-version": "1505465356262"
 120         }
 121         r1_query_vnf_to_aai = [0, json.JSONEncoder().encode(vnf_info), '200']
 122         r1_delete_vnf_to_aai = [0, json.JSONEncoder().encode({}), '200']
 123         mock_call_req.side_effect = [r1_query_vnf_to_aai, r1_delete_vnf_to_aai]
 124         response = self.client.delete("/api/vnflcm/v1/vnf_instances/1111")
 125         self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 126         self.assertEqual(None, response.data)
 127
 128     def test_delete_vnf_identifier_when_vnf_not_exist(self):
 129         response = self.client.delete("/api/vnflcm/v1/vnf_instances/1111")
 130         self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 131
 132     @mock.patch.object(TermVnf, 'run')
 133     def test_terminate_vnf(self, mock_run):
 134         mock_run.re.return_value = None
 135         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/terminate", data={}, format='json')
 136         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
 137
 138     def test_terminate_vnf_when_inst_id_not_exist(self):
 139         data = {
 140             "terminationType": "GRACEFUL",
 141             "gracefulTerminationTimeout": 120
 142         }
 143         self.nf_inst_id = str(uuid.uuid4())
 144         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
 145         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
 146         TermVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
 147         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")
 148
 149     @mock.patch.object(restcall, 'call_req')
 150     @mock.patch.object(api, 'call')
 151     def test_terminate_vnf_success(self, mock_call, mock_call_req):
 152         NfInstModel.objects.create(nfinstid='1111',
 153                                    nf_name='2222',
 154                                    package_id='todo',
 155                                    version='',
 156                                    vendor='',
 157                                    netype='',
 158                                    vnfd_model='',
 159                                    status='VNF_INSTANTIATED',
 160                                    nf_desc='',
 161                                    vnfdid='',
 162                                    vnfSoftwareVersion='',
 163                                    vnfConfigurableProperties='todo',
 164                                    localizationLanguage='EN_US',
 165                                    create_time=now_time())
 166         t1_apply_grant_result = [0, json.JSONEncoder().encode(
 167             {
 168                 "vim":{
 169                     "vimid": 'vimid_1',
 170                     "accessinfo":{
 171                         "tenant": 'tenantname_1'
 172                     }
 173                 }
 174             }), '200']
 175         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']
 176         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result]
 177         mock_call.return_value = None
 178         data = {
 179             "terminationType": "FORCEFUL",
 180             "gracefulTerminationTimeout": 120
 181         }
 182         self.nf_inst_id = '1111'
 183         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)
 184         JobUtil.add_job_status(self.job_id, 0, "INST_VNF_READY")
 185         TermVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()
 186         self.assert_job_result(self.job_id, 100, "Terminate Vnf success.")