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 import config
30 from catalog.pub.config.config import CATALOG_ROOT_PATH
31 from catalog.pub.database.models import VnfPackageModel
32 from catalog.pub.utils import toscaparser
34 VNF_BASE_URL = "/api/vnfpkgm/v1/vnf_packages"
45 class TestVnfPackage(TestCase):
47 self.client = APIClient()
50 file_path = os.path.join(CATALOG_ROOT_PATH, "222")
51 if os.path.exists(file_path):
52 shutil.rmtree(file_path)
54 @mock.patch.object(toscaparser, 'parse_vnfd')
55 def test_upload_vnf_pkg(self, mock_parse_vnfd):
56 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rt")}
57 VnfPackageModel.objects.create(
59 onboardingState="CREATED"
61 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
62 response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
63 vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
64 self.assertEqual("00342b18-a5c7-11e8-998c-bf1755941f12", vnf_pkg[0].vnfdId)
65 self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
66 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
68 def test_upload_vnf_pkg_failed(self):
69 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
70 VnfPackageModel.objects.create(
73 response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
74 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
76 @mock.patch.object(toscaparser, 'parse_vnfd')
77 @mock.patch.object(urllib.request, 'urlopen')
78 def test_upload_nf_pkg_from_uri(self, mock_urlopen, mock_parse_vnfd):
79 vnf_pkg = VnfPackageModel.objects.create(
81 onboardingState="CREATED"
83 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
84 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
85 mock_urlopen.return_value = MockReq()
86 vnf_pkg_id = vnf_pkg.vnfPackageId
87 VnfPkgUploadThread(req_data, vnf_pkg_id).run()
88 vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222")
89 self.assertEqual("00342b18-a5c7-11e8-998c-bf1755941f12", vnf_pkg1[0].vnfdId)
91 def test_upload_from_uri_bad_req(self):
92 req_data = {"username": "123"}
93 response = self.client.post("%s/111/package_content/upload_from_uri" % VNF_BASE_URL, data=req_data)
94 self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
96 @mock.patch.object(urllib.request, 'urlopen')
97 def test_upload_from_uri_failed(self, mock_urlopen):
98 vnf_pkg = VnfPackageModel.objects.create(
100 onboardingState="CREATED"
102 req_data = {"addressInformation": "error"}
103 mock_urlopen.return_value = Exception('Boom!')
104 vnf_pkg_id = vnf_pkg.vnfPackageId
105 VnfPkgUploadThread(req_data, vnf_pkg_id).run()
106 vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="333")
107 self.assertEqual("CREATED", vnf_pkg1[0].onboardingState)
109 def test_create_vnf_pkg(self):
111 "userDefinedData": {"a": "A"}
113 response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
114 resp_data = json.loads(response.content)
116 "id": resp_data.get("id"),
117 "onboardingState": "CREATED",
118 "operationalState": "DISABLED",
119 "usageState": "NOT_IN_USE",
120 "userDefinedData": {"a": "A"},
121 "_links": None # TODO
123 self.assertEqual(expect_resp_data, resp_data)
124 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
126 def test_query_single_vnf(self):
127 VnfPackageModel.objects.create(
129 vnfdId="zte-hss-1.0",
131 vnfdProductName="hss",
132 vnfSoftwareVersion="1.0.0",
134 checksum='{"algorithm":"111", "hash": "11"}',
135 onboardingState="CREATED",
136 operationalState="DISABLED",
137 usageState="NOT_IN_USE",
138 userDefinedData='{"a": "A"}'
140 response = self.client.get("%s/222" % VNF_BASE_URL)
143 "vnfdId": "zte-hss-1.0",
144 "vnfProductName": "hss",
145 "vnfSoftwareVersion": "1.0.0",
146 "vnfdVersion": "1.0.0",
147 "checksum": {"algorithm": "111", "hash": "11"},
148 "softwareImages": None,
149 "additionalArtifacts": None,
150 "onboardingState": "CREATED",
151 "operationalState": "DISABLED",
152 "usageState": "NOT_IN_USE",
153 "userDefinedData": {"a": "A"},
154 "_links": {'self': {'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222'},
156 'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222/vnfd'},
158 'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222/package_content'}
161 self.assertEqual(response.data, expect_data)
162 self.assertEqual(response.status_code, status.HTTP_200_OK)
164 def test_query_single_vnf_failed(self):
165 response = self.client.get(VNF_BASE_URL + "/222")
166 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
168 def test_query_multiple_vnf(self):
169 VnfPackageModel.objects.create(
171 vnfdId="zte-hss-1.0",
173 vnfdProductName="hss",
174 vnfSoftwareVersion="1.0.0",
176 checksum='{"algorithm":"111", "hash": "11"}',
177 onboardingState="CREATED",
178 operationalState="DISABLED",
179 usageState="NOT_IN_USE",
180 userDefinedData='{"a": "A"}'
182 VnfPackageModel.objects.create(
184 vnfdId="zte-hss-1.0",
186 vnfdProductName="hss",
187 vnfSoftwareVersion="1.0.0",
189 checksum='{"algorithm":"111", "hash": "11"}',
190 onboardingState="CREATED",
191 operationalState="DISABLED",
192 usageState="NOT_IN_USE",
193 userDefinedData='{"a": "A"}'
195 response = self.client.get(VNF_BASE_URL)
199 "vnfdId": "zte-hss-1.0",
200 "vnfProductName": "hss",
201 "vnfSoftwareVersion": "1.0.0",
202 "vnfdVersion": "1.0.0",
203 "checksum": {"algorithm": "111", "hash": "11"},
204 "softwareImages": None,
205 "additionalArtifacts": None,
206 "onboardingState": "CREATED",
207 "operationalState": "DISABLED",
208 "usageState": "NOT_IN_USE",
209 "userDefinedData": {"a": "A"},
212 "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/111"
215 "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/111/vnfd"
218 "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/111/package_content"
224 "vnfdId": "zte-hss-1.0",
225 "vnfProductName": "hss",
226 "vnfSoftwareVersion": "1.0.0",
227 "vnfdVersion": "1.0.0",
228 "checksum": {"algorithm": "111", "hash": "11"},
229 "softwareImages": None,
230 "additionalArtifacts": None,
231 "onboardingState": "CREATED",
232 "operationalState": "DISABLED",
233 "usageState": "NOT_IN_USE",
234 "userDefinedData": {"a": "A"},
235 "_links": {'self': {'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222'},
236 'vnfd': {'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222/vnfd'},
238 'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222/package_content'}}
241 self.assertEqual(response.data, expect_data)
242 self.assertEqual(response.status_code, status.HTTP_200_OK)
244 def test_delete_single_vnf_pkg(self):
245 VnfPackageModel.objects.create(
247 vnfdId="zte-hss-1.0",
249 vnfdProductName="hss",
250 vnfSoftwareVersion="1.0.0",
252 checksum='{"algorithm":"111", "hash": "11"}',
253 onboardingState="CREATED",
254 operationalState="DISABLED",
255 usageState="NOT_IN_USE",
256 userDefinedData='{"a": "A"}'
258 response = self.client.delete(VNF_BASE_URL + "/222")
259 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
260 self.assertEqual(response.data, None)
262 def test_delete_when_vnf_pkg_not_exist(self):
263 response = self.client.delete(VNF_BASE_URL + "/222")
264 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
265 self.assertEqual(response.data, None)
267 def test_fetch_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")
277 for data in response.streaming_content:
278 file_content = file_content + data.decode()
279 self.assertEqual(response.status_code, status.HTTP_200_OK)
280 self.assertEqual('AAAABBBBCCCCDDDD', file_content)
281 os.remove("vnfPackage.csar")
283 def test_fetch_partical_vnf_pkg(self):
284 with open("vnfPackage.csar", "wt") as fp:
285 fp.writelines("AAAABBBBCCCCDDDD")
286 VnfPackageModel.objects.create(
288 onboardingState="ONBOARDED",
289 localFilePath="vnfPackage.csar"
291 response = self.client.get("%s/222/package_content" % VNF_BASE_URL, HTTP_RANGE="4-7")
292 partial_file_content = ''
293 for data in response.streaming_content:
294 partial_file_content = partial_file_content + data.decode()
295 self.assertEqual(response.status_code, status.HTTP_200_OK)
296 self.assertEqual('BBB', partial_file_content)
297 os.remove("vnfPackage.csar")
299 def test_fetch_last_partical_vnf_pkg(self):
300 with open("vnfPackage.csar", "wt") as fp:
301 fp.writelines("AAAABBBBCCCCDDDD")
302 VnfPackageModel.objects.create(
304 onboardingState="ONBOARDED",
305 localFilePath="vnfPackage.csar"
307 response = self.client.get(VNF_BASE_URL + "/222/package_content", HTTP_RANGE=" 4-")
308 partial_file_content = ''
309 for data in response.streaming_content:
310 partial_file_content = partial_file_content + data.decode()
311 self.assertEqual(response.status_code, status.HTTP_200_OK)
312 self.assertEqual('BBBBCCCCDDDD', partial_file_content)
313 os.remove("vnfPackage.csar")
315 def test_fetch_vnf_pkg_when_pkg_not_exist(self):
316 response = self.client.get(VNF_BASE_URL + "/222/package_content")
317 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
319 def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
320 VnfPackageModel.objects.create(
322 onboardingState="CREATED",
323 localFilePath="vnfPackage.csar"
325 response = self.client.get(VNF_BASE_URL + "/222/package_content")
326 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
328 def test_download_vnfd(self):
329 VnfPackageModel.objects.create(
331 onboardingState="ONBOARDED",
332 localFilePath=os.path.join(CATALOG_ROOT_PATH, "resource_test.csar")
334 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
335 self.assertEqual(response.status_code, status.HTTP_200_OK)
336 with open("vnfd.csar", 'wb') as local_file:
337 for chunk in response.streaming_content:
338 local_file.write(chunk)
339 self.assertTrue(zipfile.is_zipfile("vnfd.csar"))
340 os.remove("vnfd.csar")
342 def test_download_vnfd_when_pkg_not_exist(self):
343 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
344 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
346 def test_download_vnfd_when_catch_cataloge_exception(self):
347 VnfPackageModel.objects.create(
349 onboardingState="CREATED",
350 localFilePath="vnfPackage.csar"
352 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
353 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
355 @mock.patch.object(VnfPackage, "create_vnf_pkg")
356 def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
357 mock_create_vnf_pkg.side_effect = TypeError('integer type')
359 "userDefinedData": {"a": "A"}
361 response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
362 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
364 @mock.patch.object(VnfPackage, "delete_vnf_pkg")
365 def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
366 mock_delete_vnf_pkg.side_effect = TypeError("integer type")
367 response = self.client.delete(VNF_BASE_URL + "/222")
368 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
370 @mock.patch.object(VnfPackage, "query_single")
371 def test_query_single_when_catch_exception(self, mock_query_single):
372 mock_query_single.side_effect = TypeError("integer type")
373 response = self.client.get(VNF_BASE_URL + "/222")
374 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
376 @mock.patch.object(VnfPackage, "query_multiple")
377 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
378 mock_query_muitiple.side_effect = TypeError("integer type")
379 response = self.client.get(VNF_BASE_URL)
380 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
382 @mock.patch.object(toscaparser, 'parse_vnfd')
383 def test_upload_when_catch_exception(self, mock_parse_vnfd):
384 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
385 VnfPackageModel.objects.create(
387 onboardingState="CREATED"
389 mock_parse_vnfd.side_effect = TypeError("integer type")
390 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
391 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
393 @mock.patch.object(VnfPkgUploadThread, 'start')
394 def test_upload_from_uri_when_catch_exception(self, mock_start):
395 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
396 mock_start.side_effect = TypeError("integer type")
397 response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
398 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
400 @mock.patch.object(VnfPackage, 'download')
401 def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
402 mock_download.side_effect = TypeError("integer type")
403 response = self.client.get(VNF_BASE_URL + "/222/package_content")
404 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
406 @mock.patch.object(toscaparser, 'parse_vnfd')
407 def test_fetch_vnf_artifact(self, mock_parse_vnfd):
408 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
409 VnfPackageModel.objects.create(
411 onboardingState="CREATED"
413 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
414 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
415 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
416 response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
417 self.assertEqual(response.status_code, status.HTTP_200_OK)
418 self.assertEqual(response.getvalue(), b"ubuntu_16.04\n")
420 @mock.patch.object(toscaparser, 'parse_vnfd')
421 def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
422 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
423 VnfPackageModel.objects.create(
425 onboardingState="CREATED"
427 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
428 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
429 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
430 response = self.client.get(VNF_BASE_URL + "/1451/artifacts/image")
431 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
433 @mock.patch.object(toscaparser, 'parse_vnfd')
434 def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
435 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
436 VnfPackageModel.objects.create(
438 onboardingState="CREATED"
440 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
441 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
442 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
443 response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
444 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
446 def test_upload_vnf_pkg_with_artifacts(self):
447 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "vgw.csar"), "rb")}
448 VnfPackageModel.objects.create(
450 onboardingState="CREATED"
452 response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
453 vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
454 self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
455 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
456 response = self.client.get("%s/222" % VNF_BASE_URL)
458 self.assertEqual(response.status_code, status.HTTP_200_OK)
459 expact_response_data = {
461 "vnfdId": "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
462 "vnfProductName": "vcpe_vgw",
463 "vnfSoftwareVersion": "1.0",
464 "vnfdVersion": "1.0",
465 "softwareImages": None,
466 "additionalArtifacts": [
468 "artifactPath": "MainServiceTemplate.yaml",
475 "onboardingState": "ONBOARDED",
476 "operationalState": "ENABLED",
477 "usageState": "NOT_IN_USE",
480 "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/222"
483 "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/222/vnfd"
486 "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/222/package_content"
490 self.assertEqual(response.data, expact_response_data)