Add json for scaling
[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("NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
112
113         self.client = Client()
114         NSInstModel(id=self.ns_inst_id, name="abc", nspackage_id="7", nsd_id="111").save()
115
116     def tearDown(self):
117         NSInstModel.objects.filter().delete()
118
119     @mock.patch.object(NSManualScaleService, 'run')
120     def test_ns_manual_scale(self, mock_run):
121         data = {
122             "scaleType": "SCALE_NS",
123             "scaleNsData": [{
124                 "scaleNsByStepsData": [{
125                     "aspectId": "1",
126                     "numberOfSteps": 1,
127                     "scalingDirection": "0"
128                 }]
129             }]
130         }
131         response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data=data)
132         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
133
134     @mock.patch.object(restcall, 'call_req')
135     def test_ns_manual_scale_thread(self, mock_call):
136         data = {
137             "scaleType": "SCALE_NS",
138             "scaleNsData": [{
139                 "scaleNsByStepsData": [{
140                     "aspectId": "1",
141                     "numberOfSteps": 1,
142                     "scalingDirection": "0"
143                 }]
144             }]
145         }
146         NSManualScaleService(self.ns_inst_id, data, self.job_id).run()
147         self.assertTrue(NSInstModel.objects.get(id=self.ns_inst_id).status, NS_INST_STATUS.ACTIVE)
148
149     def test_swagger_ok(self):
150         resp = self.client.get("/api/nslcm/v1/swagger.json", format='json')
151         self.assertEqual(resp.status_code, status.HTTP_200_OK)
152
153     @mock.patch.object(NSManualScaleService, 'start')
154     def test_ns_manual_scale_empty_data(self, mock_start):
155         mock_start.side_effect = NSLCMException("NS scale failed.")
156         response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data={})
157         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
158         self.assertIn("error", response.data)
159
160     @mock.patch.object(NSManualScaleService, 'start')
161     def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
162         mock_start.side_effect = NSLCMException("NS scale failed.")
163         data = {
164             "scaleType": "SCALE_NS",
165             "scaleNsData": [{
166                 "scaleNsByStepsData": [{
167                     "aspectId": "1",
168                     "numberOfSteps": 1,
169                     "scalingDirection": "0"
170                 }]
171             }]
172         }
173         response = self.client.post("/api/nslcm/v1/ns/11/scale", data=data)
174         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
175         self.assertIn("error", response.data)