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.
21 from django.test import TestCase
22 from rest_framework import status
23 from rest_framework.test import APIClient
25 from genericparser.packages.biz.vnf_package import VnfPackage, VnfPkgUploadThread
26 from genericparser.packages.const import PKG_STATUS
27 from genericparser.packages.tests.const import vnfd_data
28 from genericparser.pub.config.config import GENERICPARSER_ROOT_PATH
29 from genericparser.pub.database.models import VnfPackageModel
30 from genericparser.pub.utils import toscaparsers
41 class TestVnfPackage(TestCase):
43 self.client = APIClient()
46 file_path = os.path.join(GENERICPARSER_ROOT_PATH, "222")
47 if os.path.exists(file_path):
48 shutil.rmtree(file_path)
50 @mock.patch.object(toscaparsers, 'parse_vnfd')
51 def test_upload_vnf_pkg(self, mock_parse_vnfd):
52 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "empty.txt"), "rb")}
53 VnfPackageModel.objects.create(
55 onboardingState="CREATED"
57 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
58 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
59 vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
60 self.assertEqual("zte-hss-1.0", vnf_pkg[0].vnfdId)
61 self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
62 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
64 def test_upload_vnf_pkg_failed(self):
65 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "empty.txt"), "rb")}
66 VnfPackageModel.objects.create(
69 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
70 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
72 @mock.patch.object(toscaparsers, 'parse_vnfd')
73 @mock.patch.object(urllib2, 'urlopen')
74 def test_upload_nf_pkg_from_uri(self, mock_urlopen, mock_parse_vnfd):
75 vnf_pkg = VnfPackageModel.objects.create(
77 onboardingState="CREATED"
79 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
80 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
81 mock_urlopen.return_value = MockReq()
82 vnf_pkg_id = vnf_pkg.vnfPackageId
83 VnfPkgUploadThread(req_data, vnf_pkg_id).run()
84 vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222")
85 self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId)
87 def test_upload_from_uri_failed(self):
88 req_data = {"username": "123"}
89 response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
90 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
92 def test_create_vnf_pkg(self):
94 "userDefinedData": {"a": "A"}
96 response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
97 resp_data = json.loads(response.content)
99 "id": resp_data.get("id"),
100 "onboardingState": "CREATED",
101 "operationalState": "DISABLED",
102 "usageState": "NOT_IN_USE",
103 "userDefinedData": {"a": "A"},
104 "_links": None # TODO
106 self.assertEqual(expect_resp_data, resp_data)
107 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
109 def test_query_single_vnf(self):
110 VnfPackageModel.objects.create(
112 vnfdId="zte-hss-1.0",
114 vnfdProductName="hss",
115 vnfSoftwareVersion="1.0.0",
117 checksum='{"algorithm":"111", "hash": "11"}',
118 onboardingState="CREATED",
119 operationalState="DISABLED",
120 usageState="NOT_IN_USE",
121 userDefinedData='{"a": "A"}'
123 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
126 "vnfdId": "zte-hss-1.0",
127 "vnfProductName": "hss",
128 "vnfSoftwareVersion": "1.0.0",
129 "vnfdVersion": "1.0.0",
130 "checksum": {"algorithm": "111", "hash": "11"},
131 "softwareImages": None,
132 "additionalArtifacts": None,
133 "onboardingState": "CREATED",
134 "operationalState": "DISABLED",
135 "usageState": "NOT_IN_USE",
136 "userDefinedData": {"a": "A"},
139 self.assertEqual(response.data, expect_data)
140 self.assertEqual(response.status_code, status.HTTP_200_OK)
142 def test_query_single_vnf_failed(self):
143 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
144 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
146 def test_query_multiple_vnf(self):
147 VnfPackageModel.objects.create(
149 vnfdId="zte-hss-1.0",
151 vnfdProductName="hss",
152 vnfSoftwareVersion="1.0.0",
154 checksum='{"algorithm":"111", "hash": "11"}',
155 onboardingState="CREATED",
156 operationalState="DISABLED",
157 usageState="NOT_IN_USE",
158 userDefinedData='{"a": "A"}'
160 VnfPackageModel.objects.create(
162 vnfdId="zte-hss-1.0",
164 vnfdProductName="hss",
165 vnfSoftwareVersion="1.0.0",
167 checksum='{"algorithm":"111", "hash": "11"}',
168 onboardingState="CREATED",
169 operationalState="DISABLED",
170 usageState="NOT_IN_USE",
171 userDefinedData='{"a": "A"}'
173 response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
177 "vnfdId": "zte-hss-1.0",
178 "vnfProductName": "hss",
179 "vnfSoftwareVersion": "1.0.0",
180 "vnfdVersion": "1.0.0",
181 "checksum": {"algorithm": "111", "hash": "11"},
182 "softwareImages": None,
183 "additionalArtifacts": None,
184 "onboardingState": "CREATED",
185 "operationalState": "DISABLED",
186 "usageState": "NOT_IN_USE",
187 "userDefinedData": {"a": "A"},
192 "vnfdId": "zte-hss-1.0",
193 "vnfProductName": "hss",
194 "vnfSoftwareVersion": "1.0.0",
195 "vnfdVersion": "1.0.0",
196 "checksum": {"algorithm": "111", "hash": "11"},
197 "softwareImages": None,
198 "additionalArtifacts": None,
199 "onboardingState": "CREATED",
200 "operationalState": "DISABLED",
201 "usageState": "NOT_IN_USE",
202 "userDefinedData": {"a": "A"},
206 self.assertEqual(response.data, expect_data)
207 self.assertEqual(response.status_code, status.HTTP_200_OK)
209 def test_delete_single_vnf_pkg(self):
210 VnfPackageModel.objects.create(
212 vnfdId="zte-hss-1.0",
214 vnfdProductName="hss",
215 vnfSoftwareVersion="1.0.0",
217 checksum='{"algorithm":"111", "hash": "11"}',
218 onboardingState="CREATED",
219 operationalState="DISABLED",
220 usageState="NOT_IN_USE",
221 userDefinedData='{"a": "A"}'
223 response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
224 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
225 self.assertEqual(response.data, None)
227 def test_delete_when_vnf_pkg_not_exist(self):
228 response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
229 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
230 self.assertEqual(response.data, None)
232 def test_fetch_vnf_pkg(self):
233 with open("vnfPackage.csar", "wb") as fp:
234 fp.writelines("AAAABBBBCCCCDDDD")
235 VnfPackageModel.objects.create(
237 onboardingState="ONBOARDED",
238 localFilePath="vnfPackage.csar"
240 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
242 for data in response.streaming_content:
243 file_content = file_content + data
244 self.assertEqual(response.status_code, status.HTTP_200_OK)
245 self.assertEqual('AAAABBBBCCCCDDDD', file_content)
246 os.remove("vnfPackage.csar")
248 def test_fetch_partical_vnf_pkg(self):
249 with open("vnfPackage.csar", "wb") as fp:
250 fp.writelines("AAAABBBBCCCCDDDD")
251 VnfPackageModel.objects.create(
253 onboardingState="ONBOARDED",
254 localFilePath="vnfPackage.csar"
256 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE="4-7")
257 partial_file_content = ''
258 for data in response.streaming_content:
259 partial_file_content = partial_file_content + data
260 self.assertEqual(response.status_code, status.HTTP_200_OK)
261 self.assertEqual('BBB', partial_file_content)
262 os.remove("vnfPackage.csar")
264 def test_fetch_last_partical_vnf_pkg(self):
265 with open("vnfPackage.csar", "wb") as fp:
266 fp.writelines("AAAABBBBCCCCDDDD")
267 VnfPackageModel.objects.create(
269 onboardingState="ONBOARDED",
270 localFilePath="vnfPackage.csar"
272 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE=" 4-")
273 partial_file_content = ''
274 for data in response.streaming_content:
275 partial_file_content = partial_file_content + data
276 self.assertEqual(response.status_code, status.HTTP_200_OK)
277 self.assertEqual('BBBBCCCCDDDD', partial_file_content)
278 os.remove("vnfPackage.csar")
280 def test_fetch_vnf_pkg_when_pkg_not_exist(self):
281 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
282 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
284 def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
285 VnfPackageModel.objects.create(
287 onboardingState="CREATED",
288 localFilePath="vnfPackage.csar"
290 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
291 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
293 @mock.patch.object(VnfPackage, "create_vnf_pkg")
294 def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
295 mock_create_vnf_pkg.side_effect = TypeError('integer type')
297 "userDefinedData": {"a": "A"}
299 response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
300 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
302 @mock.patch.object(VnfPackage, "delete_vnf_pkg")
303 def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
304 mock_delete_vnf_pkg.side_effect = TypeError("integer type")
305 response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
306 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
308 @mock.patch.object(VnfPackage, "query_single")
309 def test_query_single_when_catch_exception(self, mock_query_single):
310 mock_query_single.side_effect = TypeError("integer type")
311 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
312 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
314 @mock.patch.object(VnfPackage, "query_multiple")
315 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
316 mock_query_muitiple.side_effect = TypeError("integer type")
317 response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
318 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
320 @mock.patch.object(toscaparsers, 'parse_vnfd')
321 def test_upload_when_catch_exception(self, mock_parse_vnfd):
322 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "empty.txt"), "rb")}
323 VnfPackageModel.objects.create(
325 onboardingState="CREATED"
327 mock_parse_vnfd.side_effect = TypeError("integer type")
328 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
329 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
331 @mock.patch.object(VnfPkgUploadThread, 'start')
332 def test_upload_from_uri_when_catch_exception(self, mock_start):
333 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
334 mock_start.side_effect = TypeError("integer type")
335 response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
336 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
338 @mock.patch.object(VnfPackage, 'download')
339 def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
340 mock_download.side_effect = TypeError("integer type")
341 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
342 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
344 @mock.patch.object(toscaparsers, 'parse_vnfd')
345 def test_fetch_vnf_artifact(self, mock_parse_vnfd):
346 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
347 VnfPackageModel.objects.create(
349 onboardingState="CREATED"
351 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
352 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
353 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
354 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image")
355 self.assertEqual(response.status_code, status.HTTP_200_OK)
356 self.assertEqual(response.getvalue(), "ubuntu_16.04\n")
358 @mock.patch.object(toscaparsers, 'parse_vnfd')
359 def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
360 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
361 VnfPackageModel.objects.create(
363 onboardingState="CREATED"
365 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
366 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
367 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
368 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/1451/artifacts/image")
369 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
371 @mock.patch.object(toscaparsers, 'parse_vnfd')
372 def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
373 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
374 VnfPackageModel.objects.create(
376 onboardingState="CREATED"
378 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
379 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
380 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
381 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image1")
382 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)