lcm/lcm/nf/tests/test_operate_vnf.py

   1 # Copyright (C) 2018 Verizon. All Rights Reserved.\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, Client\r
  20 from rest_framework import status\r
  21 \r
  22 from lcm.nf.biz.operate_vnf import OperateVnf\r
  23 from lcm.pub.database.models import NfInstModel\r
  24 from lcm.pub.database.models import JobStatusModel\r
  25 from lcm.pub.database.models import VmInstModel\r
  26 from lcm.pub.database.models import SubscriptionModel\r
  27 from lcm.pub.utils import restcall\r
  28 from lcm.pub.utils.jobutil import JobUtil\r
  29 from lcm.pub.utils.notificationsutil import NotificationsUtil\r
  30 from lcm.pub.utils.timeutil import now_time\r
  31 from lcm.pub.vimapi import api\r
  32 \r
  33 \r
  34 class TestNFOperate(TestCase):\r
  35     def setUp(self):\r
  36         self.client = Client()\r
  37 \r
  38     def tearDown(self):\r
  39         VmInstModel.objects.all().delete()\r
  40 \r
  41     def assert_job_result(self, job_id, job_progress, job_detail):\r
  42         jobs = JobStatusModel.objects.filter(\r
  43             jobid=job_id,\r
  44             progress=job_progress,\r
  45             descp=job_detail\r
  46         )\r
  47         self.assertEqual(1, len(jobs))\r
  48 \r
  49     def test_operate_vnf_not_found(self):\r
  50         req_data = {\r
  51             "changeStateTo": "STARTED"\r
  52         }\r
  53         response = self.client.post(\r
  54             "/api/vnflcm/v1/vnf_instances/12/operate",\r
  55             data=req_data,\r
  56             format='json'\r
  57         )\r
  58         self.failUnlessEqual(status.HTTP_404_NOT_FOUND, response.status_code)\r
  59 \r
  60     def test_operate_vnf_conflict(self):\r
  61         req_data = {\r
  62             "changeStateTo": "STARTED"\r
  63         }\r
  64         NfInstModel(\r
  65             nfinstid='12',\r
  66             nf_name='VNF1',\r
  67             status='NOT_INSTANTIATED'\r
  68         ).save()\r
  69         response = self.client.post(\r
  70             "/api/vnflcm/v1/vnf_instances/12/operate",\r
  71             data=req_data,\r
  72             format='json'\r
  73         )\r
  74         self.failUnlessEqual(status.HTTP_409_CONFLICT, response.status_code)\r
  75         NfInstModel(\r
  76             nfinstid='12',\r
  77             nf_name='VNF1',\r
  78             status='NOT_INSTANTIATED'\r
  79         ).delete()\r
  80 \r
  81     def test_operate_vnf_inner_error(self):\r
  82         NfInstModel(\r
  83             nfinstid='345',\r
  84             nf_name='VNF1',\r
  85             nf_desc="VNF DESC",\r
  86             vnfdid="1",\r
  87             netype="XGW",\r
  88             vendor="ZTE",\r
  89             vnfSoftwareVersion="V1",\r
  90             version="V1",\r
  91             package_id="2",\r
  92             status='INSTANTIATED'\r
  93         ).save()\r
  94         url = "/api/vnflcm/v1/vnf_instances/345/operate"\r
  95         response = self.client.post(\r
  96             url,\r
  97             data={},\r
  98             format='json'\r
  99         )\r
 100         NfInstModel.objects.filter(nfinstid='345').delete()\r
 101         self.failUnlessEqual(\r
 102             status.HTTP_500_INTERNAL_SERVER_ERROR,\r
 103             response.status_code\r
 104         )\r
 105 \r
 106     @mock.patch.object(OperateVnf, 'run')\r
 107     def test_operate_vnf_success(self, mock_run):\r
 108         req_data = {\r
 109             "changeStateTo": "STARTED"\r
 110         }\r
 111         NfInstModel(\r
 112             nfinstid='12',\r
 113             nf_name='VNF1',\r
 114             status='INSTANTIATED'\r
 115         ).save()\r
 116         response = self.client.post(\r
 117             "/api/vnflcm/v1/vnf_instances/12/operate",\r
 118             data=req_data,\r
 119             format='json'\r
 120         )\r
 121         mock_run.re.return_value = None\r
 122         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)\r
 123         NfInstModel(\r
 124             nfinstid='12',\r
 125             nf_name='VNF1',\r
 126             status='INSTANTIATED'\r
 127         ).delete()\r
 128 \r
 129     @mock.patch.object(restcall, 'call_req')\r
 130     @mock.patch.object(api, 'call')\r
 131     @mock.patch.object(NotificationsUtil, 'post_notification')\r
 132     def test_operate_vnf_success_start(self, mock_post_notification, mock_call, mock_call_req):\r
 133         NfInstModel.objects.create(\r
 134             nfinstid='1111',\r
 135             nf_name='2222',\r
 136             vnfminstid='1',\r
 137             package_id='todo',\r
 138             version='',\r
 139             vendor='',\r
 140             netype='',\r
 141             vnfd_model='',\r
 142             status='INSTANTIATED',\r
 143             nf_desc='',\r
 144             vnfdid='',\r
 145             vnfSoftwareVersion='',\r
 146             vnfConfigurableProperties='todo',\r
 147             localizationLanguage='EN_US',\r
 148             create_time=now_time()\r
 149         )\r
 150 \r
 151         VmInstModel.objects.create(\r
 152             vmid="1",\r
 153             vimid="1",\r
 154             resourceid="11",\r
 155             insttype=0,\r
 156             instid="1111",\r
 157             vmname="test_01",\r
 158             is_predefined=1,\r
 159             operationalstate=1\r
 160         )\r
 161 \r
 162         SubscriptionModel.objects.create(\r
 163             subscription_id=str(uuid.uuid4()),\r
 164             callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',\r
 165             auth_info=json.JSONEncoder().encode({\r
 166                 'authType': ['BASIC'],\r
 167                 'paramsBasic': {\r
 168                     'userName': 'username',\r
 169                     'password': 'password'\r
 170                 }\r
 171             }),\r
 172             notification_types=str([\r
 173                 'VnfLcmOperationOccurrenceNotification',\r
 174                 'VnfIdentifierCreationNotification',\r
 175                 'VnfIdentifierDeletionNotification'\r
 176             ]),\r
 177             operation_types=str(['OPERATE']),\r
 178             operation_states=str(['COMPLETED']),\r
 179             vnf_instance_filter=json.JSONEncoder().encode({\r
 180                 'vnfdIds': [],\r
 181                 'vnfProductsFromProviders': [],\r
 182                 'vnfInstanceIds': ['1111'],\r
 183                 'vnfInstanceNames': [],\r
 184             })\r
 185         )\r
 186 \r
 187         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 188         t3_action_vm_start_result = [0, json.JSONEncoder().encode(''), '202']\r
 189         mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_start_result]\r
 190         mock_call.return_value = None\r
 191         mock_post_notification.return_value = None\r
 192         req_data = {\r
 193             "changeStateTo": "STARTED"\r
 194         }\r
 195         self.nf_inst_id = '1111'\r
 196         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 197         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 198         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 199         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 200 \r
 201     @mock.patch.object(restcall, 'call_req')\r
 202     @mock.patch.object(api, 'call')\r
 203     @mock.patch.object(NotificationsUtil, 'post_notification')\r
 204     def test_operate_vnf_success_stop(self, mock_post_notification, mock_call, mock_call_req):\r
 205         NfInstModel.objects.create(\r
 206             nfinstid='1111',\r
 207             nf_name='2222',\r
 208             vnfminstid='1',\r
 209             package_id='todo',\r
 210             version='',\r
 211             vendor='',\r
 212             netype='',\r
 213             vnfd_model='',\r
 214             status='INSTANTIATED',\r
 215             nf_desc='',\r
 216             vnfdid='',\r
 217             vnfSoftwareVersion='',\r
 218             vnfConfigurableProperties='todo',\r
 219             localizationLanguage='EN_US',\r
 220             create_time=now_time()\r
 221         )\r
 222 \r
 223         VmInstModel.objects.create(\r
 224             vmid="1",\r
 225             vimid="1",\r
 226             resourceid="11",\r
 227             insttype=0,\r
 228             instid="1111",\r
 229             vmname="test_01",\r
 230             is_predefined=1,\r
 231             operationalstate=1\r
 232         )\r
 233 \r
 234         SubscriptionModel.objects.create(\r
 235             subscription_id=str(uuid.uuid4()),\r
 236             callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',\r
 237             auth_info=json.JSONEncoder().encode({\r
 238                 'authType': ['BASIC'],\r
 239                 'paramsBasic': {\r
 240                     'userName': 'username',\r
 241                     'password': 'password'\r
 242                 }\r
 243             }),\r
 244             notification_types=str([\r
 245                 'VnfLcmOperationOccurrenceNotification',\r
 246                 'VnfIdentifierCreationNotification',\r
 247                 'VnfIdentifierDeletionNotification'\r
 248             ]),\r
 249             operation_types=str(['OPERATE']),\r
 250             operation_states=str(['COMPLETED']),\r
 251             vnf_instance_filter=json.JSONEncoder().encode({\r
 252                 'vnfdIds': [],\r
 253                 'vnfProductsFromProviders': [],\r
 254                 'vnfInstanceIds': ['1111'],\r
 255                 'vnfInstanceNames': [],\r
 256             })\r
 257         )\r
 258 \r
 259         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 260         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 261         mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_stop_result]\r
 262         mock_call.return_value = None\r
 263         mock_post_notification.return_value = None\r
 264         req_data = {\r
 265             "changeStateTo": "STOPPED"\r
 266         }\r
 267         self.nf_inst_id = '1111'\r
 268         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 269         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 270         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 271         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 272 \r
 273     @mock.patch.object(restcall, 'call_req')\r
 274     @mock.patch.object(api, 'call')\r
 275     def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req):\r
 276         NfInstModel.objects.create(\r
 277             nfinstid='1111',\r
 278             nf_name='2222',\r
 279             vnfminstid='1',\r
 280             package_id='todo',\r
 281             version='',\r
 282             vendor='',\r
 283             netype='',\r
 284             vnfd_model='',\r
 285             status='INSTANTIATED',\r
 286             nf_desc='',\r
 287             vnfdid='',\r
 288             vnfSoftwareVersion='',\r
 289             vnfConfigurableProperties='todo',\r
 290             localizationLanguage='EN_US',\r
 291             create_time=now_time()\r
 292         )\r
 293 \r
 294         VmInstModel.objects.create(\r
 295             vmid="1",\r
 296             vimid="1",\r
 297             resourceid="11",\r
 298             insttype=0,\r
 299             instid="1111",\r
 300             vmname="test_01",\r
 301             is_predefined=1,\r
 302             operationalstate=1\r
 303         )\r
 304         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 305         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 306         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 307         mock_call_req.side_effect = [\r
 308             t1_apply_grant_result,\r
 309             t2_lcm_notify_result,\r
 310             t3_action_vm_stop_result\r
 311         ]\r
 312         mock_call.return_value = None\r
 313         req_data = {\r
 314             "changeStateTo": "STOPPED",\r
 315             "stopType": "GRACEFUL",\r
 316             "gracefulStopTimeout": 2\r
 317         }\r
 318         self.nf_inst_id = '1111'\r
 319         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 320         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 321         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 322         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 323 \r
 324     @mock.patch.object(restcall, 'call_req')\r
 325     @mock.patch.object(api, 'call')\r
 326     def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req):\r
 327         NfInstModel.objects.create(\r
 328             nfinstid='1111',\r
 329             nf_name='2222',\r
 330             vnfminstid='1',\r
 331             package_id='todo',\r
 332             version='',\r
 333             vendor='',\r
 334             netype='',\r
 335             vnfd_model='',\r
 336             status='INSTANTIATED',\r
 337             nf_desc='',\r
 338             vnfdid='',\r
 339             vnfSoftwareVersion='',\r
 340             vnfConfigurableProperties='todo',\r
 341             localizationLanguage='EN_US',\r
 342             create_time=now_time()\r
 343         )\r
 344 \r
 345         VmInstModel.objects.create(\r
 346             vmid="1",\r
 347             vimid="1",\r
 348             resourceid="11",\r
 349             insttype=0,\r
 350             instid="1111",\r
 351             vmname="test_01",\r
 352             is_predefined=1,\r
 353             operationalstate=1\r
 354         )\r
 355         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 356         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 357         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 358         mock_call_req.side_effect = [\r
 359             t1_apply_grant_result,\r
 360             t2_lcm_notify_result,\r
 361             t3_action_vm_stop_result\r
 362         ]\r
 363         mock_call.return_value = None\r
 364         req_data = {\r
 365             "changeStateTo": "STOPPED",\r
 366             "stopType": "FORCEFUL",\r
 367         }\r
 368         self.nf_inst_id = '1111'\r
 369         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 370         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 371         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 372         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r