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.
22 from django.test import TestCase
23 from rest_framework import status
24 from rest_framework.test import APIClient
25 from genericparser.packages.biz.pnf_descriptor import PnfDescriptor
26 from genericparser.packages.const import PKG_STATUS
27 from genericparser.packages.tests.const import pnfd_data
28 from genericparser.pub.config.config import GENERICPARSER_ROOT_PATH
29 from genericparser.pub.database.models import PnfPackageModel, NSPackageModel
30 from genericparser.pub.utils import toscaparsers
33 class TestPnfDescriptor(TestCase):
35 self.client = APIClient()
36 self.user_defined_data = {
41 self.expected_pnfd_info = {
47 'pnfdInvariantId': None,
48 'pnfdOnboardingState': 'CREATED',
49 'onboardingFailureDetails': None,
50 'pnfdUsageState': 'NOT_IN_USE',
51 'userDefinedData': self.user_defined_data,
55 "pnfs": [{"properties": {"id": "m6000_s"}}]
59 file_path = os.path.join(GENERICPARSER_ROOT_PATH, "22")
60 if os.path.exists(file_path):
61 shutil.rmtree(file_path)
63 def test_pnfd_create_normal(self):
64 request_data = {'userDefinedData': self.user_defined_data}
65 expected_reponse_data = {
66 'pnfdOnboardingState': 'CREATED',
67 'pnfdUsageState': 'NOT_IN_USE',
68 'userDefinedData': self.user_defined_data,
72 response = self.client.post(
73 '/api/nsd/v1/pnf_descriptors',
77 response.data.pop('id')
78 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
79 self.assertEqual(expected_reponse_data, response.data)
81 def test_query_multiple_pnfds_normal(self):
82 expected_reponse_data = [
83 copy.deepcopy(self.expected_pnfd_info),
84 copy.deepcopy(self.expected_pnfd_info)
86 expected_reponse_data[0]['id'] = '0'
87 expected_reponse_data[1]['id'] = '1'
89 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
93 onboardingState='CREATED',
94 usageState='NOT_IN_USE',
95 userDefinedData=user_defined_data
97 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
98 self.assertEqual(response.status_code, status.HTTP_200_OK)
99 self.assertEqual(expected_reponse_data, response.data)
101 def test_query_single_pnfd_normal(self):
102 expected_reponse_data = copy.deepcopy(self.expected_pnfd_info)
103 expected_reponse_data['id'] = '22'
105 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
108 onboardingState='CREATED',
109 usageState='NOT_IN_USE',
110 userDefinedData=user_defined_data
113 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', 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_pnfd_failed(self):
118 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
119 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
121 def test_delete_single_pnfd_normal(self):
122 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
125 usageState=PKG_STATUS.NOT_IN_USE,
126 userDefinedData=user_defined_data,
129 NSPackageModel.objects.create(
131 nsdModel=json.JSONEncoder().encode(self.nsdModel)
133 resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
134 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
135 self.assertEqual(None, resp.data)
137 def test_delete_single_pnfd_when_not_exist(self):
138 resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
139 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
140 self.assertEqual(None, resp.data)
142 @mock.patch.object(toscaparsers, "parse_pnfd")
143 def test_pnfd_content_upload_normal(self, mock_parse_pnfd):
144 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
147 usageState=PKG_STATUS.NOT_IN_USE,
148 userDefinedData=user_defined_data_json,
150 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
151 with open('pnfd_content.txt', 'wb') as fp:
154 with open('pnfd_content.txt', 'rb') as fp:
155 resp = self.client.put(
156 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
159 pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId="22")
160 self.assertEqual(pnf_pkg[0].pnfdId, "zte-1.0")
161 self.assertEqual(pnf_pkg[0].onboardingState, PKG_STATUS.ONBOARDED)
162 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
163 self.assertEqual(None, resp.data)
164 os.remove('pnfd_content.txt')
166 def test_pnfd_content_upload_when_pnf_not_exist(self):
167 with open('pnfd_content.txt', 'wb') as fp:
170 with open('pnfd_content.txt', 'rb') as fp:
171 resp = self.client.put(
172 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
175 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
177 @mock.patch.object(toscaparsers, "parse_pnfd")
178 def test_pnfd_content_upload_when_pnfd_exist(self, mock_parse_pnfd):
179 with open('pnfd_content.txt', 'wb') as fp:
183 usageState=PKG_STATUS.NOT_IN_USE,
188 usageState=PKG_STATUS.NOT_IN_USE,
191 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
192 with open('pnfd_content.txt', 'rb') as fp:
193 resp = self.client.put(
194 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
197 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
199 def test_pnfd_download_normal(self):
200 with open('pnfd_content.txt', 'wb') as fp:
201 fp.writelines('test1')
202 fp.writelines('test2')
203 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
206 usageState=PKG_STATUS.NOT_IN_USE,
207 onboardingState=PKG_STATUS.ONBOARDED,
208 userDefinedData=user_defined_data,
209 localFilePath="pnfd_content.txt",
212 resp = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
214 for data in resp.streaming_content:
215 file_content = '%s%s' % (file_content, data)
216 self.assertEqual(resp.status_code, status.HTTP_200_OK)
217 self.assertEqual('test1test2', file_content)
218 os.remove('pnfd_content.txt')
220 def test_pnfd_download_failed(self):
221 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
222 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
224 def test_pnfd_download_when_not_on_boarded(self):
225 with open('pnfd_content.txt', 'wb') as fp:
226 fp.writelines('test1')
227 fp.writelines('test2')
228 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
231 usageState=PKG_STATUS.NOT_IN_USE,
232 onboardingState=PKG_STATUS.CREATED,
233 userDefinedData=user_defined_data,
234 localFilePath="pnfd_content.txt",
237 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
238 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
239 os.remove('pnfd_content.txt')
241 @mock.patch.object(PnfDescriptor, "create")
242 def test_pnfd_create_when_catch_exception(self, mock_create):
243 request_data = {'userDefinedData': self.user_defined_data}
244 mock_create.side_effect = TypeError('integer type')
245 response = self.client.post('/api/nsd/v1/pnf_descriptors', data=request_data, format='json')
246 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
248 @mock.patch.object(PnfDescriptor, "delete_single")
249 def test_delete_single_when_catch_exception(self, mock_delete_single):
250 mock_delete_single.side_effect = TypeError("integer type")
251 response = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
252 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
254 @mock.patch.object(PnfDescriptor, "query_single")
255 def test_query_single_when_catch_exception(self, mock_query_single):
256 mock_query_single.side_effect = TypeError("integer type")
257 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
258 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
260 @mock.patch.object(PnfDescriptor, "query_multiple")
261 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
262 mock_query_muitiple.side_effect = TypeError("integer type")
263 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
264 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
266 @mock.patch.object(PnfDescriptor, "upload")
267 def test_upload_when_catch_exception(self, mock_upload):
268 mock_upload.side_effect = TypeError("integer type")
269 response = self.client.put("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
270 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
272 @mock.patch.object(PnfDescriptor, "download")
273 def test_download_when_catch_exception(self, mock_download):
274 mock_download.side_effect = TypeError("integer type")
275 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
276 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
278 @mock.patch.object(toscaparsers, 'parse_pnfd')
279 def test_pnfd_parse_normal(self, mock_parse_pnfd):
280 PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
281 mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
282 req_data = {"csarId": "8", "inputs": []}
283 resp = self.client.post("/api/parser/v1/parserpnfd", req_data, format='json')
284 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
285 self.assertEqual({"model": '{"c": "d"}'}, resp.data)