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
26 from catalog.packages.biz.vnf_package import VnfPackage, VnfPkgUploadThread
27 from catalog.packages.const import PKG_STATUS
28 from catalog.packages.tests.const import vnfd_data
29 from catalog.pub.config.config import CATALOG_ROOT_PATH
30 from catalog.pub.database.models import VnfPackageModel
31 from catalog.pub.utils import toscaparser
33 VNF_BASE_URL = "/api/vnfpkgm/v1/vnf_packages"
44 class TestVnfPackage(TestCase):
46 self.client = APIClient()
49 file_path = os.path.join(CATALOG_ROOT_PATH, "222")
50 if os.path.exists(file_path):
51 shutil.rmtree(file_path)
53 @mock.patch.object(toscaparser, 'parse_vnfd')
54 def test_upload_vnf_pkg(self, mock_parse_vnfd):
55 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rt")}
56 VnfPackageModel.objects.create(
58 onboardingState="CREATED"
60 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
61 response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
62 vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
63 self.assertEqual("zte-hss-1.0", vnf_pkg[0].vnfdId)
64 self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
65 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
67 def test_upload_vnf_pkg_failed(self):
68 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
69 VnfPackageModel.objects.create(
72 response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
73 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
75 @mock.patch.object(toscaparser, 'parse_vnfd')
76 @mock.patch.object(urllib.request, 'urlopen')
77 def test_upload_nf_pkg_from_uri(self, mock_urlopen, mock_parse_vnfd):
78 vnf_pkg = VnfPackageModel.objects.create(
80 onboardingState="CREATED"
82 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
83 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
84 mock_urlopen.return_value = MockReq()
85 vnf_pkg_id = vnf_pkg.vnfPackageId
86 VnfPkgUploadThread(req_data, vnf_pkg_id).run()
87 vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222")
88 self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId)
90 def test_upload_from_uri_failed(self):
91 req_data = {"username": "123"}
92 response = self.client.post("%s/111/package_content/upload_from_uri" % VNF_BASE_URL, data=req_data)
93 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
95 def test_create_vnf_pkg(self):
97 "userDefinedData": {"a": "A"}
99 response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
100 resp_data = json.loads(response.content)
102 "id": resp_data.get("id"),
103 "onboardingState": "CREATED",
104 "operationalState": "DISABLED",
105 "usageState": "NOT_IN_USE",
106 "userDefinedData": {"a": "A"},
107 "_links": None # TODO
109 self.assertEqual(expect_resp_data, resp_data)
110 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
112 def test_query_single_vnf(self):
113 VnfPackageModel.objects.create(
115 vnfdId="zte-hss-1.0",
117 vnfdProductName="hss",
118 vnfSoftwareVersion="1.0.0",
120 checksum='{"algorithm":"111", "hash": "11"}',
121 onboardingState="CREATED",
122 operationalState="DISABLED",
123 usageState="NOT_IN_USE",
124 userDefinedData='{"a": "A"}'
126 response = self.client.get("%s/222" % VNF_BASE_URL)
129 "vnfdId": "zte-hss-1.0",
130 "vnfProductName": "hss",
131 "vnfSoftwareVersion": "1.0.0",
132 "vnfdVersion": "1.0.0",
133 "checksum": {"algorithm": "111", "hash": "11"},
134 "softwareImages": None,
135 "additionalArtifacts": None,
136 "onboardingState": "CREATED",
137 "operationalState": "DISABLED",
138 "usageState": "NOT_IN_USE",
139 "userDefinedData": {"a": "A"},
142 self.assertEqual(response.data, expect_data)
143 self.assertEqual(response.status_code, status.HTTP_200_OK)
145 def test_query_single_vnf_failed(self):
146 response = self.client.get(VNF_BASE_URL + "/222")
147 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
149 def test_query_multiple_vnf(self):
150 VnfPackageModel.objects.create(
152 vnfdId="zte-hss-1.0",
154 vnfdProductName="hss",
155 vnfSoftwareVersion="1.0.0",
157 checksum='{"algorithm":"111", "hash": "11"}',
158 onboardingState="CREATED",
159 operationalState="DISABLED",
160 usageState="NOT_IN_USE",
161 userDefinedData='{"a": "A"}'
163 VnfPackageModel.objects.create(
165 vnfdId="zte-hss-1.0",
167 vnfdProductName="hss",
168 vnfSoftwareVersion="1.0.0",
170 checksum='{"algorithm":"111", "hash": "11"}',
171 onboardingState="CREATED",
172 operationalState="DISABLED",
173 usageState="NOT_IN_USE",
174 userDefinedData='{"a": "A"}'
176 response = self.client.get(VNF_BASE_URL)
180 "vnfdId": "zte-hss-1.0",
181 "vnfProductName": "hss",
182 "vnfSoftwareVersion": "1.0.0",
183 "vnfdVersion": "1.0.0",
184 "checksum": {"algorithm": "111", "hash": "11"},
185 "softwareImages": None,
186 "additionalArtifacts": None,
187 "onboardingState": "CREATED",
188 "operationalState": "DISABLED",
189 "usageState": "NOT_IN_USE",
190 "userDefinedData": {"a": "A"},
195 "vnfdId": "zte-hss-1.0",
196 "vnfProductName": "hss",
197 "vnfSoftwareVersion": "1.0.0",
198 "vnfdVersion": "1.0.0",
199 "checksum": {"algorithm": "111", "hash": "11"},
200 "softwareImages": None,
201 "additionalArtifacts": None,
202 "onboardingState": "CREATED",
203 "operationalState": "DISABLED",
204 "usageState": "NOT_IN_USE",
205 "userDefinedData": {"a": "A"},
209 self.assertEqual(response.data, expect_data)
210 self.assertEqual(response.status_code, status.HTTP_200_OK)
212 def test_delete_single_vnf_pkg(self):
213 VnfPackageModel.objects.create(
215 vnfdId="zte-hss-1.0",
217 vnfdProductName="hss",
218 vnfSoftwareVersion="1.0.0",
220 checksum='{"algorithm":"111", "hash": "11"}',
221 onboardingState="CREATED",
222 operationalState="DISABLED",
223 usageState="NOT_IN_USE",
224 userDefinedData='{"a": "A"}'
226 response = self.client.delete(VNF_BASE_URL + "/222")
227 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
228 self.assertEqual(response.data, None)
230 def test_delete_when_vnf_pkg_not_exist(self):
231 response = self.client.delete(VNF_BASE_URL + "/222")
232 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
233 self.assertEqual(response.data, None)
235 def test_fetch_vnf_pkg(self):
236 with open("vnfPackage.csar", "wt") as fp:
237 fp.writelines("AAAABBBBCCCCDDDD")
238 VnfPackageModel.objects.create(
240 onboardingState="ONBOARDED",
241 localFilePath="vnfPackage.csar"
243 response = self.client.get(VNF_BASE_URL + "/222/package_content")
245 for data in response.streaming_content:
246 file_content = file_content + data.decode()
247 self.assertEqual(response.status_code, status.HTTP_200_OK)
248 self.assertEqual('AAAABBBBCCCCDDDD', file_content)
249 os.remove("vnfPackage.csar")
251 def test_fetch_partical_vnf_pkg(self):
252 with open("vnfPackage.csar", "wt") as fp:
253 fp.writelines("AAAABBBBCCCCDDDD")
254 VnfPackageModel.objects.create(
256 onboardingState="ONBOARDED",
257 localFilePath="vnfPackage.csar"
259 response = self.client.get("%s/222/package_content" % VNF_BASE_URL, HTTP_RANGE="4-7")
260 partial_file_content = ''
261 for data in response.streaming_content:
262 partial_file_content = partial_file_content + data.decode()
263 self.assertEqual(response.status_code, status.HTTP_200_OK)
264 self.assertEqual('BBB', partial_file_content)
265 os.remove("vnfPackage.csar")
267 def test_fetch_last_partical_vnf_pkg(self):
268 with open("vnfPackage.csar", "wt") as fp:
269 fp.writelines("AAAABBBBCCCCDDDD")
270 VnfPackageModel.objects.create(
272 onboardingState="ONBOARDED",
273 localFilePath="vnfPackage.csar"
275 response = self.client.get(VNF_BASE_URL + "/222/package_content", HTTP_RANGE=" 4-")
276 partial_file_content = ''
277 for data in response.streaming_content:
278 partial_file_content = partial_file_content + data.decode()
279 self.assertEqual(response.status_code, status.HTTP_200_OK)
280 self.assertEqual('BBBBCCCCDDDD', partial_file_content)
281 os.remove("vnfPackage.csar")
283 def test_fetch_vnf_pkg_when_pkg_not_exist(self):
284 response = self.client.get(VNF_BASE_URL + "/222/package_content")
285 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
287 def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
288 VnfPackageModel.objects.create(
290 onboardingState="CREATED",
291 localFilePath="vnfPackage.csar"
293 response = self.client.get(VNF_BASE_URL + "/222/package_content")
294 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
296 def test_download_vnfd(self):
297 VnfPackageModel.objects.create(
299 onboardingState="ONBOARDED",
300 localFilePath=os.path.join(CATALOG_ROOT_PATH, "resource_test.csar")
302 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
303 self.assertEqual(response.status_code, status.HTTP_200_OK)
304 with open("vnfd.csar", 'wb') as local_file:
305 for chunk in response.streaming_content:
306 local_file.write(chunk)
307 self.assertTrue(zipfile.is_zipfile("vnfd.csar"))
308 os.remove("vnfd.csar")
311 def test_download_vnfd_when_pkg_not_exist(self):
312 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
313 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
316 def test_download_vnfd_when_catch_cataloge_exception(self):
317 VnfPackageModel.objects.create(
319 onboardingState="CREATED",
320 localFilePath="vnfPackage.csar"
322 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
323 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
326 @mock.patch.object(VnfPackage, "create_vnf_pkg")
327 def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
328 mock_create_vnf_pkg.side_effect = TypeError('integer type')
330 "userDefinedData": {"a": "A"}
332 response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
333 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
336 @mock.patch.object(VnfPackage, "delete_vnf_pkg")
337 def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
338 mock_delete_vnf_pkg.side_effect = TypeError("integer type")
339 response = self.client.delete(VNF_BASE_URL + "/222")
340 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
343 @mock.patch.object(VnfPackage, "query_single")
344 def test_query_single_when_catch_exception(self, mock_query_single):
345 mock_query_single.side_effect = TypeError("integer type")
346 response = self.client.get(VNF_BASE_URL + "/222")
347 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
350 @mock.patch.object(VnfPackage, "query_multiple")
351 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
352 mock_query_muitiple.side_effect = TypeError("integer type")
353 response = self.client.get(VNF_BASE_URL)
354 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
357 @mock.patch.object(toscaparser, 'parse_vnfd')
358 def test_upload_when_catch_exception(self, mock_parse_vnfd):
359 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
360 VnfPackageModel.objects.create(
362 onboardingState="CREATED"
364 mock_parse_vnfd.side_effect = TypeError("integer type")
365 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
366 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
369 @mock.patch.object(VnfPkgUploadThread, 'start')
370 def test_upload_from_uri_when_catch_exception(self, mock_start):
371 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
372 mock_start.side_effect = TypeError("integer type")
373 response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
374 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
377 @mock.patch.object(VnfPackage, 'download')
378 def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
379 mock_download.side_effect = TypeError("integer type")
380 response = self.client.get(VNF_BASE_URL + "/222/package_content")
381 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
384 @mock.patch.object(toscaparser, 'parse_vnfd')
385 def test_fetch_vnf_artifact(self, mock_parse_vnfd):
386 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
387 VnfPackageModel.objects.create(
389 onboardingState="CREATED"
391 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
392 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
393 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
394 response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
395 self.assertEqual(response.status_code, status.HTTP_200_OK)
396 self.assertEqual(response.getvalue(), b"ubuntu_16.04\n")
399 @mock.patch.object(toscaparser, 'parse_vnfd')
400 def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
401 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
402 VnfPackageModel.objects.create(
404 onboardingState="CREATED"
406 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
407 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
408 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
409 response = self.client.get(VNF_BASE_URL + "/1451/artifacts/image")
410 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
413 @mock.patch.object(toscaparser, 'parse_vnfd')
414 def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
415 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
416 VnfPackageModel.objects.create(
418 onboardingState="CREATED"
420 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
421 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
422 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
423 response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
424 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)