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)
80 for key, value in expected_reponse_data.items():
81 self.assertEqual(response.data[key], value)
83 def test_query_multiple_pnfds_normal(self):
84 expected_reponse_data = [
85 copy.deepcopy(self.expected_pnfd_info),
86 copy.deepcopy(self.expected_pnfd_info)
88 expected_reponse_data[0]['id'] = '0'
89 expected_reponse_data[1]['id'] = '1'
91 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
95 onboardingState='CREATED',
96 usageState='NOT_IN_USE',
97 userDefinedData=user_defined_data
99 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
100 self.assertEqual(response.status_code, status.HTTP_200_OK)
101 self.assertEqual(expected_reponse_data, response.data)
103 def test_query_single_pnfd_normal(self):
104 expected_reponse_data = copy.deepcopy(self.expected_pnfd_info)
105 expected_reponse_data['id'] = '22'
107 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
110 onboardingState='CREATED',
111 usageState='NOT_IN_USE',
112 userDefinedData=user_defined_data
115 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', 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_pnfd_failed(self):
120 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
121 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
123 def test_delete_single_pnfd_normal(self):
124 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
127 usageState=PKG_STATUS.NOT_IN_USE,
128 userDefinedData=user_defined_data,
131 NSPackageModel.objects.create(
133 nsdModel=json.JSONEncoder().encode(self.nsdModel)
135 resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
136 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
137 self.assertEqual(None, resp.data)
139 def test_delete_single_pnfd_when_not_exist(self):
140 resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
141 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
142 self.assertEqual(None, resp.data)
144 @mock.patch.object(toscaparsers, "parse_pnfd")
145 def test_pnfd_content_upload_normal(self, mock_parse_pnfd):
146 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
149 usageState=PKG_STATUS.NOT_IN_USE,
150 userDefinedData=user_defined_data_json,
152 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
153 with open('pnfd_content.txt', 'wt') as fp:
156 with open('pnfd_content.txt', 'rt') as fp:
157 resp = self.client.put(
158 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
161 pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId="22")
162 self.assertEqual(pnf_pkg[0].pnfdId, "zte-1.0")
163 self.assertEqual(pnf_pkg[0].onboardingState, PKG_STATUS.ONBOARDED)
164 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
165 self.assertEqual(None, resp.data)
166 os.remove('pnfd_content.txt')
168 def test_pnfd_content_upload_when_pnf_not_exist(self):
169 with open('pnfd_content.txt', 'wt') as fp:
172 with open('pnfd_content.txt', 'rt') as fp:
173 resp = self.client.put(
174 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
177 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
179 @mock.patch.object(toscaparsers, "parse_pnfd")
180 def test_pnfd_content_upload_when_pnfd_exist(self, mock_parse_pnfd):
181 with open('pnfd_content.txt', 'wt') as fp:
185 usageState=PKG_STATUS.NOT_IN_USE,
190 usageState=PKG_STATUS.NOT_IN_USE,
193 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
194 with open('pnfd_content.txt', 'rt') as fp:
195 resp = self.client.put(
196 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
199 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
201 def test_pnfd_download_normal(self):
202 with open('pnfd_content.txt', 'wt') as fp:
203 fp.writelines('test1')
204 fp.writelines('test2')
205 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
208 usageState=PKG_STATUS.NOT_IN_USE,
209 onboardingState=PKG_STATUS.ONBOARDED,
210 userDefinedData=user_defined_data,
211 localFilePath="pnfd_content.txt",
214 resp = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
216 for data in resp.streaming_content:
217 file_content = '%s%s' % (file_content, data)
218 self.assertEqual(resp.status_code, status.HTTP_200_OK)
219 self.assertEqual("b'test1test2'", file_content)
220 os.remove('pnfd_content.txt')
222 def test_pnfd_download_failed(self):
223 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
224 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
226 def test_pnfd_download_when_not_on_boarded(self):
227 with open('pnfd_content.txt', 'wt') as fp:
228 fp.writelines('test1')
229 fp.writelines('test2')
230 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
233 usageState=PKG_STATUS.NOT_IN_USE,
234 onboardingState=PKG_STATUS.CREATED,
235 userDefinedData=user_defined_data,
236 localFilePath="pnfd_content.txt",
239 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
240 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
241 os.remove('pnfd_content.txt')
243 @mock.patch.object(PnfDescriptor, "create")
244 def test_pnfd_create_when_catch_exception(self, mock_create):
245 request_data = {'userDefinedData': self.user_defined_data}
246 mock_create.side_effect = TypeError('integer type')
247 response = self.client.post('/api/nsd/v1/pnf_descriptors', data=request_data, format='json')
248 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
250 @mock.patch.object(PnfDescriptor, "delete_single")
251 def test_delete_single_when_catch_exception(self, mock_delete_single):
252 mock_delete_single.side_effect = TypeError("integer type")
253 response = self.client.delete("/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_single")
257 def test_query_single_when_catch_exception(self, mock_query_single):
258 mock_query_single.side_effect = TypeError("integer type")
259 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
260 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
262 @mock.patch.object(PnfDescriptor, "query_multiple")
263 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
264 mock_query_muitiple.side_effect = TypeError("integer type")
265 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
266 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
268 @mock.patch.object(PnfDescriptor, "upload")
269 def test_upload_when_catch_exception(self, mock_upload):
270 mock_upload.side_effect = TypeError("integer type")
271 response = self.client.put("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
272 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
274 @mock.patch.object(PnfDescriptor, "download")
275 def test_download_when_catch_exception(self, mock_download):
276 mock_download.side_effect = TypeError("integer type")
277 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
278 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
280 @mock.patch.object(toscaparsers, 'parse_pnfd')
281 def test_pnfd_parse_normal(self, mock_parse_pnfd):
282 PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
283 mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
284 req_data = {"csarId": "8", "inputs": []}
285 resp = self.client.post("/api/parser/v1/parserpnfd", req_data, format='json')
286 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
287 self.assertEqual({"model": '{"c": "d"}'}, resp.data)