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 for key, value in expect_resp_data.items():
108 self.assertEqual(resp_data[key], value)
109 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
111 def test_query_single_vnf(self):
112 VnfPackageModel.objects.create(
114 vnfdId="zte-hss-1.0",
116 vnfdProductName="hss",
117 vnfSoftwareVersion="1.0.0",
119 checksum='{"algorithm":"111", "hash": "11"}',
120 onboardingState="CREATED",
121 operationalState="DISABLED",
122 usageState="NOT_IN_USE",
123 userDefinedData='{"a": "A"}'
125 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
128 "vnfdId": "zte-hss-1.0",
129 "vnfProductName": "hss",
130 "vnfSoftwareVersion": "1.0.0",
131 "vnfdVersion": "1.0.0",
132 "checksum": {"algorithm": "111", "hash": "11"},
133 "softwareImages": None,
134 "additionalArtifacts": None,
135 "onboardingState": "CREATED",
136 "operationalState": "DISABLED",
137 "usageState": "NOT_IN_USE",
138 "userDefinedData": {"a": "A"},
141 for key, value in expect_data.items():
142 self.assertEqual(response.data[key], value)
143 # self.assertEqual(response.data, expect_data)
144 self.assertEqual(response.status_code, status.HTTP_200_OK)
146 def test_query_single_vnf_failed(self):
147 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
148 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
150 def test_query_multiple_vnf(self):
151 VnfPackageModel.objects.create(
153 vnfdId="zte-hss-1.0",
155 vnfdProductName="hss",
156 vnfSoftwareVersion="1.0.0",
158 checksum='{"algorithm":"111", "hash": "11"}',
159 onboardingState="CREATED",
160 operationalState="DISABLED",
161 usageState="NOT_IN_USE",
162 userDefinedData='{"a": "A"}'
164 VnfPackageModel.objects.create(
166 vnfdId="zte-hss-1.0",
168 vnfdProductName="hss",
169 vnfSoftwareVersion="1.0.0",
171 checksum='{"algorithm":"111", "hash": "11"}',
172 onboardingState="CREATED",
173 operationalState="DISABLED",
174 usageState="NOT_IN_USE",
175 userDefinedData='{"a": "A"}'
177 response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
181 "vnfdId": "zte-hss-1.0",
182 "vnfProductName": "hss",
183 "vnfSoftwareVersion": "1.0.0",
184 "vnfdVersion": "1.0.0",
185 "checksum": {"algorithm": "111", "hash": "11"},
186 "softwareImages": None,
187 "additionalArtifacts": None,
188 "onboardingState": "CREATED",
189 "operationalState": "DISABLED",
190 "usageState": "NOT_IN_USE",
191 "userDefinedData": {"a": "A"},
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"},
210 # self.assertEqual(response.data, expect_data)
211 self.assertEqual(len(response.data), len(expect_data))
212 for i in range(0, len(expect_data)):
213 for key, value in expect_data[i].items():
214 self.assertEqual(response.data[i][key], value)
216 self.assertEqual(response.status_code, status.HTTP_200_OK)
218 def test_delete_single_vnf_pkg(self):
219 VnfPackageModel.objects.create(
221 vnfdId="zte-hss-1.0",
223 vnfdProductName="hss",
224 vnfSoftwareVersion="1.0.0",
226 checksum='{"algorithm":"111", "hash": "11"}',
227 onboardingState="CREATED",
228 operationalState="DISABLED",
229 usageState="NOT_IN_USE",
230 userDefinedData='{"a": "A"}'
232 response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
233 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
234 self.assertEqual(response.data, None)
236 def test_delete_when_vnf_pkg_not_exist(self):
237 response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
238 self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
239 self.assertEqual(response.data, None)
241 def test_fetch_vnf_pkg(self):
242 with open("vnfPackage.csar", "wb") as fp:
243 fp.writelines("AAAABBBBCCCCDDDD")
244 VnfPackageModel.objects.create(
246 onboardingState="ONBOARDED",
247 localFilePath="vnfPackage.csar"
249 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
251 for data in response.streaming_content:
252 file_content = file_content + data
253 self.assertEqual(response.status_code, status.HTTP_200_OK)
254 self.assertEqual('AAAABBBBCCCCDDDD', file_content)
255 os.remove("vnfPackage.csar")
257 def test_fetch_partical_vnf_pkg(self):
258 with open("vnfPackage.csar", "wb") as fp:
259 fp.writelines("AAAABBBBCCCCDDDD")
260 VnfPackageModel.objects.create(
262 onboardingState="ONBOARDED",
263 localFilePath="vnfPackage.csar"
265 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE="4-7")
266 partial_file_content = ''
267 for data in response.streaming_content:
268 partial_file_content = partial_file_content + data
269 self.assertEqual(response.status_code, status.HTTP_200_OK)
270 self.assertEqual('BBB', partial_file_content)
271 os.remove("vnfPackage.csar")
273 def test_fetch_last_partical_vnf_pkg(self):
274 with open("vnfPackage.csar", "wb") as fp:
275 fp.writelines("AAAABBBBCCCCDDDD")
276 VnfPackageModel.objects.create(
278 onboardingState="ONBOARDED",
279 localFilePath="vnfPackage.csar"
281 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE=" 4-")
282 partial_file_content = ''
283 for data in response.streaming_content:
284 partial_file_content = partial_file_content + data
285 self.assertEqual(response.status_code, status.HTTP_200_OK)
286 self.assertEqual('BBBBCCCCDDDD', partial_file_content)
287 os.remove("vnfPackage.csar")
289 def test_fetch_vnf_pkg_when_pkg_not_exist(self):
290 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
291 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
293 def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
294 VnfPackageModel.objects.create(
296 onboardingState="CREATED",
297 localFilePath="vnfPackage.csar"
299 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
300 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
302 @mock.patch.object(VnfPackage, "create_vnf_pkg")
303 def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
304 mock_create_vnf_pkg.side_effect = TypeError('integer type')
306 "userDefinedData": {"a": "A"}
308 response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
309 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
311 @mock.patch.object(VnfPackage, "delete_vnf_pkg")
312 def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
313 mock_delete_vnf_pkg.side_effect = TypeError("integer type")
314 response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
315 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
317 @mock.patch.object(VnfPackage, "query_single")
318 def test_query_single_when_catch_exception(self, mock_query_single):
319 mock_query_single.side_effect = TypeError("integer type")
320 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
321 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
323 @mock.patch.object(VnfPackage, "query_multiple")
324 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
325 mock_query_muitiple.side_effect = TypeError("integer type")
326 response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
327 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
329 @mock.patch.object(toscaparsers, 'parse_vnfd')
330 def test_upload_when_catch_exception(self, mock_parse_vnfd):
331 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "empty.txt"), "rb")}
332 VnfPackageModel.objects.create(
334 onboardingState="CREATED"
336 mock_parse_vnfd.side_effect = TypeError("integer type")
337 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
338 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
340 @mock.patch.object(VnfPkgUploadThread, 'start')
341 def test_upload_from_uri_when_catch_exception(self, mock_start):
342 req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
343 mock_start.side_effect = TypeError("integer type")
344 response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
345 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
347 @mock.patch.object(VnfPackage, 'download')
348 def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
349 mock_download.side_effect = TypeError("integer type")
350 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
351 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
353 @mock.patch.object(toscaparsers, 'parse_vnfd')
354 def test_fetch_vnf_artifact(self, mock_parse_vnfd):
355 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
356 VnfPackageModel.objects.create(
358 onboardingState="CREATED"
360 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
361 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
362 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
363 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image")
364 self.assertEqual(response.status_code, status.HTTP_200_OK)
365 self.assertEqual(response.getvalue(), "ubuntu_16.04\n")
367 @mock.patch.object(toscaparsers, 'parse_vnfd')
368 def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
369 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
370 VnfPackageModel.objects.create(
372 onboardingState="CREATED"
374 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
375 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
376 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
377 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/1451/artifacts/image")
378 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
380 @mock.patch.object(toscaparsers, 'parse_vnfd')
381 def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
382 data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
383 VnfPackageModel.objects.create(
385 onboardingState="CREATED"
387 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
388 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
389 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
390 response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image1")
391 self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)