1 # Copyright 2018 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.
20 from django.test import TestCase
21 from rest_framework import status
22 from rest_framework.test import APIClient
23 from catalog.packages.biz.ns_descriptor import NsDescriptor
24 from catalog.packages.const import PKG_STATUS
25 from catalog.packages.tests.const import nsd_data
26 from catalog.pub.config.config import CATALOG_ROOT_PATH
27 from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
28 from catalog.pub.utils import toscaparser
31 class TestNsDescriptor(TestCase):
33 self.client = APIClient()
34 self.user_defined_data = {
39 self.expected_nsd_info = {
45 'nsdInvariantId': None,
48 'nestedNsdInfoIds': [],
49 'nsdOnboardingState': 'CREATED',
50 'onboardingFailureDetails': None,
51 'nsdOperationalState': 'DISABLED',
52 'nsdUsageState': 'NOT_IN_USE',
53 'userDefinedData': self.user_defined_data,
57 "pnfs": [{"properties": {"id": "m6000_s"}}],
58 "vnfs": [{"properties": {"id": "123"}}]
64 def test_nsd_create_normal(self):
65 reqest_data = {'userDefinedData': self.user_defined_data}
66 expected_reponse_data = {
67 'nsdOnboardingState': 'CREATED',
68 'nsdOperationalState': 'DISABLED',
69 'nsdUsageState': 'NOT_IN_USE',
70 'userDefinedData': self.user_defined_data,
74 response = self.client.post(
75 '/api/nsd/v1/ns_descriptors',
79 response.data.pop('id')
80 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
81 self.assertEqual(expected_reponse_data, response.data)
83 def test_query_multiple_nsds_normal(self):
84 expected_reponse_data = [
85 copy.deepcopy(self.expected_nsd_info),
86 copy.deepcopy(self.expected_nsd_info)
88 expected_reponse_data[0]['id'] = '0'
89 expected_reponse_data[0]['nsdId'] = '0'
90 expected_reponse_data[1]['id'] = '1'
91 expected_reponse_data[1]['nsdId'] = '1'
93 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
97 onboardingState='CREATED',
98 operationalState='DISABLED',
99 usageState='NOT_IN_USE',
100 userDefinedData=user_defined_data,
104 response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
105 self.assertEqual(response.status_code, status.HTTP_200_OK)
106 self.assertEqual(expected_reponse_data, response.data)
108 expected_reponse_data = [
109 copy.deepcopy(self.expected_nsd_info)
111 expected_reponse_data[0]['id'] = '1'
112 expected_reponse_data[0]['nsdId'] = '1'
113 response = self.client.get('/api/nsd/v1/ns_descriptors?nsdId=1', format='json')
114 self.assertEqual(response.status_code, status.HTTP_200_OK)
115 self.assertEqual(expected_reponse_data, response.data)
117 def test_query_single_nsd_normal(self):
118 expected_reponse_data = copy.deepcopy(self.expected_nsd_info)
119 expected_reponse_data['id'] = '22'
121 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
124 onboardingState='CREATED',
125 operationalState='DISABLED',
126 usageState='NOT_IN_USE',
127 userDefinedData=user_defined_data,
128 nsdModel=json.JSONEncoder().encode(self.nsdModel)
131 response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
132 self.assertEqual(response.status_code, status.HTTP_200_OK)
133 self.assertEqual(expected_reponse_data, response.data)
135 def test_query_single_when_ns_not_exist(self):
136 response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
137 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
139 def test_delete_single_nsd_normal(self):
140 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
143 operationalState='DISABLED',
144 usageState='NOT_IN_USE',
145 userDefinedData=user_defined_data,
149 response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
150 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
151 self.assertEqual(None, response.data)
153 def test_delete_when_ns_not_exist(self):
154 response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
155 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
157 @mock.patch.object(toscaparser, 'parse_nsd')
158 def test_nsd_content_upload_normal(self, mock_parse_nsd):
159 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
160 mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
163 vnfdId="vcpe_vfw_zte_1_0"
173 operationalState='DISABLED',
174 usageState='NOT_IN_USE',
175 userDefinedData=user_defined_data_json,
178 with open('nsd_content.txt', 'wt') as fp:
180 with open('nsd_content.txt', 'rt') as fp:
181 resp = self.client.put(
182 "/api/nsd/v1/ns_descriptors/22/nsd_content",
186 with open(os.path.join(CATALOG_ROOT_PATH, '22/nsd_content.txt')) as fp:
188 file_content = '%s%s' % (file_content, data)
189 ns_pkg = NSPackageModel.objects.filter(nsPackageId="22")
190 self.assertEqual("b632bddc-bccd-4180-bd8d-4e8a9578eff7", ns_pkg[0].nsdId)
191 self.assertEqual(PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
192 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
193 self.assertEqual(None, resp.data)
194 self.assertEqual(file_content, 'test')
195 os.remove('nsd_content.txt')
197 def test_nsd_content_upload_failure(self):
198 with open('nsd_content.txt', 'wt') as fp:
200 with open('nsd_content.txt', 'rt') as fp:
201 response = self.client.put(
202 "/api/nsd/v1/ns_descriptors/22/nsd_content",
205 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
207 def test_nsd_content_download_normal(self):
208 with open('nsd_content.txt', 'wt') as fp:
209 fp.writelines('test1')
210 fp.writelines('test2')
211 NSPackageModel.objects.create(
213 onboardingState='ONBOARDED',
214 localFilePath='nsd_content.txt'
216 response = self.client.get(
217 "/api/nsd/v1/ns_descriptors/23/nsd_content", format='json'
220 for data in response.streaming_content:
221 file_content = '%s%s' % (file_content, data.decode())
222 self.assertEqual(response.status_code, status.HTTP_200_OK)
223 self.assertEqual("test1test2", file_content)
224 os.remove('nsd_content.txt')
226 def test_nsd_content_download_when_ns_not_exist(self):
227 response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
228 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
230 def test_nsd_content_download_failed(self):
231 NSPackageModel.objects.create(
233 onboardingState='CREATED',
234 localFilePath='nsd_content.txt'
236 response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
237 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
239 def test_nsd_content_partial_download_normal(self):
240 with open('nsd_content.txt', 'wt') as fp:
241 fp.writelines('test1')
242 fp.writelines('test2')
245 onboardingState='ONBOARDED',
246 localFilePath='nsd_content.txt'
249 response = self.client.get(
250 "/api/nsd/v1/ns_descriptors/23/nsd_content",
254 partial_file_content = ''
255 for data in response.streaming_content:
256 partial_file_content = '%s%s' % (partial_file_content, data.decode())
257 self.assertEqual(response.status_code, status.HTTP_200_OK)
258 self.assertEqual("test2", partial_file_content)
259 os.remove('nsd_content.txt')
261 @mock.patch.object(NsDescriptor, 'create')
262 def test_nsd_create_when_catch_exception(self, mock_create):
263 reqest_data = {'userDefinedData': self.user_defined_data}
264 mock_create.side_effect = TypeError("integer type")
265 response = self.client.post('/api/nsd/v1/ns_descriptors', data=reqest_data, format='json')
266 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
268 @mock.patch.object(NsDescriptor, 'query_single')
269 def test_query_single_when_catch_exception(self, mock_query_single):
270 mock_query_single.side_effect = TypeError("integer type")
271 response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
272 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
274 @mock.patch.object(NsDescriptor, 'query_multiple')
275 def test_query_multiple_when_catch_exception(self, mock_query_multipe):
276 mock_query_multipe.side_effect = TypeError("integer type")
277 response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
278 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
280 @mock.patch.object(NsDescriptor, 'delete_single')
281 def test_delete_when_catch_exception(self, mock_delete_single):
282 mock_delete_single.side_effect = TypeError("integer type")
283 response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
284 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
286 @mock.patch.object(NsDescriptor, 'upload')
287 def test_upload_when_catch_exception(self, mock_upload):
288 mock_upload.side_effect = TypeError("integer type")
289 with open('nsd_content.txt', 'wt') as fp:
291 with open('nsd_content.txt', 'rt') as fp:
292 response = self.client.put("/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp})
293 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
294 os.remove('nsd_content.txt')
296 @mock.patch.object(NsDescriptor, 'download')
297 def test_download_when_catch_exception(self, mock_download):
298 mock_download.side_effect = TypeError("integer type")
299 response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
300 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)