1 # Copyright 2017 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.
17 from rest_framework.test import APIClient
18 from django.test import TestCase
19 from rest_framework import status
20 from genericparser.packages.biz.sdc_vnf_package import NfDistributeThread, NfPkgDeleteThread
21 from genericparser.pub.database.models import JobStatusModel, JobModel
22 from genericparser.pub.database.models import VnfPackageModel
23 from genericparser.pub.msapi import sdc
24 from genericparser.pub.utils import restcall, toscaparsers
25 from .const import vnfd_data
28 class TestNfPackage(TestCase):
30 self.client = APIClient()
31 VnfPackageModel.objects.filter().delete()
32 JobModel.objects.filter().delete()
33 JobStatusModel.objects.filter().delete()
34 self.vnfd_data = vnfd_data
39 def assert_job_result(self, job_id, job_progress, job_detail):
40 jobs = JobStatusModel.objects.filter(
42 progress=job_progress,
44 self.assertEqual(1, len(jobs))
46 @mock.patch.object(NfDistributeThread, 'run')
47 def test_nf_pkg_distribute_normal(self, mock_run):
48 resp = self.client.post(
49 "/api/parser/v1/vnfpackages",
56 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
58 def test_nf_pkg_distribute_when_csar_already_exist(self):
61 vnfdId="vcpe_vfw_zte_1_0"
69 self.assert_job_result("2", 255, "NF CSAR(1) already exists.")
71 @mock.patch.object(restcall, 'call_req')
72 @mock.patch.object(sdc, 'download_artifacts')
73 @mock.patch.object(toscaparsers, 'parse_vnfd')
74 def test_nf_pkg_distribute_when_vnfd_already_exist(self,
76 mock_download_artifacts,
78 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
79 mock_download_artifacts.return_value = "/home/hss.csar"
80 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
82 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
84 VnfPackageModel(vnfPackageId="2",
85 vnfdId="zte-hss-1.0").save()
86 NfDistributeThread(csar_id="1",
90 self.assert_job_result("2", 255, "VNF package(zte-hss-1.0) already exists.")
92 @mock.patch.object(restcall, 'call_req')
93 @mock.patch.object(sdc, 'download_artifacts')
94 @mock.patch.object(toscaparsers, 'parse_vnfd')
95 def test_nf_pkg_distribute_successfully(self,
96 mock_parse_vnfd, mock_download_artifacts, mock_call_req):
97 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
98 mock_download_artifacts.return_value = "/home/hss.csar"
99 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
101 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
103 NfDistributeThread(csar_id="1",
107 self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
109 ###############################################################################################################
111 @mock.patch.object(NfPkgDeleteThread, 'run')
112 def test_nf_pkg_delete_normal(self, mock_run):
113 resp = self.client.delete("/api/parser/v1/vnfpackages/1")
114 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
116 def test_nf_pkg_normal_delete(self):
117 VnfPackageModel(vnfPackageId="2", vnfdId="vcpe_vfw_zte_1_0").save()
118 NfPkgDeleteThread(csar_id="2", job_id="2").run()
119 self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
121 def test_nf_pkg_get_all(self):
127 vnfSoftwareVersion='',
136 vnfSoftwareVersion='',
140 resp = self.client.get("/api/parser/v1/vnfpackages")
141 self.assertEqual(resp.status_code, status.HTTP_200_OK)
151 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/3/",
165 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/4/",
172 self.assertEqual(expect_data, resp.data)
174 def test_nf_pkg_get_one(self):
180 vnfSoftwareVersion='',
185 resp = self.client.get("/api/parser/v1/vnfpackages/4")
186 self.assertEqual(resp.status_code, status.HTTP_200_OK)
195 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/4/",
201 self.assertEqual(expect_data, resp.data)
203 def test_nf_pkg_get_one_failed(self):
204 VnfPackageModel(vnfPackageId="4",
208 vnfSoftwareVersion='',
212 resp = self.client.get("/api/parser/v1/vnfpackages/2")
213 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
214 self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
216 ###############################################################################################################
218 @mock.patch.object(toscaparsers, 'parse_vnfd')
219 def test_vnfd_parse_normal(self, mock_parse_vnfd):
220 VnfPackageModel(vnfPackageId="8", vnfdId="10").save()
221 mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
222 req_data = {"csarId": "8", "inputs": []}
223 resp = self.client.post("/api/parser/v1/parservnfd", req_data, format='json')
224 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
225 self.assertEqual({"model": '{"c": "d"}'}, resp.data)
227 def test_vnfd_parse_when_csar_not_exist(self):
228 req_data = {"csarId": "1", "inputs": []}
229 resp = self.client.post("/api/parser/v1/parservnfd", req_data, format='json')
230 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
231 self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})