lcm/lcm/nf/tests/test_heal_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\r
  19 from rest_framework import status\r
  20 from rest_framework.test import APIClient\r
  21 \r
  22 from lcm.nf.biz.heal_vnf import HealVnf\r
  23 from .const import c1_data_get_tenant_id_1, c9_data_create_vm, c10_data_get_vm, vnfd_model_dict\r
  24 from lcm.pub.database.models import NfInstModel, JobStatusModel, VmInstModel\r
  25 from lcm.pub.utils import restcall\r
  26 from lcm.pub.utils.jobutil import JobUtil\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 TestNFInstantiate(TestCase):\r
  32     def setUp(self):\r
  33         self.client = APIClient()\r
  34         self.grant_result = {\r
  35             "vimid": 'vimid_1',\r
  36             "tenant": 'tenantname_1',\r
  37         }\r
  38         self.getvmResult = {\r
  39             "status": "ACTIVE",\r
  40             "id": "11",\r
  41         }\r
  42 \r
  43     def tearDown(self):\r
  44         pass\r
  45 \r
  46     def assert_job_result(self, job_id, job_progress, job_detail):\r
  47         jobs = JobStatusModel.objects.filter(jobid=job_id,\r
  48                                              progress=job_progress,\r
  49                                              descp=job_detail)\r
  50         self.assertEqual(1, len(jobs))\r
  51 \r
  52     def test_heal_vnf_not_found(self):\r
  53         req_data = {}\r
  54         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/heal", data=req_data, format='json')\r
  55         self.failUnlessEqual(status.HTTP_404_NOT_FOUND, response.status_code)\r
  56 \r
  57     def test_heal_vnf_conflict(self):\r
  58         req_data = {}\r
  59         NfInstModel(\r
  60             nfinstid='1267',\r
  61             nf_name='VNF1',\r
  62             status='NOT_INSTANTIATED').save()\r
  63         response = self.client.post("/api/vnflcm/v1/vnf_instances/1267/heal", data=req_data, format='json')\r
  64         self.failUnlessEqual(status.HTTP_409_CONFLICT, response.status_code)\r
  65         NfInstModel.objects.filter(nfinstid='1267').delete()\r
  66 \r
  67     def test_change_ext_conn_inner_error(self):\r
  68         NfInstModel(nfinstid='345',\r
  69                     nf_name='VNF1',\r
  70                     nf_desc="VNF DESC",\r
  71                     vnfdid="1",\r
  72                     netype="XGW",\r
  73                     vendor="ZTE",\r
  74                     vnfSoftwareVersion="V1",\r
  75                     version="V1",\r
  76                     package_id="2",\r
  77                     status='INSTANTIATED').save()\r
  78         url = "/api/vnflcm/v1/vnf_instances/345/heal"\r
  79         response = self.client.post(url,\r
  80                                     data={"additionalParams": "1"},\r
  81                                     format='json')\r
  82         self.failUnlessEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)\r
  83 \r
  84     @mock.patch.object(HealVnf, 'run')\r
  85     def test_heal_vnf_success(self, mock_run):\r
  86         req_data = {}\r
  87         NfInstModel(nfinstid='12',\r
  88                     nf_name='VNF1',\r
  89                     nf_desc="VNF DESC",\r
  90                     vnfdid="1",\r
  91                     netype="XGW",\r
  92                     vendor="ZTE",\r
  93                     vnfSoftwareVersion="V1",\r
  94                     version="V1",\r
  95                     package_id="2",\r
  96                     status='INSTANTIATED').save()\r
  97         response = self.client.post("/api/vnflcm/v1/vnf_instances/12/heal", data=req_data, format='json')\r
  98         mock_run.re.return_value = None\r
  99         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)\r
 100         NfInstModel(nfinstid='12').delete()\r
 101 \r
 102     @mock.patch.object(restcall, 'call_req')\r
 103     @mock.patch.object(api, 'call')\r
 104     def test_heal_vnf_success_reboot(self, mock_call, mock_call_req):\r
 105         vim_cache = {}\r
 106         res_cache = {}\r
 107 \r
 108         NfInstModel.objects.create(nfinstid='1111',\r
 109                                    nf_name='2222',\r
 110                                    vnfminstid='1',\r
 111                                    package_id='todo',\r
 112                                    version='',\r
 113                                    vendor='',\r
 114                                    netype='',\r
 115                                    vnfd_model=json.dumps(vnfd_model_dict),\r
 116                                    status='INSTANTIATED',\r
 117                                    nf_desc='',\r
 118                                    vnfdid='',\r
 119                                    vnfSoftwareVersion='',\r
 120                                    vnfConfigurableProperties='todo',\r
 121                                    localizationLanguage='EN_US',\r
 122                                    create_time=now_time(),\r
 123                                    resInfo=json.dumps(res_cache),\r
 124                                    vimInfo=json.dumps(vim_cache))\r
 125 \r
 126         VmInstModel.objects.create(vmid="1",\r
 127                                    vimid="vimid_1",\r
 128                                    resourceid="1",\r
 129                                    insttype=0,\r
 130                                    instid="1111",\r
 131                                    vmname="vduinstname",\r
 132                                    is_predefined=1,\r
 133                                    tenant="tenantname_1",\r
 134                                    operationalstate=1)\r
 135         t1_apply_grant_result = [0, json.JSONEncoder().encode(self.grant_result), '200']\r
 136         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 137         t3_action_get_vm = [0, json.JSONEncoder().encode(self.getvmResult), '202']\r
 138         t4_action_vm_start_reboot = [0, json.JSONEncoder().encode(''), '202']\r
 139         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_get_vm, t4_action_vm_start_reboot]\r
 140         mock_call.side_effect = [self.getvmResult, None]\r
 141         req_data = {\r
 142             "cause": "Error",\r
 143             "additionalParams": {\r
 144                 "action": "vmReset",\r
 145                 "affectedvm": {\r
 146                     "vmid": "1",\r
 147                     "vduid": "vdu1Id",\r
 148                     "vmname": "vduinstname"\r
 149                 }\r
 150             }\r
 151         }\r
 152         self.nf_inst_id = '1111'\r
 153         self.job_id = JobUtil.create_job('NF', 'HEAL', self.nf_inst_id)\r
 154         JobUtil.add_job_status(self.job_id, 0, "HEAL_VNF_READY")\r
 155         HealVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 156         self.assert_job_result(self.job_id, 100, "Heal Vnf success.")\r
 157 \r
 158     @mock.patch.object(restcall, 'call_req')\r
 159     @mock.patch.object(api, 'call')\r
 160     def test_heal_vnf_success_start(self, mock_call, mock_call_req):\r
 161         vim_cache = {}\r
 162         res_cache = {"volume": {}, "flavor": {}, "port": {}}\r
 163         res_cache["volume"]["volume_storage1"] = "vol1"\r
 164         res_cache["flavor"]["vdu1Id"] = "flavor1"\r
 165         res_cache["port"]["cpId1"] = "port1"\r
 166 \r
 167         NfInstModel.objects.create(nfinstid='1111',\r
 168                                    nf_name='2222',\r
 169                                    vnfminstid='1',\r
 170                                    package_id='todo',\r
 171                                    version='',\r
 172                                    vendor='',\r
 173                                    netype='',\r
 174                                    vnfd_model=json.dumps(vnfd_model_dict),\r
 175                                    status='INSTANTIATED',\r
 176                                    nf_desc='',\r
 177                                    vnfdid='',\r
 178                                    vnfSoftwareVersion='',\r
 179                                    vnfConfigurableProperties='todo',\r
 180                                    localizationLanguage='EN_US',\r
 181                                    resInfo=json.dumps(res_cache),\r
 182                                    vimInfo=json.dumps(vim_cache),\r
 183                                    create_time=now_time())\r
 184 \r
 185         t1_apply_grant_result = [0, json.JSONEncoder().encode(self.grant_result), '200']\r
 186         t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200']\r
 187         t3_action_vm_start_create = [0, json.JSONEncoder().encode(''), '202']\r
 188         mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_start_create]\r
 189         mock_call.side_effect = [c1_data_get_tenant_id_1, c9_data_create_vm, c10_data_get_vm]\r
 190         req_data = {\r
 191             "cause": "Error",\r
 192             "additionalParams": {\r
 193                 "action": "vmCreate",\r
 194                 "affectedvm": {\r
 195                     "vmid": "1",\r
 196                     "vduid": "vdu1Id",\r
 197                     "vmname": "vduinstname"\r
 198                 }\r
 199             }\r
 200         }\r
 201         self.nf_inst_id = '1111'\r
 202         self.job_id = JobUtil.create_job('NF', 'HEAL', self.nf_inst_id)\r
 203         JobUtil.add_job_status(self.job_id, 0, "HEAL_VNF_READY")\r
 204         HealVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run()\r
 205         self.assert_job_result(self.job_id, 100, "Heal Vnf success.")\r