lcm/lcm/nf/tests/test_instantiate_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 import json\r
  16 import uuid\r
  17 \r
  18 import mock\r
  19 from django.test import TestCase\r
  20 from rest_framework import status\r
  21 from rest_framework.test import APIClient\r
  22 \r
  23 from lcm.nf.biz.instantiate_vnf import InstantiateVnf\r
  24 from lcm.nf.const import c1_data_get_tenant_id, c4_data_create_network, c2_data_create_volume, \\r
  25     c5_data_create_subnet, c3_data_get_volume, c6_data_create_port, c7_data_create_flavor, c8_data_list_image, \\r
  26     c9_data_create_vm, c10_data_get_vm, inst_req_data, vnfpackage_info\r
  27 from lcm.pub.database.models import NfInstModel, JobStatusModel, SubscriptionModel\r
  28 from lcm.pub.utils import restcall\r
  29 from lcm.pub.utils.jobutil import JobUtil\r
  30 from lcm.pub.utils.timeutil import now_time\r
  31 from lcm.pub.utils.notificationsutil import NotificationsUtil\r
  32 from lcm.pub.vimapi import api\r
  33 \r
  34 \r
  35 class TestNFInstantiate(TestCase):\r
  36     def setUp(self):\r
  37         self.client = APIClient()\r
  38         self.grant_result = {\r
  39             "vimConnections": [\r
  40                 {\r
  41                     "vimid": 'vim_1',\r
  42                     "accessInfo":\r
  43                     {\r
  44                         "tenant": 'chinamobile'\r
  45                     }\r
  46                 },\r
  47             ],\r
  48             "vnfId": "413aa1fe-b4d1-11e8-8268-dff5aab95c63",\r
  49             "vimAssets":\r
  50             {\r
  51                 "computeResourceFlavours": [\r
  52                     {\r
  53                         "resourceProviderId": "vgw",\r
  54                         "vimFlavourId": "yui",\r
  55                         "vimConnectionId": ""\r
  56                     },\r
  57                 ]\r
  58             }\r
  59         }\r
  60 \r
  61     def tearDown(self):\r
  62         NfInstModel.objects.all().delete()\r
  63 \r
  64     def assert_job_result(self, job_id, job_progress, job_detail):\r
  65         jobs = JobStatusModel.objects.filter(jobid=job_id,\r
  66                                              progress=job_progress,\r
  67                                              descp=job_detail)\r
  68         self.assertEqual(1, len(jobs))\r
  69 \r
  70     @mock.patch.object(InstantiateVnf, 'run')\r
  71     def test_instantiate_vnf(self, mock_run):\r
  72         NfInstModel(\r
  73             nfinstid='12',\r
  74             nf_name='VNF1',\r
  75             status='NOT_INSTANTIATED'\r
  76         ).save()\r
  77         mock_run.re.return_value = None\r
  78         response = self.client.post(\r
  79             '/api/vnflcm/v1/vnf_instances/12/instantiate',\r
  80             data=inst_req_data,\r
  81             format='json'\r
  82         )\r
  83         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)\r
  84 \r
  85     @mock.patch.object(InstantiateVnf, 'run')\r
  86     def test_instantiate_vnf_notfound(self, mock_run):\r
  87         mock_run.re.return_value = None\r
  88         response = self.client.post(\r
  89             '/api/vnflcm/v1/vnf_instances/3421/instantiate',\r
  90             data=inst_req_data,\r
  91             format='json'\r
  92         )\r
  93         self.failUnlessEqual(status.HTTP_404_NOT_FOUND, response.status_code, response.content)\r
  94 \r
  95     @mock.patch.object(InstantiateVnf, 'run')\r
  96     def test_instantiate_vnf_conflict(self, mock_run):\r
  97         NfInstModel(\r
  98             nfinstid='1123',\r
  99             nf_name='VNF1',\r
 100             status='INSTANTIATED'\r
 101         ).save()\r
 102         mock_run.re.return_value = None\r
 103         response = self.client.post(\r
 104             '/api/vnflcm/v1/vnf_instances/1123/instantiate',\r
 105             data=inst_req_data,\r
 106             format='json'\r
 107         )\r
 108         self.failUnlessEqual(status.HTTP_409_CONFLICT, response.status_code, response.content)\r
 109 \r
 110     def test_instantiate_vnf_when_inst_id_not_exist(self):\r
 111         self.nf_inst_id = str(uuid.uuid4())\r
 112         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 113         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 114         jobs = JobStatusModel.objects.filter(jobid=self.job_id,\r
 115                                              progress=0,\r
 116                                              descp='INST_VNF_READY')\r
 117         self.assertEqual(1, len(jobs))\r
 118         data = inst_req_data\r
 119         InstantiateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 120         self.assert_job_result(self.job_id, 255, 'VNF nf_inst_id is not exist.')\r
 121 \r
 122     def test_instantiate_vnf_when_already_instantiated(self):\r
 123         NfInstModel.objects.create(nfinstid='1111',\r
 124                                    nf_name='vFW_01',\r
 125                                    package_id='222',\r
 126                                    version='',\r
 127                                    vendor='',\r
 128                                    netype='',\r
 129                                    vnfd_model='',\r
 130                                    status='INSTANTIATED',\r
 131                                    nf_desc='vFW in Nanjing TIC Edge',\r
 132                                    vnfdid='111',\r
 133                                    create_time=now_time())\r
 134         self.nf_inst_id = '1111'\r
 135         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 136         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 137         jobs = JobStatusModel.objects.filter(jobid=self.job_id,\r
 138                                              progress=0,\r
 139                                              descp='INST_VNF_READY')\r
 140         self.assertEqual(1, len(jobs))\r
 141         data = inst_req_data\r
 142         InstantiateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 143         self.assert_job_result(self.job_id, 255, 'VNF instantiationState is not NOT_INSTANTIATED.')\r
 144 \r
 145     @mock.patch.object(restcall, 'call_req')\r
 146     def test_instantiate_vnf_when_get_packageinfo_by_csarid_failed(self, mock_call_req):\r
 147         NfInstModel.objects.create(nfinstid='1111',\r
 148                                    nf_name='vFW_01',\r
 149                                    package_id='222',\r
 150                                    version='',\r
 151                                    vendor='',\r
 152                                    netype='',\r
 153                                    vnfd_model='',\r
 154                                    status='NOT_INSTANTIATED',\r
 155                                    nf_desc='vFW in Nanjing TIC Edge',\r
 156                                    vnfdid='111',\r
 157                                    create_time=now_time())\r
 158         r1_get_vnfpackage_by_vnfdid = [1, json.JSONEncoder().encode(vnfpackage_info), '200']\r
 159         mock_call_req.side_effect = [r1_get_vnfpackage_by_vnfdid]\r
 160         self.nf_inst_id = '1111'\r
 161         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 162         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 163         data = inst_req_data\r
 164         InstantiateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 165         self.assert_job_result(self.job_id, 255, 'Failed to query vnf CSAR(111) from catalog.')\r
 166 \r
 167     @mock.patch.object(restcall, 'call_req')\r
 168     def test_instantiate_vnf_when_applay_grant_failed(self, mock_call_req):\r
 169         NfInstModel.objects.create(nfinstid='1111',\r
 170                                    nf_name='vFW_01',\r
 171                                    package_id='222',\r
 172                                    version='',\r
 173                                    vendor='',\r
 174                                    netype='',\r
 175                                    vnfd_model='',\r
 176                                    status='NOT_INSTANTIATED',\r
 177                                    nf_desc='vFW in Nanjing TIC Edge',\r
 178                                    vnfdid='111',\r
 179                                    create_time=now_time())\r
 180         r1_get_vnfpackage_by_vnfdid = [0, json.JSONEncoder().encode(vnfpackage_info), '200']\r
 181         r2_apply_grant_result = [1, json.JSONEncoder().encode(self.grant_result), '200']\r
 182         mock_call_req.side_effect = [r1_get_vnfpackage_by_vnfdid, r2_apply_grant_result]\r
 183         self.nf_inst_id = '1111'\r
 184         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 185         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 186         data = inst_req_data\r
 187         InstantiateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 188         self.assert_job_result(self.job_id, 255, 'Nf instancing apply grant exception')\r
 189 \r
 190     @mock.patch.object(restcall, 'call_req')\r
 191     @mock.patch.object(api, 'call')\r
 192     def test_instantiate_vnf_when_unexpected_exception(self, mock_call, mock_call_req):\r
 193         NfInstModel.objects.create(nfinstid='1111',\r
 194                                    nf_name='vFW_01',\r
 195                                    package_id='222',\r
 196                                    version='',\r
 197                                    vendor='',\r
 198                                    netype='',\r
 199                                    vnfd_model='',\r
 200                                    status='NOT_INSTANTIATED',\r
 201                                    nf_desc='vFW in Nanjing TIC Edge',\r
 202                                    vnfdid='111',\r
 203                                    create_time=now_time())\r
 204         r1_get_vnfpackage_by_vnfdid = [0, json.JSONEncoder().encode(vnfpackage_info), '200']\r
 205         r2_apply_grant_result = [0, json.JSONEncoder().encode(self.grant_result), '200']\r
 206         mock_call_req.side_effect = [r1_get_vnfpackage_by_vnfdid, r2_apply_grant_result]\r
 207         mock_call.side_effect = [c1_data_get_tenant_id, c2_data_create_volume, c3_data_get_volume]\r
 208         self.nf_inst_id = '1111'\r
 209         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 210         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 211         data = inst_req_data\r
 212         InstantiateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 213         self.assert_job_result(self.job_id, 255, 'unexpected exception')\r
 214 \r
 215     @mock.patch.object(restcall, 'call_req')\r
 216     @mock.patch.object(api, 'call')\r
 217     @mock.patch.object(NotificationsUtil, 'post_notification')\r
 218     def test_instantiate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):\r
 219         NfInstModel.objects.create(nfinstid='1111',\r
 220                                    nf_name='vFW_01',\r
 221                                    package_id='222',\r
 222                                    version='',\r
 223                                    vendor='',\r
 224                                    netype='',\r
 225                                    vnfd_model='',\r
 226                                    status='NOT_INSTANTIATED',\r
 227                                    nf_desc='vFW in Nanjing TIC Edge',\r
 228                                    vnfdid='111',\r
 229                                    create_time=now_time())\r
 230         SubscriptionModel.objects.create(\r
 231             subscription_id=str(uuid.uuid4()),\r
 232             callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',\r
 233             auth_info=json.JSONEncoder().encode({\r
 234                 'authType': ['BASIC'],\r
 235                 'paramsBasic': {\r
 236                     'userName': 'username',\r
 237                     'password': 'password'\r
 238                 }\r
 239             }),\r
 240             notification_types=str([\r
 241                 'VnfLcmOperationOccurrenceNotification',\r
 242                 'VnfIdentifierCreationNotification',\r
 243                 'VnfIdentifierDeletionNotification'\r
 244             ]),\r
 245             operation_types=str(['INSTANTIATE']),\r
 246             operation_states=str(['COMPLETED']),\r
 247             vnf_instance_filter=json.JSONEncoder().encode({\r
 248                 'vnfdIds': ['111'],\r
 249                 'vnfProductsFromProviders': [],\r
 250                 'vnfInstanceIds': ['1111'],\r
 251                 'vnfInstanceNames': [],\r
 252             })\r
 253         )\r
 254         r1_get_vnfpackage_by_vnfdid = [0, json.JSONEncoder().encode(vnfpackage_info), '200']\r
 255         r2_apply_grant_result = [0, json.JSONEncoder().encode(self.grant_result), '200']\r
 256         r3_all_aai_result = [1, json.JSONEncoder().encode(''), '404']\r
 257         r4_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 258         mock_call_req.side_effect = [r1_get_vnfpackage_by_vnfdid, r2_apply_grant_result, r3_all_aai_result, r4_lcm_notify_result]\r
 259         mock_call.side_effect = [c1_data_get_tenant_id,\r
 260                                  c2_data_create_volume, c3_data_get_volume,\r
 261                                  c4_data_create_network,\r
 262                                  c5_data_create_subnet,\r
 263                                  c6_data_create_port,\r
 264                                  c7_data_create_flavor,\r
 265                                  c8_data_list_image, c9_data_create_vm, c10_data_get_vm]\r
 266         mock_post_notification.side_effect = None\r
 267         self.nf_inst_id = '1111'\r
 268         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 269         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 270         data = inst_req_data\r
 271         InstantiateVnf(data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 272         # self.assert_job_result(self.job_id, 100, 'Instantiate Vnf success.')\r