lcm/lcm/nf/tests/test_change_ext_conn.py

   1 # Copyright (C) 2019 ZTE. All Rights Reserved.
   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 import json
  15 import mock
  16
  17 from django.test import TestCase
  18 from rest_framework import status
  19 from rest_framework.test import APIClient
  20
  21 from lcm.pub.database.models import NfInstModel, JobStatusModel, StorageInstModel, NetworkInstModel, \
  22     SubNetworkInstModel, PortInstModel, FlavourInstModel, VmInstModel, VNFCInstModel
  23 from lcm.pub.exceptions import NFLCMException
  24 from lcm.pub.utils import restcall
  25 from lcm.pub.vimapi import api
  26 from lcm.pub.utils.jobutil import JobUtil
  27 from lcm.nf.biz.change_ext_conn import ChangeExtConn
  28 from . import const
  29
  30
  31 class TestChangeExtConn(TestCase):
  32     def setUp(self):
  33         self.client = APIClient()
  34         NfInstModel(nfinstid='12345',
  35                     nf_name='VNF1',
  36                     nf_desc="VNF DESC",
  37                     vnfdid="1",
  38                     netype="XGW",
  39                     vendor="ZTE",
  40                     vnfSoftwareVersion="V1",
  41                     version="V1",
  42                     package_id="2",
  43                     status='NOT_INSTANTIATED').save()
  44         NfInstModel(nfinstid='123',
  45                     nf_name='VNF1',
  46                     nf_desc="VNF DESC",
  47                     vnfdid="1",
  48                     netype="XGW",
  49                     vendor="ZTE",
  50                     vnfSoftwareVersion="V1",
  51                     version="V1",
  52                     package_id="2",
  53                     status='INSTANTIATED').save()
  54         self.req_data = {
  55             "extVirtualLinks": [{
  56                 "id": "string",
  57                 "resourceId": "329efb86-5cbb-4fc0-bc7c-6ea28f9d7389",
  58                 "resourceSubnetId": "429efb86-5cbb-4fc0-bc7c-6ea28f9d7389",
  59                 "extCps": [{
  60                     "cpdId": "ext_cp",
  61                     "cpConfig": [{
  62                         "cpInstanceId": "",
  63                         "cpProtocolData": [{
  64                             "layerProtocol": "IP_OVER_ETHERNET",
  65                             "ipOverEthernet": {
  66                                 "ipAddresses": [{
  67                                     "type": "IPV4",
  68                                     "numDynamicAddresses": 0,
  69                                     "subnetId": "59e9ffa9-b67e-4c05-b191-ed179007536e"
  70                                 }]
  71                             }
  72                         }]
  73                     }]
  74                 }],
  75                 "extLinkPorts": []
  76             }],
  77             "vimConnectionInfo": [{
  78                 "id": "tecs_RegionOne",
  79                 "vimType": "openstack",
  80                 "vimId": "tecs_RegionOne",
  81                 "accessInfo": {
  82                     "tenant": "chinamobile"
  83                 }
  84             }],
  85             "additionalParams": {
  86                 "vmid": "552ea058-6441-4de5-b4c1-b0a52c7557e8"
  87             }
  88         }
  89
  90     def tearDown(self):
  91         NfInstModel.objects.filter(nfinstid='12345').delete()
  92         NfInstModel.objects.filter(nfinstid='123').delete()
  93
  94     def assert_job_result(self, job_id, job_progress, job_detail):
  95         jobs = JobStatusModel.objects.filter(
  96             jobid=job_id,
  97             progress=job_progress,
  98             descp=job_detail
  99         )
 100         self.assertEqual(1, len(jobs))
 101
 102     def test_change_ext_conn_not_found(self):
 103         url = "/api/vnflcm/v1/vnf_instances/12/change_ext_conn"
 104         response = self.client.post(url,
 105                                     data=self.req_data,
 106                                     format='json')
 107         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 108
 109     def test_change_ext_conn_conflict(self):
 110         url = "/api/vnflcm/v1/vnf_instances/12345/change_ext_conn"
 111         response = self.client.post(url,
 112                                     data=self.req_data,
 113                                     format='json')
 114         self.assertEqual(status.HTTP_409_CONFLICT, response.status_code)
 115
 116     def test_change_ext_conn_badreq(self):
 117         url = "/api/vnflcm/v1/vnf_instances/123/change_ext_conn"
 118         response = self.client.post(url,
 119                                     data={},
 120                                     format='json')
 121         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 122
 123     @mock.patch.object(JobUtil, 'create_job')
 124     def test_change_ext_conn_inner_error(self, mock_run):
 125         mock_run.return_value = NFLCMException('Boom!')
 126         url = "/api/vnflcm/v1/vnf_instances/123/change_ext_conn"
 127         response = self.client.post(url,
 128                                     data=self.req_data,
 129                                     format='json')
 130         self.assertEqual(
 131             status.HTTP_500_INTERNAL_SERVER_ERROR,
 132             response.status_code)
 133
 134     @mock.patch.object(restcall, 'call_req')
 135     @mock.patch.object(api, 'call')
 136     def test_change_ext_conn_sucess(self, mock_call, mock_call_req):
 137         self.nf_inst_id = '12345'
 138         res_cache = {"volume": {}, "flavor": {}, "port": {}}
 139         res_cache["port"]["ext_cp"] = "port1"
 140         NfInstModel(nfinstid=self.nf_inst_id,
 141                     nf_name='VNF1',
 142                     nf_desc="VNF DESC",
 143                     vnfdid="1",
 144                     netype="XGW",
 145                     vendor="ZTE",
 146                     vnfSoftwareVersion="V1",
 147                     version="V1",
 148                     package_id="2",
 149                     status='INSTANTIATED',
 150                     vnfd_model=json.dumps(const.vnfd_for_scale),
 151                     vimInfo=json.dumps({}),
 152                     resInfo=json.dumps(res_cache)).save()
 153         StorageInstModel.objects.create(
 154             storageid="1",
 155             vimid="1",
 156             resourceid="11",
 157             insttype=0,
 158             instid=self.nf_inst_id,
 159             is_predefined=1
 160         )
 161         NetworkInstModel.objects.create(
 162             networkid='1',
 163             vimid='1',
 164             resourceid='1',
 165             name='pnet_network',
 166             is_predefined=1,
 167             tenant='admin',
 168             insttype=0,
 169             instid=self.nf_inst_id
 170         )
 171         SubNetworkInstModel.objects.create(
 172             subnetworkid='1',
 173             vimid='1',
 174             resourceid='1',
 175             networkid='1',
 176             is_predefined=1,
 177             name='sub_pnet',
 178             tenant='admin',
 179             insttype=0,
 180             instid=self.nf_inst_id
 181         )
 182         PortInstModel.objects.create(
 183             portid='1',
 184             networkid='1',
 185             subnetworkid='1',
 186             vimid='1',
 187             resourceid='1',
 188             is_predefined=1,
 189             name='ext_cp',
 190             tenant='admin',
 191             insttype=0,
 192             instid=self.nf_inst_id
 193         )
 194         FlavourInstModel.objects.create(
 195             flavourid="1",
 196             vimid="1",
 197             resourceid="11",
 198             instid=self.nf_inst_id,
 199             is_predefined=1,
 200             name="Flavor_sunshine"
 201         )
 202         VmInstModel.objects.create(
 203             vmid="1",
 204             vimid="1",
 205             resourceid="11",
 206             insttype=0,
 207             instid=self.nf_inst_id,
 208             vmname="test_01",
 209             is_predefined=1,
 210             operationalstate=1
 211         )
 212         VmInstModel.objects.create(
 213             vmid="2",
 214             vimid="1",
 215             resourceid="22",
 216             insttype=0,
 217             instid=self.nf_inst_id,
 218             vmname="test_02",
 219             is_predefined=1,
 220             operationalstate=1
 221         )
 222         VNFCInstModel.objects.create(
 223             vnfcinstanceid="1",
 224             instid=self.nf_inst_id,
 225             vmid="1"
 226         )
 227         VNFCInstModel.objects.create(
 228             vnfcinstanceid="2",
 229             instid=self.nf_inst_id,
 230             vmid="2"
 231         )
 232         r1_apply_grant_result = [
 233             0,
 234             json.JSONEncoder().encode(const.instantiate_grant_result),
 235             '200'
 236         ]
 237         mock_call_req.side_effect = [
 238             r1_apply_grant_result,
 239         ]
 240         mock_call.side_effect = [
 241             const.c1_data_get_tenant_id,
 242             const.c7_data_create_flavor,
 243             const.c6_data_create_port
 244         ]
 245         self.job_id = JobUtil.create_job('NF', 'VNF_CHANGE_EXT_CONN', self.nf_inst_id)
 246         JobUtil.add_job_status(self.job_id, 0, "VNF_'VNF_CHANGE_EXT_CONN'_READY")
 247
 248         ChangeExtConn(self.req_data, self.nf_inst_id, self.job_id,).run()
 249
 250         print([{job.progress: job.descp} for job in JobStatusModel.objects.filter(jobid=self.job_id)])
 251         self.assert_job_result(
 252             self.job_id,
 253             100,
 254             'Change ext conn success.'
 255         )