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("00342b18-a5c7-11e8-998c-bf1755941f12", 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("00342b18-a5c7-11e8-998c-bf1755941f12", vnf_pkg1[0].vnfdId)
90 def test_upload_from_uri_bad_req(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_400_BAD_REQUEST)
95 @mock.patch.object(urllib.request, 'urlopen')
96 def test_upload_from_uri_failed(self, mock_urlopen):
97 vnf_pkg = VnfPackageModel.objects.create(
99 onboardingState="CREATED"
101 req_data = {"addressInformation": "error"}
102 mock_urlopen.return_value = Exception('Boom!')
103 vnf_pkg_id = vnf_pkg.vnfPackageId
104 VnfPkgUploadThread(req_data, vnf_pkg_id).run()
105 vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="333")
106 self.assertEqual("CREATED", vnf_pkg1[0].onboardingState)
108 def test_create_vnf_pkg(self):
110 "userDefinedData": {"a": "A"}
112 response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
113 resp_data = json.loads(response.content)
115 "id": resp_data.get("id"),
116 "onboardingState": "CREATED",
117 "operationalState": "DISABLED",
118 "usageState": "NOT_IN_USE",
119 "userDefinedData": {"a": "A"},
120 "_links": None # TODO
122 self.assertEqual(expect_resp_data, resp_data)
123 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
125 def test_query_single_vnf(self):
126 VnfPackageModel.objects.create(
128 vnfdId="zte-hss-1.0",
130 vnfdProductName="hss",
131 vnfSoftwareVersion="1.0.0",
133 checksum='{"algorithm":"111", "hash": "11"}',
134 onboardingState="CREATED",
135 operationalState="DISABLED",
136 usageState="NOT_IN_USE",
137 userDefinedData='{"a": "A"}'
139 response = self.client.get("%s/222" % VNF_BASE_URL)
142 "vnfdId": "zte-hss-1.0",
143 "vnfProductName": "hss",
144 "vnfSoftwareVersion": "1.0.0",
145 "vnfdVersion": "1.0.0",
146 "checksum": {"algorithm": "111", "hash": "11"},
147 "softwareImages": None,
148 "additionalArtifacts": None,
149 "onboardingState": "CREATED",
150 "operationalState": "DISABLED",
151 "usageState": "NOT_IN_USE",
152 "userDefinedData": {"a": "A"},
153 "_links": {'self': {'href': 'http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222'},
154 'vnfd': {'href': 'http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/vnfd'},
155 'packageContent': {'href': 'http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/package_content'}
158 self.assertEqual(response.data, expect_data)
159 self.assertEqual(response.status_code, status.HTTP_200_OK)
161 def test_query_single_vnf_failed(self):
162 response = self.client.get(VNF_BASE_URL + "/222")
163 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
165 def test_query_multiple_vnf(self):
166 VnfPackageModel.objects.create(
168 vnfdId="zte-hss-1.0",
170 vnfdProductName="hss",
171 vnfSoftwareVersion="1.0.0",
173 checksum='{"algorithm":"111", "hash": "11"}',
174 onboardingState="CREATED",
175 operationalState="DISABLED",
176 usageState="NOT_IN_USE",
177 userDefinedData='{"a": "A"}'
179 VnfPackageModel.objects.create(
181 vnfdId="zte-hss-1.0",
183 vnfdProductName="hss",
184 vnfSoftwareVersion="1.0.0",
186 checksum='{"algorithm":"111", "hash": "11"}',
187 onboardingState="CREATED",
188 operationalState="DISABLED",
189 usageState="NOT_IN_USE",
190 userDefinedData='{"a": "A"}'
192 response = self.client.get(VNF_BASE_URL)
196 "vnfdId": "zte-hss-1.0",
197 "vnfProductName": "hss",
198 "vnfSoftwareVersion": "1.0.0",
199 "vnfdVersion": "1.0.0",
200 "checksum": {"algorithm": "111", "hash": "11"},
201 "softwareImages": None,
202 "additionalArtifacts": None,
203 "onboardingState": "CREATED",
204 "operationalState": "DISABLED",
205 "usageState": "NOT_IN_USE",
206 "userDefinedData": {"a": "A"},
209 "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/111"
212 "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/111/vnfd"
215 "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/111/package_content"
221 "vnfdId": "zte-hss-1.0",
222 "vnfProductName": "hss",
223 "vnfSoftwareVersion": "1.0.0",
224 "vnfdVersion": "1.0.0",
225 "checksum": {"algorithm": "111", "hash": "11"},
226 "softwareImages": None,
227 "additionalArtifacts": None,
228 "onboardingState": "CREATED",
229 "operationalState": "DISABLED",
230 "usageState": "NOT_IN_USE",
231 "userDefinedData": {"a": "A"},
232 "_links": {'self': {'href': 'http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222'},
233 'vnfd': {'href': 'http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/vnfd'},
235 'href': 'http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/package_content'}}
238 self.assertEqual(response.data, expect_data)
239 self.assertEqual(response.status_code, status.HTTP_200_OK)
241 def test_delete_single_vnf_pkg(self):
242 VnfPackageModel.objects.create(
244 vnfdId="zte-hss-1.0",
246 vnfdProductName="hss",
247 vnfSoftwareVersion="1.0.0",
249 checksum='{"algorithm":"111", "hash": "11"}',
250 onboardingState="CREATED",
251 operationalState="DISABLED",
252 usageState="NOT_IN_USE",
253 userDefinedData='{"a": "A"}'
255 response = self.client.delete(VNF_BASE_URL + "/222")
256 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
257 self.assertEqual(response.data, None)
259 def test_delete_when_vnf_pkg_not_exist(self):
260 response = self.client.delete(VNF_BASE_URL + "/222")
261 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
262 self.assertEqual(response.data, None)
264 def test_fetch_vnf_pkg(self):
265 with open("vnfPackage.csar", "wt") as fp:
266 fp.writelines("AAAABBBBCCCCDDDD")
267 VnfPackageModel.objects.create(
269 onboardingState="ONBOARDED",
270 localFilePath="vnfPackage.csar"
272 response = self.client.get(VNF_BASE_URL + "/222/package_content")
274 for data in response.streaming_content:
275 file_content = file_content + data.decode()
276 self.assertEqual(response.status_code, status.HTTP_200_OK)
277 self.assertEqual('AAAABBBBCCCCDDDD', file_content)
278 os.remove("vnfPackage.csar")
280 def test_fetch_partical_vnf_pkg(self):
281 with open("vnfPackage.csar", "wt") as fp:
282 fp.writelines("AAAABBBBCCCCDDDD")
283 VnfPackageModel.objects.create(
285 onboardingState="ONBOARDED",
286 localFilePath="vnfPackage.csar"
288 response = self.client.get("%s/222/package_content" % VNF_BASE_URL, HTTP_RANGE="4-7")
289 partial_file_content = ''
290 for data in response.streaming_content:
291 partial_file_content = partial_file_content + data.decode()
292 self.assertEqual(response.status_code, status.HTTP_200_OK)
293 self.assertEqual('BBB', partial_file_content)
294 os.remove("vnfPackage.csar")
296 def test_fetch_last_partical_vnf_pkg(self):
297 with open("vnfPackage.csar", "wt") as fp:
298 fp.writelines("AAAABBBBCCCCDDDD")
299 VnfPackageModel.objects.create(
301 onboardingState="ONBOARDED",
302 localFilePath="vnfPackage.csar"
304 response = self.client.get(VNF_BASE_URL + "/222/package_content", HTTP_RANGE=" 4-")
305 partial_file_content = ''
306 for data in response.streaming_content:
307 partial_file_content = partial_file_content + data.decode()
308 self.assertEqual(response.status_code, status.HTTP_200_OK)
309 self.assertEqual('BBBBCCCCDDDD', partial_file_content)
310 os.remove("vnfPackage.csar")
312 def test_fetch_vnf_pkg_when_pkg_not_exist(self):
313 response = self.client.get(VNF_BASE_URL + "/222/package_content")
314 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
316 def test_fetch_vnf_pkg_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/package_content")
323 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
325 def test_download_vnfd(self):
326 VnfPackageModel.objects.create(
328 onboardingState="ONBOARDED",
329 localFilePath=os.path.join(CATALOG_ROOT_PATH, "resource_test.csar")
331 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
332 self.assertEqual(response.status_code, status.HTTP_200_OK)
333 with open("vnfd.csar", 'wb') as local_file:
334 for chunk in response.streaming_content:
335 local_file.write(chunk)
336 self.assertTrue(zipfile.is_zipfile("vnfd.csar"))
337 os.remove("vnfd.csar")
339 def test_download_vnfd_when_pkg_not_exist(self):
340 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
341 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
343 def test_download_vnfd_when_catch_cataloge_exception(self):
344 VnfPackageModel.objects.create(
346 onboardingState="CREATED",
347 localFilePath="vnfPackage.csar"
349 response = self.client.get(VNF_BASE_URL + "/222/vnfd")
350 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
352 @mock.patch.object(VnfPackage, "create_vnf_pkg")
353 def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
354 mock_create_vnf_pkg.side_effect = TypeError('integer type')
356 "userDefinedData": {"a": "A"}
358 response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
359 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
361 @mock.patch.object(VnfPackage, "delete_vnf_pkg")
362 def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
363 mock_delete_vnf_pkg.side_effect = TypeError("integer type")
364 response = self.client.delete(VNF_BASE_URL + "/222")
365 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
367 @mock.patch.object(VnfPackage, "query_single")
368 def test_query_single_when_catch_exception(self, mock_query_single):
369 mock_query_single.side_effect = TypeError("integer type")
370 response = self.client.get(VNF_BASE_URL + "/222")
371 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
373 @mock.patch.object(VnfPackage, "query_multiple")
374 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
375 mock_query_muitiple.side_effect = TypeError("integer type")
376 response = self.client.get(VNF_BASE_URL)
377 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
379 @mock.patch.object(toscaparser, 'parse_vnfd')
380 def test_upload_when_catch_exception(self, mock_parse_vnfd):
381 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
382 VnfPackageModel.objects.create(
384 onboardingState="CREATED"
386 mock_parse_vnfd.side_effect = TypeError("integer type")
387 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
388 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
390 @mock.patch.object(VnfPkgUploadThread, 'start')
391 def test_upload_from_uri_when_catch_exception(self, mock_start):
392 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
393 mock_start.side_effect = TypeError("integer type")
394 response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
395 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
397 @mock.patch.object(VnfPackage, 'download')
398 def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
399 mock_download.side_effect = TypeError("integer type")
400 response = self.client.get(VNF_BASE_URL + "/222/package_content")
401 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
403 @mock.patch.object(toscaparser, 'parse_vnfd')
404 def test_fetch_vnf_artifact(self, mock_parse_vnfd):
405 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
406 VnfPackageModel.objects.create(
408 onboardingState="CREATED"
410 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
411 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
412 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
413 response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
414 self.assertEqual(response.status_code, status.HTTP_200_OK)
415 self.assertEqual(response.getvalue(), b"ubuntu_16.04\n")
417 @mock.patch.object(toscaparser, 'parse_vnfd')
418 def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
419 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
420 VnfPackageModel.objects.create(
422 onboardingState="CREATED"
424 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
425 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
426 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
427 response = self.client.get(VNF_BASE_URL + "/1451/artifacts/image")
428 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
430 @mock.patch.object(toscaparser, 'parse_vnfd')
431 def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
432 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
433 VnfPackageModel.objects.create(
435 onboardingState="CREATED"
437 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
438 response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
439 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
440 response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
441 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
443 def test_upload_vnf_pkg_with_artifacts(self):
444 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "vgw.csar"), "rb")}
445 VnfPackageModel.objects.create(
447 onboardingState="CREATED"
449 response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
450 vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
451 self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
452 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
453 response = self.client.get("%s/222" % VNF_BASE_URL)
455 self.assertEqual(response.status_code, status.HTTP_200_OK)
456 expact_response_data = {
458 "vnfdId": "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
459 "vnfProductName": "vcpe_vgw",
460 "vnfSoftwareVersion": "1.0",
461 "vnfdVersion": "1.0",
462 "softwareImages": None,
463 "additionalArtifacts": [
465 "artifactPath": "MainServiceTemplate.yaml",
472 "onboardingState": "ONBOARDED",
473 "operationalState": "ENABLED",
474 "usageState": "NOT_IN_USE",
477 "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222"
480 "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/vnfd"
483 "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/package_content"
487 self.assertEqual(response.data, expact_response_data)