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.pub.enum import JOB_MODEL_STATUS, JOB_TYPE
  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("NS", JOB_TYPE.MANUAL_SCALE_VNF, 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(
  55             "NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
  56         package_id = "23"
  57         NSInstModel(
  58             id=ns_inst_id,
  59             name="abc",
  60             nspackage_id=package_id,
  61             nsd_id=package_id).save()
  62         return ns_inst_id, job_id
  63
  64     def insert_new_nf(self):
  65         self.vnfm_inst_id = "1"
  66         NfInstModel.objects.create(
  67             nfinstid="233",
  68             nf_name="name_1",
  69             vnf_id="1",
  70             vnfm_inst_id="1",
  71             ns_inst_id=self.ns_inst_id,
  72             max_cpu='14',
  73             max_ram='12296',
  74             max_hd='101',
  75             max_shd="20",
  76             max_net=10,
  77             status='active',
  78             mnfinstid="ab34-3g5j-de13-ab85-ij93",
  79             package_id="nf_zte_hss",
  80             vnfd_model=json.dumps(VNFD_MODEL_DICT))
  81
  82     @mock.patch.object(NSManualScaleService, 'run')
  83     def test_ns_manual_scale(self, mock_run):
  84         response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data=SCALE_NS_DICT, format='json')
  85         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
  86
  87     def test_ns_manual_scale_error_scaletype(self):
  88         scale_ns_json = SCALE_NS_DICT.copy()
  89         scale_ns_json["scaleType"] = "SCALE_ERR"
  90         NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
  91         jobs = JobModel.objects.filter(jobid=self.job_id)
  92         self.assertEqual(255, jobs[0].progress)
  93
  94     def test_ns_manual_scale_error_nsd_id(self):
  95         scale_ns_json = SCALE_NS_DICT.copy()
  96         scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
  97         NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
  98         jobs = JobModel.objects.filter(jobid=self.job_id)
  99         self.assertEqual(255, jobs[0].progress)
 100
 101     def test_ns_manual_scale_error_aspect(self):
 102         scale_ns_json = SCALE_NS_DICT.copy()
 103         scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
 104         ns_inst_id, job_id = self.insert_new_ns()
 105         job_id = JobUtil.create_job("NS", JOB_TYPE.MANUAL_SCALE_VNF, ns_inst_id)
 106         NSManualScaleService(ns_inst_id, scale_ns_json, job_id).run()
 107         jobs = JobModel.objects.filter(jobid=job_id)
 108         self.assertEqual(255, jobs[0].progress)
 109
 110     @mock.patch.object(catalog, 'get_scalingmap_json_package')
 111     @mock.patch.object(NSManualScaleService, 'do_vnfs_scale')
 112     def test_ns_manual_scale_success(self, mock_do_vnfs_scale, mock_get_scalingmap_json_package):
 113         scale_ns_json = SCALE_NS_DICT.copy()
 114         scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "TIC_EDGE_IMS"
 115         mock_get_scalingmap_json_package.return_value = self.scaling_map_json
 116         mock_do_vnfs_scale.return_value = JOB_MODEL_STATUS.FINISHED
 117         ns_inst_id, job_id = self.insert_new_ns()
 118         job_id = JobUtil.create_job("NS", JOB_TYPE.MANUAL_SCALE_VNF, ns_inst_id)
 119         self.insert_new_nf()
 120         NSManualScaleService(ns_inst_id, scale_ns_json, job_id).run()
 121         jobs = JobModel.objects.filter(jobid=job_id)
 122         self.assertEqual(255, jobs[0].progress)
 123
 124     @mock.patch.object(restcall, 'call_req')
 125     def test_ns_manual_scale_thread(self, mock_call):
 126         scale_ns_json = SCALE_NS_DICT.copy()
 127         NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
 128         self.assertTrue(NSInstModel.objects.get(id=self.ns_inst_id).status, NS_INST_STATUS.ACTIVE)
 129
 130     @mock.patch.object(NSManualScaleService, 'start')
 131     def test_ns_manual_scale_empty_data(self, mock_start):
 132         mock_start.side_effect = NSLCMException("NS scale failed.")
 133         response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data={}, format='json')
 134         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 135         self.assertIn("error", response.data)
 136
 137     @mock.patch.object(NSManualScaleService, 'start')
 138     def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
 139         mock_start.side_effect = NSLCMException("NS scale failed.")
 140         scale_ns_json = SCALE_NS_DICT.copy()
 141         response = self.client.post("/api/nslcm/v1/ns/11/scale", data=scale_ns_json, format='json')
 142         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 143         self.assertIn("error", response.data)