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