Update python2 to python3
[vfc/nfvo/lcm.git] / lcm / ns / tests / test_sol_ns_scale_api.py
1 # Copyright 2019 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 import mock
18 from django.test import TestCase
19 from rest_framework import status
20 from rest_framework.test import APIClient
21 from lcm.ns.biz.ns_manual_scale import NSManualScaleService
22 from lcm.pub.database.models import NSInstModel, JobModel, NfInstModel
23 from lcm.pub.exceptions import NSLCMException
24 from lcm.pub.utils.jobutil import JobUtil
25 from lcm.pub.enum import JOB_TYPE
26 from lcm.ns.tests import VNFD_MODEL_DICT, SCALE_NS_DICT
27
28
29 class TestScaleNsApi(TestCase):
30     def setUp(self):
31         self.url = "/api/nslcm/v1/ns_instances/%s/scale"
32         self.ns_inst_id = str(uuid.uuid4())
33         self.job_id = JobUtil.create_job("NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
34         self.package_id = "7"
35         self.client = APIClient()
36         NSInstModel(
37             id=self.ns_inst_id,
38             name="abc",
39             nspackage_id=self.package_id,
40             nsd_id="111").save()
41
42     def tearDown(self):
43         NSInstModel.objects.filter().delete()
44         JobModel.objects.filter().delete()
45
46     def insert_new_ns(self):
47         ns_inst_id = str(uuid.uuid4())
48         job_id = JobUtil.create_job("NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
49         package_id = "23"
50         NSInstModel(
51             id=ns_inst_id,
52             name="abc",
53             nspackage_id=package_id,
54             nsd_id=package_id).save()
55         return ns_inst_id, job_id
56
57     def insert_new_nf(self):
58         self.nf_name = "name_1"
59         self.vnf_id = "1"
60         self.vnfm_inst_id = "1"
61         nf_inst_id = "233"
62         package_id = "nf_zte_hss"
63         nf_uuid = "ab34-3g5j-de13-ab85-ij93"
64         NfInstModel.objects.create(
65             nfinstid=nf_inst_id,
66             nf_name=self.nf_name,
67             vnf_id=self.vnf_id,
68             vnfm_inst_id=self.vnfm_inst_id,
69             ns_inst_id=self.ns_inst_id,
70             max_cpu='14',
71             max_ram='12296',
72             max_hd='101',
73             max_shd="20",
74             max_net=10,
75             status='active',
76             mnfinstid=nf_uuid,
77             package_id=package_id,
78             vnfd_model=json.dumps(VNFD_MODEL_DICT)
79         )
80
81     @mock.patch.object(NSManualScaleService, 'run')
82     def test_ns_scale(self, mock_run):
83         response = self.client.post(self.url % self.ns_inst_id, data=SCALE_NS_DICT)
84         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
85         self.assertIsNotNone(response['Location'])
86         response = self.client.get(response['Location'], format='json')
87         self.assertEqual(response.status_code, status.HTTP_200_OK)
88
89     @mock.patch.object(NSManualScaleService, 'start')
90     def test_ns_manual_scale_empty_data(self, mock_start):
91         mock_start.side_effect = NSLCMException("NS scale failed.")
92         response = self.client.post(self.url % self.ns_inst_id, data={}, format='json')
93         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
94
95     @mock.patch.object(NSManualScaleService, 'start')
96     def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
97         mock_start.side_effect = NSLCMException("NS scale failed.")
98         response = self.client.post(self.url % '11', data=SCALE_NS_DICT)
99         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
100
101     def test_method_not_allowed(self):
102         response = self.client.put(self.url % '1', data={}, format='json')
103         self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
104         response = self.client.patch(self.url % '1', data={}, format='json')
105         self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
106         response = self.client.delete(self.url % '1', data={}, format='json')
107         self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
108         response = self.client.get(self.url % '1', data={}, format='json')
109         self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)