_links is not implemented in the response of the Query VNF API
[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.config import CATALOG_ROOT_PATH
30 from catalog.pub.database.models import VnfPackageModel
31 from catalog.pub.utils import toscaparser
32
33 VNF_BASE_URL = "/api/vnfpkgm/v1/vnf_packages"
34
35
36 class MockReq():
37     def read(self):
38         return "1"
39
40     def close(self):
41         pass
42
43
44 class TestVnfPackage(TestCase):
45     def setUp(self):
46         self.client = APIClient()
47
48     def tearDown(self):
49         file_path = os.path.join(CATALOG_ROOT_PATH, "222")
50         if os.path.exists(file_path):
51             shutil.rmtree(file_path)
52
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(
57             vnfPackageId="222",
58             onboardingState="CREATED"
59         )
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)
66
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(
70             vnfPackageId="222",
71         )
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)
74
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(
79             vnfPackageId="222",
80             onboardingState="CREATED"
81         )
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)
89
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)
94
95     @mock.patch.object(urllib.request, 'urlopen')
96     def test_upload_from_uri_failed(self, mock_urlopen):
97         vnf_pkg = VnfPackageModel.objects.create(
98             vnfPackageId="333",
99             onboardingState="CREATED"
100         )
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)
107
108     def test_create_vnf_pkg(self):
109         req_data = {
110             "userDefinedData": {"a": "A"}
111         }
112         response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
113         resp_data = json.loads(response.content)
114         expect_resp_data = {
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
121         }
122         self.assertEqual(expect_resp_data, resp_data)
123         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
124
125     def test_query_single_vnf(self):
126         VnfPackageModel.objects.create(
127             vnfPackageId="222",
128             vnfdId="zte-hss-1.0",
129             vnfVendor="zte",
130             vnfdProductName="hss",
131             vnfSoftwareVersion="1.0.0",
132             vnfdVersion="1.0.0",
133             checksum='{"algorithm":"111", "hash": "11"}',
134             onboardingState="CREATED",
135             operationalState="DISABLED",
136             usageState="NOT_IN_USE",
137             userDefinedData='{"a": "A"}'
138         )
139         response = self.client.get("%s/222" % VNF_BASE_URL)
140         expect_data = {
141             "id": "222",
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'}
156                        }
157         }
158         self.assertEqual(response.data, expect_data)
159         self.assertEqual(response.status_code, status.HTTP_200_OK)
160
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)
164
165     def test_query_multiple_vnf(self):
166         VnfPackageModel.objects.create(
167             vnfPackageId="111",
168             vnfdId="zte-hss-1.0",
169             vnfVendor="zte",
170             vnfdProductName="hss",
171             vnfSoftwareVersion="1.0.0",
172             vnfdVersion="1.0.0",
173             checksum='{"algorithm":"111", "hash": "11"}',
174             onboardingState="CREATED",
175             operationalState="DISABLED",
176             usageState="NOT_IN_USE",
177             userDefinedData='{"a": "A"}'
178         )
179         VnfPackageModel.objects.create(
180             vnfPackageId="222",
181             vnfdId="zte-hss-1.0",
182             vnfVendor="zte",
183             vnfdProductName="hss",
184             vnfSoftwareVersion="1.0.0",
185             vnfdVersion="1.0.0",
186             checksum='{"algorithm":"111", "hash": "11"}',
187             onboardingState="CREATED",
188             operationalState="DISABLED",
189             usageState="NOT_IN_USE",
190             userDefinedData='{"a": "A"}'
191         )
192         response = self.client.get(VNF_BASE_URL)
193         expect_data = [
194             {
195                 "id": "111",
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"},
207                 "_links": {
208                     "self": {
209                         "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/111"
210                     },
211                     "vnfd": {
212                         "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/111/vnfd"
213                     },
214                     "packageContent": {
215                         "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/111/package_content"
216                     }
217                 }
218             },
219             {
220                 "id": "222",
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'},
234                            'packageContent': {
235                                'href': 'http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/package_content'}}
236             }
237         ]
238         self.assertEqual(response.data, expect_data)
239         self.assertEqual(response.status_code, status.HTTP_200_OK)
240
241     def test_delete_single_vnf_pkg(self):
242         VnfPackageModel.objects.create(
243             vnfPackageId="222",
244             vnfdId="zte-hss-1.0",
245             vnfVendor="zte",
246             vnfdProductName="hss",
247             vnfSoftwareVersion="1.0.0",
248             vnfdVersion="1.0.0",
249             checksum='{"algorithm":"111", "hash": "11"}',
250             onboardingState="CREATED",
251             operationalState="DISABLED",
252             usageState="NOT_IN_USE",
253             userDefinedData='{"a": "A"}'
254         )
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)
258
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)
263
264     def test_fetch_vnf_pkg(self):
265         with open("vnfPackage.csar", "wt") as fp:
266             fp.writelines("AAAABBBBCCCCDDDD")
267         VnfPackageModel.objects.create(
268             vnfPackageId="222",
269             onboardingState="ONBOARDED",
270             localFilePath="vnfPackage.csar"
271         )
272         response = self.client.get(VNF_BASE_URL + "/222/package_content")
273         file_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")
279
280     def test_fetch_partical_vnf_pkg(self):
281         with open("vnfPackage.csar", "wt") as fp:
282             fp.writelines("AAAABBBBCCCCDDDD")
283         VnfPackageModel.objects.create(
284             vnfPackageId="222",
285             onboardingState="ONBOARDED",
286             localFilePath="vnfPackage.csar"
287         )
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")
295
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(
300             vnfPackageId="222",
301             onboardingState="ONBOARDED",
302             localFilePath="vnfPackage.csar"
303         )
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")
311
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)
315
316     def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
317         VnfPackageModel.objects.create(
318             vnfPackageId="222",
319             onboardingState="CREATED",
320             localFilePath="vnfPackage.csar"
321         )
322         response = self.client.get(VNF_BASE_URL + "/222/package_content")
323         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
324
325     def test_download_vnfd(self):
326         VnfPackageModel.objects.create(
327             vnfPackageId="222",
328             onboardingState="ONBOARDED",
329             localFilePath=os.path.join(CATALOG_ROOT_PATH, "resource_test.csar")
330         )
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")
338
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)
342
343     def test_download_vnfd_when_catch_cataloge_exception(self):
344         VnfPackageModel.objects.create(
345             vnfPackageId="222",
346             onboardingState="CREATED",
347             localFilePath="vnfPackage.csar"
348         )
349         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
350         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
351
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')
355         req_data = {
356             "userDefinedData": {"a": "A"}
357         }
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)
360
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)
366
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)
372
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)
378
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(
383             vnfPackageId="222",
384             onboardingState="CREATED"
385         )
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)
389
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)
396
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)
402
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(
407             vnfPackageId="222",
408             onboardingState="CREATED"
409         )
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")
416
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(
421             vnfPackageId="222",
422             onboardingState="CREATED"
423         )
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)
429
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(
434             vnfPackageId="222",
435             onboardingState="CREATED"
436         )
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)
442
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(
446             vnfPackageId="222",
447             onboardingState="CREATED"
448         )
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)
454         print(response.data)
455         self.assertEqual(response.status_code, status.HTTP_200_OK)
456         expact_response_data = {
457             "id": "222",
458             "vnfdId": "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
459             "vnfProductName": "vcpe_vgw",
460             "vnfSoftwareVersion": "1.0",
461             "vnfdVersion": "1.0",
462             "softwareImages": None,
463             "additionalArtifacts": [
464                 {
465                     "artifactPath": "MainServiceTemplate.yaml",
466                     "checksum": {
467                         "algorithm": "Null",
468                         "hash": "Null"
469                     }
470                 }
471             ],
472             "onboardingState": "ONBOARDED",
473             "operationalState": "ENABLED",
474             "usageState": "NOT_IN_USE",
475             "_links": {
476                 "self": {
477                     "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222"
478                 },
479                 "vnfd": {
480                     "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/vnfd"
481                 },
482                 "packageContent": {
483                     "href": "http://127.0.0.1:80/api/vnfpkgm/v1/vnf_packages/222/package_content"
484                 }
485             }
486         }
487         self.assertEqual(response.data, expact_response_data)