Merge "Modify code because of interface"
[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             "scaleNsByStepsData": {
129                 "aspectId": "1",
130                 "numberOfSteps": 1,
131                 "scalingDirection": "0"
132             }
133         }
134         response = self.client.post(
135             "/api/nslcm/v1/ns/%s/scale" %
136             self.ns_inst_id, data=data)
137         self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
138
139     @mock.patch.object(restcall, 'call_req')
140     def test_ns_manual_scale_thread(self, mock_call):
141         data = {
142             "scaleType": "SCALE_NS",
143             "scaleNsData": [{
144                 "scaleNsByStepsData": [{
145                     "aspectId": "1",
146                     "numberOfSteps": 1,
147                     "scalingDirection": "0"
148                 }]
149             }]
150         }
151         NSManualScaleService(self.ns_inst_id, data, self.job_id).run()
152         self.assertTrue(
153             NSInstModel.objects.get(
154                 id=self.ns_inst_id).status,
155             NS_INST_STATUS.ACTIVE)
156
157     def test_swagger_ok(self):
158         resp = self.client.get("/api/nslcm/v1/swagger.json", format='json')
159         self.assertEqual(resp.status_code, status.HTTP_200_OK)
160
161     @mock.patch.object(NSManualScaleService, 'start')
162     def test_ns_manual_scale_empty_data(self, mock_start):
163         mock_start.side_effect = NSLCMException("NS scale failed.")
164         response = self.client.post(
165             "/api/nslcm/v1/ns/%s/scale" %
166             self.ns_inst_id, data={})
167         self.assertEqual(
168             response.status_code,
169             status.HTTP_500_INTERNAL_SERVER_ERROR)
170         self.assertIn("error", response.data)
171
172     @mock.patch.object(NSManualScaleService, 'start')
173     def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
174         mock_start.side_effect = NSLCMException("NS scale failed.")
175         data = {
176             "scaleType": "SCALE_NS",
177             "scaleNsData": [{
178                 "scaleNsByStepsData": [{
179                     "aspectId": "1",
180                     "numberOfSteps": 1,
181                     "scalingDirection": "0"
182                 }]
183             }]
184         }
185         response = self.client.post("/api/nslcm/v1/ns/11/scale", data=data)
186         self.assertEqual(
187             response.status_code,
188             status.HTTP_500_INTERNAL_SERVER_ERROR)
189         self.assertIn("error", response.data)