Modify a scale_data structure and test cases
[vfc/nfvo/lcm.git] / 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 uuid
16
17 import mock
18 from django.test import Client
19 from django.test import TestCase
20 from rest_framework import status
21
22 from lcm.ns.const import NS_INST_STATUS
23 from lcm.ns.ns_manual_scale import NSManualScaleService
24 from lcm.pub.database.models import NSInstModel
25 from lcm.pub.exceptions import NSLCMException
26 from lcm.pub.utils import restcall
27 from lcm.pub.utils.jobutil import JobUtil, JOB_TYPE
28
29
30 SCALING_JSON = {
31     "scale_options": [
32         {
33             "nsd_id": "ns_ims",
34             "ns_scale_aspect": "TIC_CORE_IMS",
35             "ns_scale_info": [
36                 {
37                     "step": "1",
38                     "scale_list": [
39                         {
40                             "vnfd_id": "zte_ims_cscf",
41                             "vnf_scale_aspect": "mpu",
42                             "numberOfSteps": "1"
43                         },
44                         {
45                             "vnfd_id": "zte_ims_hss",
46                             "vnf_scale_aspect": "fpu",
47                             "numberOfSteps": "3"
48                         }
49                     ]
50                 },
51                 {
52                     "step": "2",
53                     "scale_list": [
54                         {
55                             "vnfd_id": "zte_ims_cscf",
56                             "vnf_scale_aspect": "mpu",
57                             "numberOfSteps": "2"
58                         },
59                         {
60                             "vnfd_id": "zte_ims_hss",
61                             "vnf_scale_aspect": "fpu",
62                             "numberOfSteps": "6"
63                         }
64                     ]
65                 }
66             ]
67         },
68         {
69             "nsd_id": "ns_epc",
70             "ns_scale_aspect": "TIC_EDGE_EPC",
71             "ns_scale_info": [
72                 {
73                     "step": "1",
74                     "scale_list": [
75                         {
76                             "vnfd_id": "zte_epc_spgw",
77                             "vnf_scale_aspect": "gpu",
78                             "numberOfSteps": "1"
79                         },
80                         {
81                             "vnfd_id": "zte_epc_tas",
82                             "vnf_scale_aspect": "fpu",
83                             "numberOfSteps": "2"
84                         }
85                     ]
86                 },
87                 {
88                     "step": "2",
89                     "scale_list": [
90                         {
91                             "vnfd_id": "zte_epc_spgw",
92                             "vnf_scale_aspect": "mpu",
93                             "numberOfSteps": "2"
94                         },
95                         {
96                             "vnfd_id": "zte_epc_tas",
97                             "vnf_scale_aspect": "fpu",
98                             "numberOfSteps": "4"
99                         }
100                     ]
101                 }
102             ]
103         }
104     ]
105 }
106
107
108 class TestNsManualScale(TestCase):
109     def setUp(self):
110         self.ns_inst_id = str(uuid.uuid4())
111         self.job_id = JobUtil.create_job(
112             "NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
113
114         self.client = Client()
115         NSInstModel(
116             id=self.ns_inst_id,
117             name="abc",
118             nspackage_id="7",
119             nsd_id="111").save()
120
121     def tearDown(self):
122         NSInstModel.objects.filter().delete()
123
124     @mock.patch.object(NSManualScaleService, 'run')
125     def test_ns_manual_scale(self, mock_run):
126         data = {
127             "scaleType": "SCALE_NS",
128             "scaleNsData": [{
129                 "scaleNsByStepsData": [{
130                     "aspectId": "1",
131                     "numberOfSteps": 1,
132                     "scalingDirection": "0"
133                 }]
134             }]
135         }
136         response = self.client.post(
137             "/api/nslcm/v1/ns/%s/scale" %
138             self.ns_inst_id, data=data)
139         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
140
141     @mock.patch.object(restcall, 'call_req')
142     def test_ns_manual_scale_thread(self, mock_call):
143         data = {
144             "scaleType": "SCALE_NS",
145             "scaleNsData": [{
146                 "scaleNsByStepsData": [{
147                     "aspectId": "1",
148                     "numberOfSteps": 1,
149                     "scalingDirection": "0"
150                 }]
151             }]
152         }
153         NSManualScaleService(self.ns_inst_id, data, self.job_id).run()
154         self.assertTrue(
155             NSInstModel.objects.get(
156                 id=self.ns_inst_id).status,
157             NS_INST_STATUS.ACTIVE)
158
159     def test_swagger_ok(self):
160         resp = self.client.get("/api/nslcm/v1/swagger.json", format='json')
161         self.assertEqual(resp.status_code, status.HTTP_200_OK)
162
163     @mock.patch.object(NSManualScaleService, 'start')
164     def test_ns_manual_scale_empty_data(self, mock_start):
165         mock_start.side_effect = NSLCMException("NS scale failed.")
166         response = self.client.post(
167             "/api/nslcm/v1/ns/%s/scale" %
168             self.ns_inst_id, data={})
169         self.assertEqual(
170             response.status_code,
171             status.HTTP_500_INTERNAL_SERVER_ERROR)
172         self.assertIn("error", response.data)
173
174     @mock.patch.object(NSManualScaleService, 'start')
175     def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
176         mock_start.side_effect = NSLCMException("NS scale failed.")
177         data = {
178             "scaleType": "SCALE_NS",
179             "scaleNsData": [{
180                 "scaleNsByStepsData": [{
181                     "aspectId": "1",
182                     "numberOfSteps": 1,
183                     "scalingDirection": "0"
184                 }]
185             }]
186         }
187         response = self.client.post("/api/nslcm/v1/ns/11/scale", data=data)
188         self.assertEqual(
189             response.status_code,
190             status.HTTP_500_INTERNAL_SERVER_ERROR)
191         self.assertIn("error", response.data)