lcm/ns/tests/test_ns_manual_scale.py

   1 # Copyright 2017 ZTE Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import json
  16 import uuid
  17
  18 import mock
  19 from rest_framework.test import APIClient
  20 from django.test import TestCase
  21 from rest_framework import status
  22
  23 from lcm.ns.biz.ns_manual_scale import NSManualScaleService
  24 from lcm.ns.enum import NS_INST_STATUS
  25 from lcm.pub.database.models import NSInstModel, JobModel, NfInstModel
  26 from lcm.pub.exceptions import NSLCMException
  27 from lcm.pub.msapi import catalog
  28 from lcm.pub.utils import restcall
  29 from lcm.pub.utils.jobutil import JobUtil
  30 from lcm.jobs.enum import JOB_MODEL_STATUS, JOB_TYPE, JOB_ACTION
  31 from lcm.ns.tests import SCALING_MAP_DICT, VNFD_MODEL_DICT, SCALE_NS_DICT
  32
  33
  34 class TestNsManualScale(TestCase):
  35
  36     def setUp(self):
  37         self.scaling_map_json = SCALING_MAP_DICT
  38         self.ns_inst_id = str(uuid.uuid4())
  39         self.job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, self.ns_inst_id)
  40         self.client = APIClient()
  41         self.package_id = "7"
  42         NSInstModel(
  43             id=self.ns_inst_id,
  44             name="abc",
  45             nspackage_id=self.package_id,
  46             nsd_id="111").save()
  47
  48     def tearDown(self):
  49         NSInstModel.objects.filter().delete()
  50         JobModel.objects.filter().delete()
  51
  52     def insert_new_ns(self):
  53         ns_inst_id = str(uuid.uuid4())
  54         job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, self.ns_inst_id)
  55         package_id = "23"
  56         NSInstModel(
  57             id=ns_inst_id,
  58             name="abc",
  59             nspackage_id=package_id,
  60             nsd_id=package_id).save()
  61         return ns_inst_id, job_id
  62
  63     def insert_new_nf(self):
  64         self.vnfm_inst_id = "1"
  65         NfInstModel.objects.create(
  66             nfinstid="233",
  67             nf_name="name_1",
  68             vnf_id="1",
  69             vnfm_inst_id="1",
  70             ns_inst_id=self.ns_inst_id,
  71             max_cpu='14',
  72             max_ram='12296',
  73             max_hd='101',
  74             max_shd="20",
  75             max_net=10,
  76             status='active',
  77             mnfinstid="ab34-3g5j-de13-ab85-ij93",
  78             package_id="nf_zte_hss",
  79             vnfd_model=json.dumps(VNFD_MODEL_DICT))
  80
  81     @mock.patch.object(NSManualScaleService, 'run')
  82     def test_ns_manual_scale(self, mock_run):
  83         response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data=SCALE_NS_DICT, format='json')
  84         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
  85
  86     def test_ns_manual_scale_error_scaletype(self):
  87         scale_ns_json = SCALE_NS_DICT.copy()
  88         scale_ns_json["scaleType"] = "SCALE_ERR"
  89         NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
  90         jobs = JobModel.objects.filter(jobid=self.job_id)
  91         self.assertEqual(255, jobs[0].progress)
  92
  93     def test_ns_manual_scale_error_nsd_id(self):
  94         scale_ns_json = SCALE_NS_DICT.copy()
  95         scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
  96         NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
  97         jobs = JobModel.objects.filter(jobid=self.job_id)
  98         self.assertEqual(255, jobs[0].progress)
  99
 100     def test_ns_manual_scale_error_aspect(self):
 101         scale_ns_json = SCALE_NS_DICT.copy()
 102         scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
 103         ns_inst_id, job_id = self.insert_new_ns()
 104         job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, ns_inst_id)
 105         NSManualScaleService(ns_inst_id, scale_ns_json, job_id).run()
 106         jobs = JobModel.objects.filter(jobid=job_id)
 107         self.assertEqual(255, jobs[0].progress)
 108
 109     @mock.patch.object(catalog, 'get_scalingmap_json_package')
 110     @mock.patch.object(NSManualScaleService, 'do_vnfs_scale')
 111     def test_ns_manual_scale_success(self, mock_do_vnfs_scale, mock_get_scalingmap_json_package):
 112         scale_ns_json = SCALE_NS_DICT.copy()
 113         scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "TIC_EDGE_IMS"
 114         mock_get_scalingmap_json_package.return_value = self.scaling_map_json
 115         mock_do_vnfs_scale.return_value = JOB_MODEL_STATUS.FINISHED
 116         ns_inst_id, job_id = self.insert_new_ns()
 117         job_id = JobUtil.create_job(JOB_TYPE.NS, JOB_ACTION.MANUAL_SCALE, ns_inst_id)
 118         self.insert_new_nf()
 119         NSManualScaleService(ns_inst_id, scale_ns_json, job_id).run()
 120         jobs = JobModel.objects.filter(jobid=job_id)
 121         self.assertEqual(255, jobs[0].progress)
 122
 123     @mock.patch.object(restcall, 'call_req')
 124     def test_ns_manual_scale_thread(self, mock_call):
 125         scale_ns_json = SCALE_NS_DICT.copy()
 126         NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
 127         self.assertTrue(NSInstModel.objects.get(id=self.ns_inst_id).status, NS_INST_STATUS.ACTIVE)
 128
 129     @mock.patch.object(NSManualScaleService, 'start')
 130     def test_ns_manual_scale_empty_data(self, mock_start):
 131         mock_start.side_effect = NSLCMException("NS scale failed.")
 132         response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data={}, format='json')
 133         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 134         self.assertIn("detail", response.data)
 135
 136     @mock.patch.object(NSManualScaleService, 'start')
 137     def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
 138         mock_start.side_effect = NSLCMException("NS scale failed.")
 139         scale_ns_json = SCALE_NS_DICT.copy()
 140         response = self.client.post("/api/nslcm/v1/ns/11/scale", data=scale_ns_json, format='json')
 141         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 142         self.assertIn("detail", response.data)