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