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 \r
  17 import mock\r
  18 from django.test import TestCase, Client\r
  19 from rest_framework import status\r
  20 \r
  21 from lcm.nf.biz.operate_vnf import OperateVnf\r
  22 from lcm.pub.database.models import NfInstModel, JobStatusModel, VmInstModel\r
  23 from lcm.pub.utils import restcall\r
  24 from lcm.pub.utils.jobutil import JobUtil\r
  25 from lcm.pub.utils.timeutil import now_time\r
  26 from lcm.pub.vimapi import api\r
  27 \r
  28 \r
  29 class TestNFOperate(TestCase):\r
  30     def setUp(self):\r
  31         self.client = Client()\r
  32 \r
  33     def tearDown(self):\r
  34         VmInstModel.objects.all().delete()\r
  35 \r
  36     def assert_job_result(self, job_id, job_progress, job_detail):\r
  37         jobs = JobStatusModel.objects.filter(jobid=job_id,\r
  38                                              progress=job_progress,\r
  39                                              descp=job_detail)\r
  40         self.assertEqual(1, len(jobs))\r
  41 \r
  42     def test_operate_vnf_not_found(self):\r
  43         req_data = {\r
  44             "changeStateTo": "STARTED"\r
  45         }\r
  46         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json')\r
  47         self.failUnlessEqual(status.HTTP_404_NOT_FOUND, response.status_code)\r
  48 \r
  49     def test_operate_vnf_conflict(self):\r
  50         req_data = {\r
  51             "changeStateTo": "STARTED"\r
  52         }\r
  53         NfInstModel(nfinstid='12', nf_name='VNF1', status='NOT_INSTANTIATED').save()\r
  54         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json')\r
  55         self.failUnlessEqual(status.HTTP_409_CONFLICT, response.status_code)\r
  56         NfInstModel(nfinstid='12', nf_name='VNF1', status='NOT_INSTANTIATED').delete()\r
  57 \r
  58     @mock.patch.object(OperateVnf, 'run')\r
  59     def test_operate_vnf_success(self, mock_run):\r
  60         req_data = {\r
  61             "changeStateTo": "STARTED"\r
  62         }\r
  63         NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').save()\r
  64         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json')\r
  65         mock_run.re.return_value = None\r
  66         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)\r
  67         NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').delete()\r
  68 \r
  69     @mock.patch.object(restcall, 'call_req')\r
  70     @mock.patch.object(api, 'call')\r
  71     def test_operate_vnf_success_start(self, mock_call, mock_call_req):\r
  72         NfInstModel.objects.create(nfinstid='1111',\r
  73                                    nf_name='2222',\r
  74                                    vnfminstid='1',\r
  75                                    package_id='todo',\r
  76                                    version='',\r
  77                                    vendor='',\r
  78                                    netype='',\r
  79                                    vnfd_model='',\r
  80                                    status='INSTANTIATED',\r
  81                                    nf_desc='',\r
  82                                    vnfdid='',\r
  83                                    vnfSoftwareVersion='',\r
  84                                    vnfConfigurableProperties='todo',\r
  85                                    localizationLanguage='EN_US',\r
  86                                    create_time=now_time())\r
  87 \r
  88         VmInstModel.objects.create(vmid="1",\r
  89                                    vimid="1",\r
  90                                    resouceid="11",\r
  91                                    insttype=0,\r
  92                                    instid="1111",\r
  93                                    vmname="test_01",\r
  94                                    is_predefined=1,\r
  95                                    operationalstate=1)\r
  96         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
  97         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
  98         t3_action_vm_start_result = [0, json.JSONEncoder().encode(''), '202']\r
  99         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_start_result]\r
 100         mock_call.return_value = None\r
 101         req_data = {\r
 102             "changeStateTo": "STARTED"\r
 103         }\r
 104         self.nf_inst_id = '1111'\r
 105         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 106         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 107         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 108         vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11")\r
 109         self.assertEqual("ACTIVE", vm[0].operationalstate)\r
 110         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 111 \r
 112     @mock.patch.object(restcall, 'call_req')\r
 113     @mock.patch.object(api, 'call')\r
 114     def test_operate_vnf_success_stop(self, mock_call, mock_call_req):\r
 115         NfInstModel.objects.create(nfinstid='1111',\r
 116                                    nf_name='2222',\r
 117                                    vnfminstid='1',\r
 118                                    package_id='todo',\r
 119                                    version='',\r
 120                                    vendor='',\r
 121                                    netype='',\r
 122                                    vnfd_model='',\r
 123                                    status='INSTANTIATED',\r
 124                                    nf_desc='',\r
 125                                    vnfdid='',\r
 126                                    vnfSoftwareVersion='',\r
 127                                    vnfConfigurableProperties='todo',\r
 128                                    localizationLanguage='EN_US',\r
 129                                    create_time=now_time())\r
 130 \r
 131         VmInstModel.objects.create(vmid="1",\r
 132                                    vimid="1",\r
 133                                    resouceid="11",\r
 134                                    insttype=0,\r
 135                                    instid="1111",\r
 136                                    vmname="test_01",\r
 137                                    is_predefined=1,\r
 138                                    operationalstate=1)\r
 139         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 140         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 141         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 142         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]\r
 143         mock_call.return_value = None\r
 144         req_data = {\r
 145             "changeStateTo": "STOPPED"\r
 146         }\r
 147         self.nf_inst_id = '1111'\r
 148         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 149         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 150         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 151         vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11")\r
 152         self.assertEqual("INACTIVE", vm[0].operationalstate)\r
 153         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 154 \r
 155     @mock.patch.object(restcall, 'call_req')\r
 156     @mock.patch.object(api, 'call')\r
 157     def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req):\r
 158         NfInstModel.objects.create(nfinstid='1111',\r
 159                                    nf_name='2222',\r
 160                                    vnfminstid='1',\r
 161                                    package_id='todo',\r
 162                                    version='',\r
 163                                    vendor='',\r
 164                                    netype='',\r
 165                                    vnfd_model='',\r
 166                                    status='INSTANTIATED',\r
 167                                    nf_desc='',\r
 168                                    vnfdid='',\r
 169                                    vnfSoftwareVersion='',\r
 170                                    vnfConfigurableProperties='todo',\r
 171                                    localizationLanguage='EN_US',\r
 172                                    create_time=now_time())\r
 173 \r
 174         VmInstModel.objects.create(vmid="1",\r
 175                                    vimid="1",\r
 176                                    resouceid="11",\r
 177                                    insttype=0,\r
 178                                    instid="1111",\r
 179                                    vmname="test_01",\r
 180                                    is_predefined=1,\r
 181                                    operationalstate=1)\r
 182         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 183         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 184         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 185         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]\r
 186         mock_call.return_value = None\r
 187         req_data = {\r
 188             "changeStateTo": "STOPPED",\r
 189             "stopType": "GRACEFUL",\r
 190             "gracefulStopTimeout": 2\r
 191         }\r
 192         self.nf_inst_id = '1111'\r
 193         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 194         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 195         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 196         vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11")\r
 197         self.assertEqual("INACTIVE", vm[0].operationalstate)\r
 198         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r
 199 \r
 200     @mock.patch.object(restcall, 'call_req')\r
 201     @mock.patch.object(api, 'call')\r
 202     def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req):\r
 203         NfInstModel.objects.create(nfinstid='1111',\r
 204                                    nf_name='2222',\r
 205                                    vnfminstid='1',\r
 206                                    package_id='todo',\r
 207                                    version='',\r
 208                                    vendor='',\r
 209                                    netype='',\r
 210                                    vnfd_model='',\r
 211                                    status='INSTANTIATED',\r
 212                                    nf_desc='',\r
 213                                    vnfdid='',\r
 214                                    vnfSoftwareVersion='',\r
 215                                    vnfConfigurableProperties='todo',\r
 216                                    localizationLanguage='EN_US',\r
 217                                    create_time=now_time())\r
 218 \r
 219         VmInstModel.objects.create(vmid="1",\r
 220                                    vimid="1",\r
 221                                    resouceid="11",\r
 222                                    insttype=0,\r
 223                                    instid="1111",\r
 224                                    vmname="test_01",\r
 225                                    is_predefined=1,\r
 226                                    operationalstate=1)\r
 227         t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200']\r
 228         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 229         t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202']\r
 230         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result]\r
 231         mock_call.return_value = None\r
 232         req_data = {\r
 233             "changeStateTo": "STOPPED",\r
 234             "stopType": "FORCEFUL",\r
 235         }\r
 236         self.nf_inst_id = '1111'\r
 237         self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id)\r
 238         JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY")\r
 239         OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 240         vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11")\r
 241         self.assertEqual("INACTIVE", vm[0].operationalstate)\r
 242         self.assert_job_result(self.job_id, 100, "Operate Vnf success.")\r