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.
23 from django.test import TestCase
24 from rest_framework import status
25 from rest_framework.test import APIClient
26 from catalog.packages.biz.pnf_descriptor import PnfDescriptor
27 from catalog.packages.const import PKG_STATUS
28 from catalog.packages.tests.const import pnfd_data
29 from catalog.pub.config.config import CATALOG_ROOT_PATH
30 from catalog.pub.database.models import PnfPackageModel, NSPackageModel
31 from catalog.pub.utils import toscaparser
34 class TestPnfDescriptor(TestCase):
36 self.client = APIClient()
37 self.user_defined_data = {
42 self.expected_pnfd_info = {
48 'pnfdInvariantId': None,
49 'pnfdOnboardingState': 'CREATED',
50 'onboardingFailureDetails': None,
51 'pnfdUsageState': 'NOT_IN_USE',
52 'userDefinedData': self.user_defined_data,
56 "pnfs": [{"properties": {"id": "m6000_s"}}]
60 file_path = os.path.join(CATALOG_ROOT_PATH, "22")
61 if os.path.exists(file_path):
62 shutil.rmtree(file_path)
64 def test_pnfd_create_normal(self):
65 request_data = {'userDefinedData': self.user_defined_data}
66 expected_reponse_data = {
67 'pnfdOnboardingState': 'CREATED',
68 'pnfdUsageState': 'NOT_IN_USE',
69 'userDefinedData': self.user_defined_data,
73 response = self.client.post(
74 '/api/nsd/v1/pnf_descriptors',
78 response.data.pop('id')
79 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
80 self.assertEqual(expected_reponse_data, response.data)
82 def test_query_multiple_pnfds_normal(self):
83 expected_reponse_data = [
84 copy.deepcopy(self.expected_pnfd_info),
85 copy.deepcopy(self.expected_pnfd_info)
87 expected_reponse_data[0]['id'] = '0'
88 expected_reponse_data[1]['id'] = '1'
90 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
94 onboardingState='CREATED',
95 usageState='NOT_IN_USE',
96 userDefinedData=user_defined_data
98 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
99 self.assertEqual(response.status_code, status.HTTP_200_OK)
100 self.assertEqual(expected_reponse_data, response.data)
102 def test_query_single_pnfd_normal(self):
103 expected_reponse_data = copy.deepcopy(self.expected_pnfd_info)
104 expected_reponse_data['id'] = '22'
106 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
109 onboardingState='CREATED',
110 usageState='NOT_IN_USE',
111 userDefinedData=user_defined_data
114 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
115 self.assertEqual(response.status_code, status.HTTP_200_OK)
116 self.assertEqual(expected_reponse_data, response.data)
118 def test_query_single_pnfd_failed(self):
119 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
120 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
122 def test_delete_single_pnfd_normal(self):
123 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
126 usageState=PKG_STATUS.NOT_IN_USE,
127 userDefinedData=user_defined_data,
130 NSPackageModel.objects.create(
132 nsdModel=json.JSONEncoder().encode(self.nsdModel)
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 def test_delete_single_pnfd_when_not_exist(self):
139 resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
140 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
141 self.assertEqual(None, resp.data)
143 @mock.patch.object(toscaparser, "parse_pnfd")
144 def test_pnfd_content_upload_normal(self, mock_parse_pnfd):
145 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
148 usageState=PKG_STATUS.NOT_IN_USE,
149 userDefinedData=user_defined_data_json,
151 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
152 with open('pnfd_content.txt', 'wt') as fp:
155 with open('pnfd_content.txt', 'rt') as fp:
156 resp = self.client.put(
157 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
160 pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId="22")
161 self.assertEqual(pnf_pkg[0].pnfdId, "zte-1.0")
162 self.assertEqual(pnf_pkg[0].onboardingState, PKG_STATUS.ONBOARDED)
163 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
164 self.assertEqual(None, resp.data)
165 os.remove('pnfd_content.txt')
167 def test_pnfd_content_upload_when_pnf_not_exist(self):
168 with open('pnfd_content.txt', 'wt') as fp:
171 with open('pnfd_content.txt', 'rt') as fp:
172 resp = self.client.put(
173 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
176 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
178 @mock.patch.object(toscaparser, "parse_pnfd")
179 def test_pnfd_content_upload_when_pnfd_exist(self, mock_parse_pnfd):
180 with open('pnfd_content.txt', 'wt') as fp:
184 usageState=PKG_STATUS.NOT_IN_USE,
189 usageState=PKG_STATUS.NOT_IN_USE,
192 mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
193 with open('pnfd_content.txt', 'rt') as fp:
194 resp = self.client.put(
195 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
198 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
200 def test_pnfd_download_normal(self):
201 with open('pnfd_content.txt', 'wt') as fp:
202 fp.writelines('test1')
203 fp.writelines('test2')
204 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
207 usageState=PKG_STATUS.NOT_IN_USE,
208 onboardingState=PKG_STATUS.ONBOARDED,
209 userDefinedData=user_defined_data,
210 localFilePath="pnfd_content.txt",
213 resp = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
215 for data in resp.streaming_content:
216 file_content = '%s%s' % (file_content, data.decode())
217 self.assertEqual(resp.status_code, status.HTTP_200_OK)
218 self.assertEqual("test1test2", file_content)
219 os.remove('pnfd_content.txt')
221 def test_pnfd_download_failed(self):
222 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
223 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
225 def test_pnfd_download_when_not_on_boarded(self):
226 with open('pnfd_content.txt', 'wt') as fp:
227 fp.writelines('test1')
228 fp.writelines('test2')
229 user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
232 usageState=PKG_STATUS.NOT_IN_USE,
233 onboardingState=PKG_STATUS.CREATED,
234 userDefinedData=user_defined_data,
235 localFilePath="pnfd_content.txt",
238 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
239 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
240 os.remove('pnfd_content.txt')
242 @mock.patch.object(PnfDescriptor, "create")
243 def test_pnfd_create_when_catch_exception(self, mock_create):
244 request_data = {'userDefinedData': self.user_defined_data}
245 mock_create.side_effect = TypeError('integer type')
246 response = self.client.post('/api/nsd/v1/pnf_descriptors', data=request_data, format='json')
247 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
249 @mock.patch.object(PnfDescriptor, "delete_single")
250 def test_delete_single_when_catch_exception(self, mock_delete_single):
251 mock_delete_single.side_effect = TypeError("integer type")
252 response = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
253 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
255 @mock.patch.object(PnfDescriptor, "query_single")
256 def test_query_single_when_catch_exception(self, mock_query_single):
257 mock_query_single.side_effect = TypeError("integer type")
258 response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
259 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
261 @mock.patch.object(PnfDescriptor, "query_multiple")
262 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
263 mock_query_muitiple.side_effect = TypeError("integer type")
264 response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
265 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
267 @mock.patch.object(PnfDescriptor, "upload")
268 def test_upload_when_catch_exception(self, mock_upload):
269 mock_upload.side_effect = TypeError("integer type")
270 response = self.client.put("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
271 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
273 @mock.patch.object(PnfDescriptor, "download")
274 def test_download_when_catch_exception(self, mock_download):
275 mock_download.side_effect = TypeError("integer type")
276 response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
277 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
279 @mock.patch.object(toscaparser, 'parse_pnfd')
280 def test_pnfd_parse_normal(self, mock_parse_pnfd):
281 PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
282 mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
283 req_data = {"csarId": "8", "inputs": []}
284 resp = self.client.post("/api/catalog/v1/parserpnfd", req_data, format='json')
285 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
286 self.assertEqual({"model": '{"c": "d"}'}, resp.data)