289b600d42783c5bc49f96390a9c90a346810bf5
[modeling/etsicatalog.git] / catalog / packages / tests / test_vnf_package.py
1 # Copyright 2018 ZTE Corporation.
2 #
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
6 #
7 #         http://www.apache.org/licenses/LICENSE-2.0
8 #
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.
14
15 import json
16 import os
17 import shutil
18 import urllib
19 import zipfile
20
21 import mock
22 from django.test import TestCase
23 from rest_framework import status
24 from rest_framework.test import APIClient
25
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
33
34 VNF_BASE_URL = "/api/vnfpkgm/v1/vnf_packages"
35
36
37 class MockReq():
38     def read(self):
39         return "1"
40
41     def close(self):
42         pass
43
44
45 class TestVnfPackage(TestCase):
46     def setUp(self):
47         self.client = APIClient()
48
49     def tearDown(self):
50         file_path = os.path.join(CATALOG_ROOT_PATH, "222")
51         if os.path.exists(file_path):
52             shutil.rmtree(file_path)
53
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(
58             vnfPackageId="222",
59             onboardingState="CREATED"
60         )
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)
67
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(
71             vnfPackageId="222",
72         )
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)
75
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(
80             vnfPackageId="222",
81             onboardingState="CREATED"
82         )
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)
90
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)
95
96     @mock.patch.object(urllib.request, 'urlopen')
97     def test_upload_from_uri_failed(self, mock_urlopen):
98         vnf_pkg = VnfPackageModel.objects.create(
99             vnfPackageId="333",
100             onboardingState="CREATED"
101         )
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)
108
109     def test_create_vnf_pkg(self):
110         req_data = {
111             "userDefinedData": {"a": "A"}
112         }
113         response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
114         resp_data = json.loads(response.content)
115         expect_resp_data = {
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
122         }
123         self.assertEqual(expect_resp_data, resp_data)
124         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
125
126     def test_query_single_vnf(self):
127         VnfPackageModel.objects.create(
128             vnfPackageId="222",
129             vnfdId="zte-hss-1.0",
130             vnfVendor="zte",
131             vnfdProductName="hss",
132             vnfSoftwareVersion="1.0.0",
133             vnfdVersion="1.0.0",
134             checksum='{"algorithm":"111", "hash": "11"}',
135             onboardingState="CREATED",
136             operationalState="DISABLED",
137             usageState="NOT_IN_USE",
138             userDefinedData='{"a": "A"}'
139         )
140         response = self.client.get("%s/222" % VNF_BASE_URL)
141         expect_data = {
142             "id": "222",
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'},
155                        'vnfd': {
156                            'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222/vnfd'},
157                        'packageContent': {
158                            'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222/package_content'}
159                        }
160         }
161         self.assertEqual(response.data, expect_data)
162         self.assertEqual(response.status_code, status.HTTP_200_OK)
163
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)
167
168     def test_query_multiple_vnf(self):
169         VnfPackageModel.objects.create(
170             vnfPackageId="111",
171             vnfdId="zte-hss-1.0",
172             vnfVendor="zte",
173             vnfdProductName="hss",
174             vnfSoftwareVersion="1.0.0",
175             vnfdVersion="1.0.0",
176             checksum='{"algorithm":"111", "hash": "11"}',
177             onboardingState="CREATED",
178             operationalState="DISABLED",
179             usageState="NOT_IN_USE",
180             userDefinedData='{"a": "A"}'
181         )
182         VnfPackageModel.objects.create(
183             vnfPackageId="222",
184             vnfdId="zte-hss-1.0",
185             vnfVendor="zte",
186             vnfdProductName="hss",
187             vnfSoftwareVersion="1.0.0",
188             vnfdVersion="1.0.0",
189             checksum='{"algorithm":"111", "hash": "11"}',
190             onboardingState="CREATED",
191             operationalState="DISABLED",
192             usageState="NOT_IN_USE",
193             userDefinedData='{"a": "A"}'
194         )
195         response = self.client.get(VNF_BASE_URL)
196         expect_data = [
197             {
198                 "id": "111",
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"},
210                 "_links": {
211                     "self": {
212                         "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/111"
213                     },
214                     "vnfd": {
215                         "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/111/vnfd"
216                     },
217                     "packageContent": {
218                         "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/111/package_content"
219                     }
220                 }
221             },
222             {
223                 "id": "222",
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'},
237                            'packageContent': {
238                                'href': config.MSB_BASE_URL + '/api/vnfpkgm/v1/vnf_packages/222/package_content'}}
239             }
240         ]
241         self.assertEqual(response.data, expect_data)
242         self.assertEqual(response.status_code, status.HTTP_200_OK)
243
244     def test_delete_single_vnf_pkg(self):
245         VnfPackageModel.objects.create(
246             vnfPackageId="222",
247             vnfdId="zte-hss-1.0",
248             vnfVendor="zte",
249             vnfdProductName="hss",
250             vnfSoftwareVersion="1.0.0",
251             vnfdVersion="1.0.0",
252             checksum='{"algorithm":"111", "hash": "11"}',
253             onboardingState="CREATED",
254             operationalState="DISABLED",
255             usageState="NOT_IN_USE",
256             userDefinedData='{"a": "A"}'
257         )
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)
261
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)
266
267     def test_fetch_vnf_pkg(self):
268         with open("vnfPackage.csar", "wt") as fp:
269             fp.writelines("AAAABBBBCCCCDDDD")
270         VnfPackageModel.objects.create(
271             vnfPackageId="222",
272             onboardingState="ONBOARDED",
273             localFilePath="vnfPackage.csar"
274         )
275         response = self.client.get(VNF_BASE_URL + "/222/package_content")
276         file_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")
282
283     def test_fetch_partical_vnf_pkg(self):
284         with open("vnfPackage.csar", "wt") as fp:
285             fp.writelines("AAAABBBBCCCCDDDD")
286         VnfPackageModel.objects.create(
287             vnfPackageId="222",
288             onboardingState="ONBOARDED",
289             localFilePath="vnfPackage.csar"
290         )
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")
298
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(
303             vnfPackageId="222",
304             onboardingState="ONBOARDED",
305             localFilePath="vnfPackage.csar"
306         )
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")
314
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)
318
319     def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
320         VnfPackageModel.objects.create(
321             vnfPackageId="222",
322             onboardingState="CREATED",
323             localFilePath="vnfPackage.csar"
324         )
325         response = self.client.get(VNF_BASE_URL + "/222/package_content")
326         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
327
328     def test_download_vnfd(self):
329         VnfPackageModel.objects.create(
330             vnfPackageId="222",
331             onboardingState="ONBOARDED",
332             localFilePath=os.path.join(CATALOG_ROOT_PATH, "resource_test.csar")
333         )
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")
341
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)
345
346     def test_download_vnfd_when_catch_cataloge_exception(self):
347         VnfPackageModel.objects.create(
348             vnfPackageId="222",
349             onboardingState="CREATED",
350             localFilePath="vnfPackage.csar"
351         )
352         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
353         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
354
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')
358         req_data = {
359             "userDefinedData": {"a": "A"}
360         }
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)
363
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)
369
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)
375
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)
381
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(
386             vnfPackageId="222",
387             onboardingState="CREATED"
388         )
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)
392
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)
399
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)
405
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(
410             vnfPackageId="222",
411             onboardingState="CREATED"
412         )
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")
419
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(
424             vnfPackageId="222",
425             onboardingState="CREATED"
426         )
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)
432
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(
437             vnfPackageId="222",
438             onboardingState="CREATED"
439         )
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)
445
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(
449             vnfPackageId="222",
450             onboardingState="CREATED"
451         )
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)
457         print(response.data)
458         self.assertEqual(response.status_code, status.HTTP_200_OK)
459         expact_response_data = {
460             "id": "222",
461             "vnfdId": "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
462             "vnfProductName": "vcpe_vgw",
463             "vnfSoftwareVersion": "1.0",
464             "vnfdVersion": "1.0",
465             "softwareImages": None,
466             "additionalArtifacts": [
467                 {
468                     "artifactPath": "MainServiceTemplate.yaml",
469                     "checksum": {
470                         "algorithm": "Null",
471                         "hash": "Null"
472                     }
473                 }
474             ],
475             "onboardingState": "ONBOARDED",
476             "operationalState": "ENABLED",
477             "usageState": "NOT_IN_USE",
478             "_links": {
479                 "self": {
480                     "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/222"
481                 },
482                 "vnfd": {
483                     "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/222/vnfd"
484                 },
485                 "packageContent": {
486                     "href": config.MSB_BASE_URL + "/api/vnfpkgm/v1/vnf_packages/222/package_content"
487                 }
488             }
489         }
490         self.assertEqual(response.data, expact_response_data)