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 import mock\r
  18 \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 .const import c1_data_get_tenant_id\r
  24 from .const import c4_data_create_network\r
  25 from .const import c2_data_create_volume\r
  26 from .const import c5_data_create_subnet\r
  27 from .const import c3_data_get_volume\r
  28 from .const import c6_data_create_port\r
  29 from .const import c7_data_create_flavor\r
  30 from .const import c8_data_list_image\r
  31 from .const import c9_data_create_vm\r
  32 from .const import c10_data_get_vm\r
  33 from .const import inst_req_data\r
  34 from .const import vnfpackage_info\r
  35 from .const import instantiate_grant_result\r
  36 \r
  37 from lcm.pub.database.models import NfInstModel, VNFLcmOpOccModel\r
  38 from lcm.pub.database.models import JobStatusModel\r
  39 from lcm.pub.database.models import SubscriptionModel\r
  40 from lcm.pub.utils import restcall\r
  41 from lcm.pub.utils.jobutil import JobUtil\r
  42 from lcm.pub.utils.timeutil import now_time\r
  43 from lcm.pub.utils.notificationsutil import NotificationsUtil\r
  44 from lcm.pub.vimapi import api\r
  45 from lcm.pub.exceptions import NFLCMException\r
  46 \r
  47 from lcm.nf.biz.instantiate_vnf import InstantiateVnf\r
  48 from lcm.nf import const\r
  49 \r
  50 \r
  51 class TestNFInstantiate(TestCase):\r
  52     def setUp(self):\r
  53         self.client = APIClient()\r
  54         self.grant_result = instantiate_grant_result\r
  55 \r
  56     def tearDown(self):\r
  57         NfInstModel.objects.all().delete()\r
  58 \r
  59     def assert_job_result(self, job_id, job_progress, job_detail):\r
  60         jobs = JobStatusModel.objects.filter(\r
  61             jobid=job_id,\r
  62             progress=job_progress,\r
  63             descp=job_detail\r
  64         )\r
  65         self.assertEqual(1, len(jobs))\r
  66 \r
  67     @mock.patch.object(InstantiateVnf, 'run')\r
  68     def test_instantiate_vnf(self, mock_run):\r
  69         NfInstModel(\r
  70             nfinstid='12',\r
  71             nf_name='VNF1',\r
  72             status='NOT_INSTANTIATED'\r
  73         ).save()\r
  74         mock_run.re.return_value = None\r
  75         response = self.client.post(\r
  76             '/api/vnflcm/v1/vnf_instances/12/instantiate',\r
  77             data=inst_req_data,\r
  78             format='json'\r
  79         )\r
  80         self.assertEqual(\r
  81             status.HTTP_202_ACCEPTED,\r
  82             response.status_code\r
  83         )\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.assertEqual(\r
  94             status.HTTP_404_NOT_FOUND,\r
  95             response.status_code,\r
  96             response.content\r
  97         )\r
  98 \r
  99     @mock.patch.object(InstantiateVnf, 'run')\r
 100     def test_instantiate_vnf_conflict(self, mock_run):\r
 101         NfInstModel(\r
 102             nfinstid='1123',\r
 103             nf_name='VNF1',\r
 104             status='INSTANTIATED'\r
 105         ).save()\r
 106         mock_run.re.return_value = None\r
 107         response = self.client.post(\r
 108             '/api/vnflcm/v1/vnf_instances/1123/instantiate',\r
 109             data=inst_req_data,\r
 110             format='json'\r
 111         )\r
 112         self.assertEqual(\r
 113             status.HTTP_409_CONFLICT,\r
 114             response.status_code,\r
 115             response.content\r
 116         )\r
 117 \r
 118     def test_instantiate_vnf_when_inst_id_not_exist(self):\r
 119         self.nf_inst_id = str(uuid.uuid4())\r
 120         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 121         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 122         jobs = JobStatusModel.objects.filter(\r
 123             jobid=self.job_id,\r
 124             progress=0,\r
 125             descp='INST_VNF_READY'\r
 126         )\r
 127         self.assertEqual(1, len(jobs))\r
 128         data = inst_req_data\r
 129         InstantiateVnf(\r
 130             data,\r
 131             nf_inst_id=self.nf_inst_id,\r
 132             job_id=self.job_id\r
 133         ).run()\r
 134         self.assert_job_result(\r
 135             self.job_id,\r
 136             255,\r
 137             'VNF nf_inst_id is not exist.'\r
 138         )\r
 139 \r
 140     def test_instantiate_vnf_when_already_instantiated(self):\r
 141         NfInstModel.objects.create(\r
 142             nfinstid='1111',\r
 143             nf_name='vFW_01',\r
 144             package_id='222',\r
 145             version='',\r
 146             vendor='',\r
 147             netype='',\r
 148             vnfd_model='',\r
 149             status='INSTANTIATED',\r
 150             nf_desc='vFW in Nanjing TIC Edge',\r
 151             vnfdid='111',\r
 152             create_time=now_time()\r
 153         )\r
 154         self.nf_inst_id = '1111'\r
 155         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 156         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 157         jobs = JobStatusModel.objects.filter(\r
 158             jobid=self.job_id,\r
 159             progress=0,\r
 160             descp='INST_VNF_READY'\r
 161         )\r
 162         self.assertEqual(1, len(jobs))\r
 163         data = inst_req_data\r
 164         InstantiateVnf(\r
 165             data,\r
 166             nf_inst_id=self.nf_inst_id,\r
 167             job_id=self.job_id\r
 168         ).run()\r
 169         self.assert_job_result(\r
 170             self.job_id,\r
 171             255,\r
 172             'VNF instantiationState is not NOT_INSTANTIATED.'\r
 173         )\r
 174 \r
 175     @mock.patch.object(restcall, 'call_req')\r
 176     def test_instantiate_vnf_when_get_packageinfo_by_csarid_failed(self, mock_call_req):\r
 177         NfInstModel.objects.create(\r
 178             nfinstid='1111',\r
 179             nf_name='vFW_01',\r
 180             package_id='222',\r
 181             version='',\r
 182             vendor='',\r
 183             netype='',\r
 184             vnfd_model='',\r
 185             status='NOT_INSTANTIATED',\r
 186             nf_desc='vFW in Nanjing TIC Edge',\r
 187             vnfdid='111',\r
 188             create_time=now_time()\r
 189         )\r
 190         r1_get_vnfpackage_by_vnfdid = [\r
 191             1,\r
 192             json.JSONEncoder().encode(vnfpackage_info),\r
 193             '200'\r
 194         ]\r
 195         mock_call_req.side_effect = [r1_get_vnfpackage_by_vnfdid]\r
 196         self.nf_inst_id = '1111'\r
 197         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 198         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 199         data = inst_req_data\r
 200         InstantiateVnf(\r
 201             data,\r
 202             nf_inst_id=self.nf_inst_id,\r
 203             job_id=self.job_id\r
 204         ).run()\r
 205         self.assert_job_result(\r
 206             self.job_id,\r
 207             255,\r
 208             'Failed to query vnf CSAR(111) from catalog.'\r
 209         )\r
 210 \r
 211     @mock.patch.object(restcall, 'call_req')\r
 212     def test_instantiate_vnf_when_applay_grant_failed(self, mock_call_req):\r
 213         NfInstModel.objects.create(\r
 214             nfinstid='1111',\r
 215             nf_name='vFW_01',\r
 216             package_id='222',\r
 217             version='',\r
 218             vendor='',\r
 219             netype='',\r
 220             vnfd_model='',\r
 221             status='NOT_INSTANTIATED',\r
 222             nf_desc='vFW in Nanjing TIC Edge',\r
 223             vnfdid='111',\r
 224             create_time=now_time()\r
 225         )\r
 226         r1_get_vnfpackage_by_vnfdid = [\r
 227             0,\r
 228             json.JSONEncoder().encode(vnfpackage_info),\r
 229             '200'\r
 230         ]\r
 231         r2_apply_grant_result = [\r
 232             1,\r
 233             json.JSONEncoder().encode(self.grant_result),\r
 234             '200'\r
 235         ]\r
 236         mock_call_req.side_effect = [\r
 237             r1_get_vnfpackage_by_vnfdid,\r
 238             r2_apply_grant_result\r
 239         ]\r
 240         self.nf_inst_id = '1111'\r
 241         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 242         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 243         data = inst_req_data\r
 244         InstantiateVnf(\r
 245             data,\r
 246             nf_inst_id=self.nf_inst_id,\r
 247             job_id=self.job_id\r
 248         ).run()\r
 249         self.assert_job_result(\r
 250             self.job_id,\r
 251             255,\r
 252             'Nf instancing apply grant exception'\r
 253         )\r
 254 \r
 255     @mock.patch.object(restcall, 'call_req')\r
 256     @mock.patch.object(api, 'call')\r
 257     def test_instantiate_vnf_when_unexpected_exception(self, mock_call, mock_call_req):\r
 258         NfInstModel.objects.create(\r
 259             nfinstid='1111',\r
 260             nf_name='vFW_01',\r
 261             package_id='222',\r
 262             version='',\r
 263             vendor='',\r
 264             netype='',\r
 265             vnfd_model='',\r
 266             status='NOT_INSTANTIATED',\r
 267             nf_desc='vFW in Nanjing TIC Edge',\r
 268             vnfdid='111',\r
 269             create_time=now_time()\r
 270         )\r
 271         r1_get_vnfpackage_by_vnfdid = [\r
 272             0,\r
 273             json.JSONEncoder().encode(vnfpackage_info),\r
 274             '200'\r
 275         ]\r
 276         r2_apply_grant_result = [\r
 277             0,\r
 278             json.JSONEncoder().encode(self.grant_result),\r
 279             '200'\r
 280         ]\r
 281         mock_call_req.side_effect = [\r
 282             r1_get_vnfpackage_by_vnfdid,\r
 283             r2_apply_grant_result\r
 284         ]\r
 285         mock_call.side_effect = [\r
 286             c1_data_get_tenant_id,\r
 287             c2_data_create_volume,\r
 288             c3_data_get_volume\r
 289         ]\r
 290         self.nf_inst_id = '1111'\r
 291         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 292         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 293         data = inst_req_data\r
 294         InstantiateVnf(\r
 295             data,\r
 296             nf_inst_id=self.nf_inst_id,\r
 297             job_id=self.job_id\r
 298         ).run()\r
 299         self.assert_job_result(\r
 300             self.job_id,\r
 301             255,\r
 302             'unexpected exception'\r
 303         )\r
 304 \r
 305     @mock.patch.object(restcall, 'call_req')\r
 306     @mock.patch.object(api, 'call')\r
 307     @mock.patch.object(NotificationsUtil, 'post_notification')\r
 308     def test_instantiate_vnf_success(self, mock_post_notification, mock_call, mock_call_req):\r
 309         NfInstModel.objects.create(\r
 310             nfinstid='1111',\r
 311             nf_name='vFW_01',\r
 312             package_id='222',\r
 313             version='',\r
 314             vendor='',\r
 315             netype='',\r
 316             vnfd_model='',\r
 317             status='NOT_INSTANTIATED',\r
 318             nf_desc='vFW in Nanjing TIC Edge',\r
 319             vnfdid='111',\r
 320             create_time=now_time()\r
 321         )\r
 322         SubscriptionModel.objects.create(\r
 323             subscription_id=str(uuid.uuid4()),\r
 324             callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',\r
 325             auth_info=json.JSONEncoder().encode({\r
 326                 'authType': ['BASIC'],\r
 327                 'paramsBasic': {\r
 328                     'userName': 'username',\r
 329                     'password': 'password'\r
 330                 }\r
 331             }),\r
 332             notification_types=str([\r
 333                 'VnfLcmOperationOccurrenceNotification',\r
 334                 'VnfIdentifierCreationNotification',\r
 335                 'VnfIdentifierDeletionNotification'\r
 336             ]),\r
 337             operation_types=str(['INSTANTIATE']),\r
 338             operation_states=str(['COMPLETED']),\r
 339             vnf_instance_filter=json.JSONEncoder().encode({\r
 340                 'vnfdIds': ['111'],\r
 341                 'vnfProductsFromProviders': [],\r
 342                 'vnfInstanceIds': ['1111'],\r
 343                 'vnfInstanceNames': [],\r
 344             })\r
 345         )\r
 346         r1_get_vnfpackage_by_vnfdid = [\r
 347             0,\r
 348             json.JSONEncoder().encode(vnfpackage_info),\r
 349             '200'\r
 350         ]\r
 351         r2_apply_grant_result = [\r
 352             0,\r
 353             json.JSONEncoder().encode(self.grant_result),\r
 354             '200'\r
 355         ]\r
 356         r3_all_aai_result = [\r
 357             1,\r
 358             json.JSONEncoder().encode(''),\r
 359             '404'\r
 360         ]\r
 361         r4_lcm_notify_result = [\r
 362             0,\r
 363             json.JSONEncoder().encode(''),\r
 364             '200'\r
 365         ]\r
 366         mock_call_req.side_effect = [\r
 367             r1_get_vnfpackage_by_vnfdid,\r
 368             r2_apply_grant_result,\r
 369             r3_all_aai_result,\r
 370             r4_lcm_notify_result\r
 371         ]\r
 372         mock_call.side_effect = [\r
 373             c1_data_get_tenant_id,\r
 374             c2_data_create_volume,\r
 375             c3_data_get_volume,\r
 376             c4_data_create_network,\r
 377             c5_data_create_subnet,\r
 378             c6_data_create_port,\r
 379             c7_data_create_flavor,\r
 380             c8_data_list_image,\r
 381             c9_data_create_vm,\r
 382             c10_data_get_vm\r
 383         ]\r
 384         mock_post_notification.side_effect = None\r
 385         self.nf_inst_id = '1111'\r
 386         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 387         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 388         data = inst_req_data\r
 389         InstantiateVnf(\r
 390             data,\r
 391             nf_inst_id=self.nf_inst_id,\r
 392             job_id=self.job_id\r
 393         ).run()\r
 394         self.assert_job_result(\r
 395             self.job_id,\r
 396             100,\r
 397             'Instantiate Vnf success.'\r
 398         )\r
 399 \r
 400     @mock.patch.object(JobUtil, 'create_job')\r
 401     def test_instantiate_inner_error(self, mock_run):\r
 402         NfInstModel(\r
 403             nfinstid='144',\r
 404             nf_name='VNF1',\r
 405             status='NOT_INSTANTIATED'\r
 406         ).save()\r
 407         mock_run.return_value = NFLCMException('Boom!')\r
 408         response = self.client.post(\r
 409             '/api/vnflcm/v1/vnf_instances/144/instantiate',\r
 410             data=inst_req_data,\r
 411             format='json'\r
 412         )\r
 413         NfInstModel.objects.filter(nfinstid='144').delete()\r
 414         self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)\r
 415 \r
 416     def test_instantiate_badreq(self):\r
 417         NfInstModel(\r
 418             nfinstid='144',\r
 419             nf_name='VNF1',\r
 420             status='NOT_INSTANTIATED'\r
 421         ).save()\r
 422         response = self.client.post(\r
 423             '/api/vnflcm/v1/vnf_instances/144/instantiate',\r
 424             data={},\r
 425             format='json'\r
 426         )\r
 427         NfInstModel.objects.filter(nfinstid='144').delete()\r
 428         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)\r
 429 \r
 430     @mock.patch.object(restcall, 'call_req')\r
 431     def test_instantiate_operating_fail(self, mock_call_req):\r
 432         NfInstModel.objects.create(\r
 433             nfinstid='1111',\r
 434             nf_name='vFW_01',\r
 435             package_id='222',\r
 436             version='',\r
 437             vendor='',\r
 438             netype='',\r
 439             vnfd_model='',\r
 440             status='NOT_INSTANTIATED',\r
 441             nf_desc='vFW in Nanjing TIC Edge',\r
 442             vnfdid='111',\r
 443             create_time=now_time()\r
 444         )\r
 445         r1_get_vnfpackage_by_vnfdid = [\r
 446             0,\r
 447             json.JSONEncoder().encode(vnfpackage_info),\r
 448             '200'\r
 449         ]\r
 450         mock_call_req.side_effect = [\r
 451             r1_get_vnfpackage_by_vnfdid\r
 452         ]\r
 453         self.nf_inst_id = '1111'\r
 454         self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
 455         JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
 456         VNFLcmOpOccModel.objects.create(vnf_instance_id=self.nf_inst_id,\r
 457                                         id=self.job_id,\r
 458                                         operation=const.OPERATION_TYPE.INSTANTIATE,\r
 459                                         operation_state=const.OPERATION_STATE_TYPE.PROCESSING)\r
 460         InstantiateVnf(\r
 461             inst_req_data,\r
 462             nf_inst_id=self.nf_inst_id,\r
 463             job_id=self.job_id\r
 464         ).run()\r
 465         self.assert_job_result(\r
 466             self.job_id,\r
 467             255,\r
 468             'VNF(%s) %s in processing.' % (self.nf_inst_id, const.OPERATION_TYPE.INSTANTIATE)\r
 469         )\r