1 # Copyright 2019 ZTE Corporation.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
19 from django.test import TestCase
20 from rest_framework import status
21 from rest_framework.test import APIClient
22 from lcm.pub.database.models import NSInstModel
23 from lcm.pub.utils import restcall
24 from lcm.ns.biz.ns_create import CreateNSService
25 from lcm.pub.exceptions import NSLCMException
28 class TestNsInstanceApi(TestCase):
31 self.apiClient = APIClient()
33 self.ns_instances_url = '/api/nslcm/v1/ns_instances'
34 self.nsd_id = str(uuid.uuid4())
35 self.ns_package_id = str(uuid.uuid4())
38 NSInstModel.objects.all().delete()
40 @mock.patch.object(restcall, 'call_req')
41 def test_create_ns(self, mock_call_req):
43 "csarId": self.ns_package_id,
45 "nsPackageId": self.ns_package_id,
49 r1_query_nspackage_from_catalog = [0, json.JSONEncoder().encode(nspackage_info), '201']
50 r2_create_ns_to_aai = [0, json.JSONEncoder().encode({}), '201']
51 mock_call_req.side_effect = [r1_query_nspackage_from_catalog, r2_create_ns_to_aai]
54 'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
55 'HTTP_SERVICETYPE': 'service-type-test1'
61 "nsDescription": "description"
63 response = self.apiClient.post(self.ns_instances_url, data=data, format=self.format, **header)
64 self.failUnlessEqual(status.HTTP_201_CREATED, response.status_code)
65 return response.data['id']
67 @mock.patch.object(restcall, 'call_req')
68 def test_create_ns_when_ns_name_exist(self, mock_call_req):
69 NSInstModel.objects.all().delete()
70 NSInstModel(id="1", name="ns").save()
71 nspackage_info = json.JSONEncoder().encode({
72 "csarId": self.ns_package_id,
75 mock_call_req.return_value = [0, nspackage_info, '200']
77 'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
78 'HTTP_SERVICETYPE': 'service-type-test1'
84 "nsDescription": "description"
86 response = self.apiClient.post(self.ns_instances_url, data=data, format=self.format, **header)
87 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
89 @mock.patch.object(CreateNSService, "do_biz")
90 def test_create_ns_empty_data(self, mock_do_biz):
91 mock_do_biz.side_effect = Exception("Exception in CreateNS.")
96 'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
97 'HTTP_SERVICETYPE': 'service-type-test1'
99 response = self.apiClient.post(self.ns_instances_url, data=data, format=self.format, **header)
100 self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
102 @mock.patch.object(CreateNSService, "do_biz")
103 def test_create_ns_no_header(self, mock_do_biz):
104 mock_do_biz.side_effect = Exception("Exception in CreateNS.")
106 "nsdId": self.nsd_id,
108 "nsDescription": "description"
110 response = self.apiClient.post(self.ns_instances_url, data=data, format=self.format)
111 self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
113 @mock.patch.object(CreateNSService, "do_biz")
114 def test_create_ns_non_existing_nsd(self, mock_do_biz):
115 mock_do_biz.side_effect = NSLCMException("nsd not exists.")
117 "nsdId": self.nsd_id,
119 "nsDescription": "description"
122 'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
123 'HTTP_SERVICETYPE': 'service-type-test1'
125 response = self.apiClient.post(self.ns_instances_url, data=data, format=self.format, **header)
126 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
128 @mock.patch.object(restcall, 'call_req')
129 def test_create_ns_when_fail_to_get_nsd(self, mock_call_req):
130 mock_call_req.return_value = [1, "Failed to get nsd.", '500']
132 "nsdId": self.nsd_id,
134 "nsDescription": "description"
137 'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
138 'HTTP_SERVICETYPE': 'service-type-test1'
140 response = self.apiClient.post(self.ns_instances_url, data=data, format=self.format, **header)
141 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
143 def test_ns_instances_method_not_allowed(self):
145 'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
146 'HTTP_SERVICETYPE': 'service-type-test1'
150 "nsdId": self.nsd_id,
152 "nsDescription": "description"
154 response = self.apiClient.delete(self.ns_instances_url, data=data, format=self.format, **header)
155 self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
156 response = self.apiClient.put(self.ns_instances_url, data=data, format=self.format, **header)
157 self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
158 response = self.apiClient.patch(self.ns_instances_url, data=data, format=self.format, **header)
159 self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
161 def test_invidual_ns_instance_method_not_allowed(self):
163 'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
164 'HTTP_SERVICETYPE': 'service-type-test1'
168 "nsdId": self.nsd_id,
170 "nsDescription": "description"
172 url = self.ns_instances_url + '/1'
173 response = self.apiClient.post(url, data=data, format=self.format, **header)
174 self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
175 response = self.apiClient.put(url, data=data, format=self.format, **header)
176 self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
177 response = self.apiClient.patch(url, data=data, format=self.format, **header)
178 self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
180 def test_query_ns(self):
181 NSInstModel.objects.all().delete()
182 self.test_create_ns()
183 response = self.apiClient.get(self.ns_instances_url)
184 self.failUnlessEqual(status.HTTP_200_OK, response.status_code, response.data)
185 self.assertIsNotNone(response.data)
186 self.assertEqual(1, len(response.data))
187 self.assertEquals(self.nsd_id, response.data[0]['nsdId'])
188 self.assertEquals('ns', response.data[0]['nsInstanceName'])
189 self.assertEquals('NOT_INSTANTIATED', response.data[0]['nsState'])
191 def test_query_one_ns(self):
192 NSInstModel.objects.all().delete()
193 id = self.test_create_ns()
194 url = self.ns_instances_url + '/' + id
195 response = self.apiClient.get(url)
196 self.failUnlessEqual(status.HTTP_200_OK, response.status_code, response.data)
197 self.assertIsNotNone(response.data)
198 self.assertEquals(self.nsd_id, response.data['nsdId'])
199 self.assertEquals('ns', response.data['nsInstanceName'])
200 self.assertEquals('NOT_INSTANTIATED', response.data['nsState'])
202 @mock.patch.object(restcall, 'call_req')
203 def test_delete_ns(self, mock_call_req):
204 NSInstModel(id="1", nspackage_id="7", nsd_id="2").save()
206 "service-instance-id": "service-instance-id-9b9348f2-f75d-4559-823d-db7ac138ed34",
207 "service-instance-name": "service-instance-name-9b9348f2-f75d-4559-823d-db7ac138ed34",
208 "service-type": "service-type-9b9348f2-f75d-4559-823d-db7ac138ed34",
209 "service-role": "service-role-9b9348f2-f75d-4559-823d-db7ac138ed34",
210 "resource-version": "1505350720009"
212 r1_query_ns_to_aai = [0, json.JSONEncoder().encode(ns_info), '200']
213 r2_delete_ns_to_aai = [0, json.JSONEncoder().encode({}), '200']
214 mock_call_req.side_effect = [r1_query_ns_to_aai, r2_delete_ns_to_aai]
215 url = self.ns_instances_url + '/1'
216 response = self.apiClient.delete(url)
217 self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)