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 genericparser.packages.biz.ns_descriptor import NsDescriptor
24 from genericparser.packages.const import PKG_STATUS
25 from genericparser.packages.tests.const import nsd_data
26 from genericparser.pub.config.config import GENERICPARSER_ROOT_PATH
27 from genericparser.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
28 from genericparser.pub.utils import toscaparsers
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)
82 for key, value in expected_reponse_data.items():
83 self.assertEqual(response.data[key], value)
85 def test_query_multiple_nsds_normal(self):
86 expected_reponse_data = [
87 copy.deepcopy(self.expected_nsd_info),
88 copy.deepcopy(self.expected_nsd_info)
90 expected_reponse_data[0]['id'] = '0'
91 expected_reponse_data[0]['nsdId'] = '0'
92 expected_reponse_data[1]['id'] = '1'
93 expected_reponse_data[1]['nsdId'] = '1'
95 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
99 onboardingState='CREATED',
100 operationalState='DISABLED',
101 usageState='NOT_IN_USE',
102 userDefinedData=user_defined_data,
106 response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
107 self.assertEqual(response.status_code, status.HTTP_200_OK)
108 self.assertEqual(expected_reponse_data, response.data)
110 expected_reponse_data = [
111 copy.deepcopy(self.expected_nsd_info)
113 expected_reponse_data[0]['id'] = '1'
114 expected_reponse_data[0]['nsdId'] = '1'
115 response = self.client.get('/api/nsd/v1/ns_descriptors?nsdId=1', format='json')
116 self.assertEqual(response.status_code, status.HTTP_200_OK)
117 self.assertEqual(expected_reponse_data, response.data)
119 def test_query_single_nsd_normal(self):
120 expected_reponse_data = copy.deepcopy(self.expected_nsd_info)
121 expected_reponse_data['id'] = '22'
123 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
126 onboardingState='CREATED',
127 operationalState='DISABLED',
128 usageState='NOT_IN_USE',
129 userDefinedData=user_defined_data,
130 nsdModel=json.JSONEncoder().encode(self.nsdModel)
133 response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
134 self.assertEqual(response.status_code, status.HTTP_200_OK)
135 self.assertEqual(expected_reponse_data, response.data)
137 def test_query_single_when_ns_not_exist(self):
138 response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
139 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
141 def test_delete_single_nsd_normal(self):
142 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
145 operationalState='DISABLED',
146 usageState='NOT_IN_USE',
147 userDefinedData=user_defined_data,
151 response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
152 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
153 self.assertEqual(None, response.data)
155 def test_delete_when_ns_not_exist(self):
156 response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
157 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
159 @mock.patch.object(toscaparsers, 'parse_nsd')
160 def test_nsd_content_upload_normal(self, mock_parse_nsd):
161 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
162 mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
165 vnfdId="vcpe_vfw_zte_1_0"
175 operationalState='DISABLED',
176 usageState='NOT_IN_USE',
177 userDefinedData=user_defined_data_json,
180 with open('nsd_content.txt', 'wt') as fp:
182 with open('nsd_content.txt', 'rt') as fp:
183 resp = self.client.put(
184 "/api/nsd/v1/ns_descriptors/22/nsd_content",
188 with open(os.path.join(GENERICPARSER_ROOT_PATH, '22/nsd_content.txt')) as fp:
190 file_content = '%s%s' % (file_content, data)
191 ns_pkg = NSPackageModel.objects.filter(nsPackageId="22")
192 self.assertEqual("VCPE_NS", ns_pkg[0].nsdId)
193 self.assertEqual(PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
194 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
195 self.assertEqual(None, resp.data)
196 self.assertEqual(file_content, 'test')
197 os.remove('nsd_content.txt')
199 def test_nsd_content_upload_failure(self):
200 with open('nsd_content.txt', 'wt') as fp:
202 with open('nsd_content.txt', 'rt') as fp:
203 response = self.client.put(
204 "/api/nsd/v1/ns_descriptors/22/nsd_content",
207 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
209 def test_nsd_content_download_normal(self):
210 with open('nsd_content.txt', 'wt') as fp:
211 fp.writelines('test1')
212 fp.writelines('test2')
213 NSPackageModel.objects.create(
215 onboardingState='ONBOARDED',
216 localFilePath='nsd_content.txt'
218 response = self.client.get(
219 "/api/nsd/v1/ns_descriptors/23/nsd_content", format='json'
222 for data in response.streaming_content:
223 file_content = '%s%s' % (file_content, data)
224 self.assertEqual(response.status_code, status.HTTP_200_OK)
225 self.assertEqual("b'test1test2'", file_content)
226 os.remove('nsd_content.txt')
228 def test_nsd_content_download_when_ns_not_exist(self):
229 response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
230 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
232 def test_nsd_content_download_failed(self):
233 NSPackageModel.objects.create(
235 onboardingState='CREATED',
236 localFilePath='nsd_content.txt'
238 response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
239 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
241 def test_nsd_content_partial_download_normal(self):
242 with open('nsd_content.txt', 'wt') as fp:
243 fp.writelines('test1')
244 fp.writelines('test2')
247 onboardingState='ONBOARDED',
248 localFilePath='nsd_content.txt'
251 response = self.client.get(
252 "/api/nsd/v1/ns_descriptors/23/nsd_content",
256 partial_file_content = ''
257 for data in response.streaming_content:
258 partial_file_content = '%s%s' % (partial_file_content, data)
259 self.assertEqual(response.status_code, status.HTTP_200_OK)
260 self.assertEqual("b'test2'", partial_file_content)
261 os.remove('nsd_content.txt')
263 @mock.patch.object(NsDescriptor, 'create')
264 def test_nsd_create_when_catch_exception(self, mock_create):
265 reqest_data = {'userDefinedData': self.user_defined_data}
266 mock_create.side_effect = TypeError("integer type")
267 response = self.client.post('/api/nsd/v1/ns_descriptors', data=reqest_data, format='json')
268 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
270 @mock.patch.object(NsDescriptor, 'query_single')
271 def test_query_single_when_catch_exception(self, mock_query_single):
272 mock_query_single.side_effect = TypeError("integer type")
273 response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
274 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
276 @mock.patch.object(NsDescriptor, 'query_multiple')
277 def test_query_multiple_when_catch_exception(self, mock_query_multipe):
278 mock_query_multipe.side_effect = TypeError("integer type")
279 response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
280 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
282 @mock.patch.object(NsDescriptor, 'delete_single')
283 def test_delete_when_catch_exception(self, mock_delete_single):
284 mock_delete_single.side_effect = TypeError("integer type")
285 response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
286 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
288 @mock.patch.object(NsDescriptor, 'upload')
289 def test_upload_when_catch_exception(self, mock_upload):
290 mock_upload.side_effect = TypeError("integer type")
291 with open('nsd_content.txt', 'wt') as fp:
293 with open('nsd_content.txt', 'rt') as fp:
294 response = self.client.put("/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp})
295 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
296 os.remove('nsd_content.txt')
298 @mock.patch.object(NsDescriptor, 'download')
299 def test_download_when_catch_exception(self, mock_download):
300 mock_download.side_effect = TypeError("integer type")
301 response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
302 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)