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         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 140 \r
 141     @mock.patch.object(restcall, 'call_req')\r
 142     @mock.patch.object(api, 'call')\r
 143     @mock.patch.object(NotificationsUtil, 'post_notification')\r
 144     def test_operate_vnf_success_stop(self, mock_post_notification, mock_call, mock_call_req):\r
 145         NfInstModel.objects.create(nfinstid='1111',\r
 146                                    nf_name='2222',\r
 147                                    vnfminstid='1',\r
 148                                    package_id='todo',\r
 149                                    version='',\r
 150                                    vendor='',\r
 151                                    netype='',\r
 152                                    vnfd_model='',\r
 153                                    status='INSTANTIATED',\r
 154                                    nf_desc='',\r
 155                                    vnfdid='',\r
 156                                    vnfSoftwareVersion='',\r
 157                                    vnfConfigurableProperties='todo',\r
 158                                    localizationLanguage='EN_US',\r
 159                                    create_time=now_time())\r
 160 \r
 161         VmInstModel.objects.create(vmid="1",\r
 162                                    vimid="1",\r
 163                                    resourceid="11",\r
 164                                    insttype=0,\r
 165                                    instid="1111",\r
 166                                    vmname="test_01",\r
 167                                    is_predefined=1,\r
 168                                    operationalstate=1)\r
 169 \r
 170         SubscriptionModel.objects.create(\r
 171             subscription_id=str(uuid.uuid4()),\r
 172             callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification',\r
 173             auth_info=json.JSONEncoder().encode({\r
 174                 'authType': ['BASIC'],\r
 175                 'paramsBasic': {\r
 176                     'userName': 'username',\r
 177                     'password': 'password'\r
 178                 }\r
 179             }),\r
 180             notification_types=str([\r
 181                 'VnfLcmOperationOccurrenceNotification',\r
 182                 'VnfIdentifierCreationNotification',\r
 183                 'VnfIdentifierDeletionNotification'\r
 184             ]),\r
 185             operation_types=str(['OPERATE']),\r
 186             operation_states=str(['COMPLETED']),\r
 187             vnf_instance_filter=json.JSONEncoder().encode({\r
 188                 'vnfdIds': [],\r
 189                 'vnfProductsFromProviders': [],\r
 190                 'vnfInstanceIds': ['1111'],\r
 191                 'vnfInstanceNames': [],\r
 192             })\r
 193         )\r
 194 \r
 195         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 196         # t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 197         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 198         # mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]\r
 199         mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_stop_result]\r
 200         mock_call.return_value = None\r
 201         mock_post_notification.return_value = None\r
 202         req_data = {\r
 203             "changeStateTo": "STOPPED"\r
 204         }\r
 205         self.nf_inst_id = '1111'\r
 206         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 207         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 208         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 209         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 210 \r
 211     @mock.patch.object(restcall, 'call_req')\r
 212     @mock.patch.object(api, 'call')\r
 213     def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req):\r
 214         NfInstModel.objects.create(nfinstid='1111',\r
 215                                    nf_name='2222',\r
 216                                    vnfminstid='1',\r
 217                                    package_id='todo',\r
 218                                    version='',\r
 219                                    vendor='',\r
 220                                    netype='',\r
 221                                    vnfd_model='',\r
 222                                    status='INSTANTIATED',\r
 223                                    nf_desc='',\r
 224                                    vnfdid='',\r
 225                                    vnfSoftwareVersion='',\r
 226                                    vnfConfigurableProperties='todo',\r
 227                                    localizationLanguage='EN_US',\r
 228                                    create_time=now_time())\r
 229 \r
 230         VmInstModel.objects.create(vmid="1",\r
 231                                    vimid="1",\r
 232                                    resourceid="11",\r
 233                                    insttype=0,\r
 234                                    instid="1111",\r
 235                                    vmname="test_01",\r
 236                                    is_predefined=1,\r
 237                                    operationalstate=1)\r
 238         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 239         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 240         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 241         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]\r
 242         mock_call.return_value = None\r
 243         req_data = {\r
 244             "changeStateTo": "STOPPED",\r
 245             "stopType": "GRACEFUL",\r
 246             "gracefulStopTimeout": 2\r
 247         }\r
 248         self.nf_inst_id = '1111'\r
 249         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 250         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 251         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 252         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 253 \r
 254     @mock.patch.object(restcall, 'call_req')\r
 255     @mock.patch.object(api, 'call')\r
 256     def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req):\r
 257         NfInstModel.objects.create(nfinstid='1111',\r
 258                                    nf_name='2222',\r
 259                                    vnfminstid='1',\r
 260                                    package_id='todo',\r
 261                                    version='',\r
 262                                    vendor='',\r
 263                                    netype='',\r
 264                                    vnfd_model='',\r
 265                                    status='INSTANTIATED',\r
 266                                    nf_desc='',\r
 267                                    vnfdid='',\r
 268                                    vnfSoftwareVersion='',\r
 269                                    vnfConfigurableProperties='todo',\r
 270                                    localizationLanguage='EN_US',\r
 271                                    create_time=now_time())\r
 272 \r
 273         VmInstModel.objects.create(vmid="1",\r
 274                                    vimid="1",\r
 275                                    resourceid="11",\r
 276                                    insttype=0,\r
 277                                    instid="1111",\r
 278                                    vmname="test_01",\r
 279                                    is_predefined=1,\r
 280                                    operationalstate=1)\r
 281         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 282         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 283         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 284         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]\r
 285         mock_call.return_value = None\r
 286         req_data = {\r
 287             "changeStateTo": "STOPPED",\r
 288             "stopType": "FORCEFUL",\r
 289         }\r
 290         self.nf_inst_id = '1111'\r
 291         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 292         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 293         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 294         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r