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.
21 from django.test import TestCase
22 from rest_framework import status
23 from rest_framework.test import APIClient
24 from genericparser.packages.biz.pnf_descriptor import PnfDescriptor
25 from genericparser.packages.const import PKG_STATUS
26 from genericparser.packages.tests.const import pnfd_data
27 from genericparser.pub.database.models import PnfPackageModel, NSPackageModel
28 from genericparser.pub.utils import toscaparsers
31 class TestPnfDescriptor(TestCase):
33 self.client = APIClient()
34 self.user_defined_data = {
39 self.expected_pnfd_info = {
45 'pnfdInvariantId': None,
46 'pnfdOnboardingState': 'CREATED',
47 'onboardingFailureDetails': None,
48 'pnfdUsageState': 'NOT_IN_USE',
49 'userDefinedData': self.user_defined_data,
53 "pnfs": [{"properties": {"id": "m6000_s"}}]
59 def test_pnfd_create_normal(self):
60 request_data = {'userDefinedData': self.user_defined_data}
61 expected_reponse_data = {
62 'pnfdOnboardingState': 'CREATED',
63 'pnfdUsageState': 'NOT_IN_USE',
64 'userDefinedData': self.user_defined_data,
68 response = self.client.post(
69 '/api/nsd/v1/pnf_descriptors',
73 response.data.pop('id')
74 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
75 self.assertEqual(expected_reponse_data, response.data)
77 def test_query_multiple_pnfds_normal(self):
78 expected_reponse_data = [
79 copy.deepcopy(self.expected_pnfd_info),
80 copy.deepcopy(self.expected_pnfd_info)
82 expected_reponse_data[0]['id'] = '0'
83 expected_reponse_data[1]['id'] = '1'
85 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
89 onboardingState='CREATED',
90 usageState='NOT_IN_USE',
91 userDefinedData=user_defined_data
93 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
94 self.assertEqual(response.status_code, status.HTTP_200_OK)
95 self.assertEqual(expected_reponse_data, response.data)
97 def test_query_single_pnfd_normal(self):
98 expected_reponse_data = copy.deepcopy(self.expected_pnfd_info)
99 expected_reponse_data['id'] = '22'
101 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
104 onboardingState='CREATED',
105 usageState='NOT_IN_USE',
106 userDefinedData=user_defined_data
109 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
110 self.assertEqual(response.status_code, status.HTTP_200_OK)
111 self.assertEqual(expected_reponse_data, response.data)
113 def test_query_single_pnfd_failed(self):
114 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
115 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
117 def test_delete_single_pnfd_normal(self):
118 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
121 usageState=PKG_STATUS.NOT_IN_USE,
122 userDefinedData=user_defined_data,
125 NSPackageModel.objects.create(
127 nsdModel=json.JSONEncoder().encode(self.nsdModel)
129 resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
130 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
131 self.assertEqual(None, resp.data)
133 def test_delete_single_pnfd_when_not_exist(self):
134 resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
135 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
136 self.assertEqual(None, resp.data)
138 @mock.patch.object(toscaparsers, "parse_pnfd")
139 def test_pnfd_content_upload_normal(self, mock_parse_pnfd):
140 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
143 usageState=PKG_STATUS.NOT_IN_USE,
144 userDefinedData=user_defined_data_json,
146 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
147 with open('pnfd_content.txt', 'wb') as fp:
150 with open('pnfd_content.txt', 'rb') as fp:
151 resp = self.client.put(
152 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
155 pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId="22")
156 self.assertEqual(pnf_pkg[0].pnfdId, "zte-1.0")
157 self.assertEqual(pnf_pkg[0].onboardingState, PKG_STATUS.ONBOARDED)
158 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
159 self.assertEqual(None, resp.data)
160 os.remove('pnfd_content.txt')
162 def test_pnfd_content_upload_when_pnf_not_exist(self):
163 with open('pnfd_content.txt', 'wb') as fp:
166 with open('pnfd_content.txt', 'rb') as fp:
167 resp = self.client.put(
168 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
171 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
173 @mock.patch.object(toscaparsers, "parse_pnfd")
174 def test_pnfd_content_upload_when_pnfd_exist(self, mock_parse_pnfd):
175 with open('pnfd_content.txt', 'wb') as fp:
179 usageState=PKG_STATUS.NOT_IN_USE,
184 usageState=PKG_STATUS.NOT_IN_USE,
187 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
188 with open('pnfd_content.txt', 'rb') as fp:
189 resp = self.client.put(
190 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
193 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
195 def test_pnfd_download_normal(self):
196 with open('pnfd_content.txt', 'wb') as fp:
197 fp.writelines('test1')
198 fp.writelines('test2')
199 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
202 usageState=PKG_STATUS.NOT_IN_USE,
203 onboardingState=PKG_STATUS.ONBOARDED,
204 userDefinedData=user_defined_data,
205 localFilePath="pnfd_content.txt",
208 resp = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
210 for data in resp.streaming_content:
211 file_content = '%s%s' % (file_content, data)
212 self.assertEqual(resp.status_code, status.HTTP_200_OK)
213 self.assertEqual('test1test2', file_content)
214 os.remove('pnfd_content.txt')
216 def test_pnfd_download_failed(self):
217 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
218 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
220 def test_pnfd_download_when_not_on_boarded(self):
221 with open('pnfd_content.txt', 'wb') as fp:
222 fp.writelines('test1')
223 fp.writelines('test2')
224 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
227 usageState=PKG_STATUS.NOT_IN_USE,
228 onboardingState=PKG_STATUS.CREATED,
229 userDefinedData=user_defined_data,
230 localFilePath="pnfd_content.txt",
233 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
234 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
235 os.remove('pnfd_content.txt')
237 @mock.patch.object(PnfDescriptor, "create")
238 def test_pnfd_create_when_catch_exception(self, mock_create):
239 request_data = {'userDefinedData': self.user_defined_data}
240 mock_create.side_effect = TypeError('integer type')
241 response = self.client.post('/api/nsd/v1/pnf_descriptors', data=request_data, format='json')
242 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
244 @mock.patch.object(PnfDescriptor, "delete_single")
245 def test_delete_single_when_catch_exception(self, mock_delete_single):
246 mock_delete_single.side_effect = TypeError("integer type")
247 response = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
248 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
250 @mock.patch.object(PnfDescriptor, "query_single")
251 def test_query_single_when_catch_exception(self, mock_query_single):
252 mock_query_single.side_effect = TypeError("integer type")
253 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
254 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
256 @mock.patch.object(PnfDescriptor, "query_multiple")
257 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
258 mock_query_muitiple.side_effect = TypeError("integer type")
259 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
260 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
262 @mock.patch.object(PnfDescriptor, "upload")
263 def test_upload_when_catch_exception(self, mock_upload):
264 mock_upload.side_effect = TypeError("integer type")
265 response = self.client.put("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
266 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
268 @mock.patch.object(PnfDescriptor, "download")
269 def test_download_when_catch_exception(self, mock_download):
270 mock_download.side_effect = TypeError("integer type")
271 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
272 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
274 @mock.patch.object(toscaparsers, 'parse_pnfd')
275 def test_pnfd_parse_normal(self, mock_parse_pnfd):
276 PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
277 mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
278 req_data = {"csarId": "8", "inputs": []}
279 resp = self.client.post("/api/parser/v1/parserpnfd", req_data, format='json')
280 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
281 self.assertEqual({"model": '{"c": "d"}'}, resp.data)