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 import status
18 from django.test import TestCase
19 from django.test import Client
21 from genericparser.pub.utils import restcall, toscaparsers
22 from genericparser.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
23 from genericparser.pub.msapi import sdc
24 from .const import nsd_data
27 class TestNsPackage(TestCase):
29 self.client = Client()
30 NSPackageModel.objects.filter().delete()
31 VnfPackageModel.objects.filter().delete()
32 self.nsd_data = nsd_data
37 def test_ns_pkg_distribute_when_ns_exists(self):
38 NSPackageModel(nsPackageId="1", nsdId="2").save()
39 resp = self.client.post(
40 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
41 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
42 self.assertEqual("failed", resp.data["status"])
44 "NS CSAR(1) already exists.",
45 resp.data["statusDescription"])
47 @mock.patch.object(restcall, 'call_req')
48 def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req):
49 mock_call_req.return_value = [0, "[]", '200']
50 resp = self.client.post(
51 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
52 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
53 self.assertEqual("failed", resp.data["status"])
55 "Failed to query artifact(services,1) from sdc.",
56 resp.data["statusDescription"])
58 @mock.patch.object(restcall, 'call_req')
59 @mock.patch.object(sdc, 'download_artifacts')
60 @mock.patch.object(toscaparsers, 'parse_nsd')
61 def test_ns_pkg_distribute_when_nsd_already_exists(
62 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
63 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
64 mock_download_artifacts.return_value = "/home/vcpe.csar"
65 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
67 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
68 "distributionStatus": "DISTRIBUTED"
70 NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
71 resp = self.client.post(
72 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
73 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
74 self.assertEqual("failed", resp.data["status"])
76 "NSD(VCPE_NS) already exists.",
77 resp.data["statusDescription"])
79 @mock.patch.object(restcall, 'call_req')
80 @mock.patch.object(sdc, 'download_artifacts')
81 @mock.patch.object(toscaparsers, 'parse_nsd')
82 def test_ns_pkg_distribute_when_nf_not_distributed(
83 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
84 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
85 mock_download_artifacts.return_value = "/home/vcpe.csar"
86 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
88 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
89 "distributionStatus": "DISTRIBUTED",
91 resp = self.client.post(
92 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
93 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
94 self.assertEqual("failed", resp.data["status"])
96 "VNF package(vcpe_vfw_zte_1_0) is not distributed.",
97 resp.data["statusDescription"])
99 @mock.patch.object(restcall, 'call_req')
100 @mock.patch.object(sdc, 'download_artifacts')
101 @mock.patch.object(toscaparsers, 'parse_nsd')
102 def test_ns_pkg_distribute_when_successfully(
103 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
104 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
105 mock_download_artifacts.return_value = "/home/vcpe.csar"
106 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
108 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
109 "distributionStatus": "DISTRIBUTED"
111 VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
112 PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
113 resp = self.client.post(
114 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
115 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
116 self.assertEqual("success", resp.data["status"])
118 "CSAR(1) distributed successfully.",
119 resp.data["statusDescription"])
121 @mock.patch.object(sdc, 'get_artifacts')
122 def test_ns_when_not_distributed_by_sdc(self, mock_get_artifacts):
123 mock_get_artifacts.return_value = [{
125 "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
126 "name": "underlayvpn",
128 "toscaModelURL": "/sdc/v1/genericparser/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
130 "subCategory": "VolteVNF",
131 "resourceType": "VF",
132 "lifecycleState": "CERTIFIED",
133 "distributionStatus": "DISTRIBUTION_APPROVED",
134 "lastUpdaterUserId": "jh0003"
136 resp = self.client.post(
137 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
138 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
139 self.assertEqual("failed", resp.data["status"])
141 "The artifact (services,1) is not distributed from sdc.",
142 resp.data["statusDescription"])
144 ##########################################################################
146 def test_ns_pkg_normal_delete(self):
147 NSPackageModel(nsPackageId="8", nsdId="2").save()
148 resp = self.client.delete("/api/parser/v1/nspackages/8")
149 self.assertEqual(resp.status_code, status.HTTP_200_OK)
150 self.assertEqual("success", resp.data["status"])
152 "Delete CSAR(8) successfully.",
153 resp.data["statusDescription"])
155 def test_ns_pkg_get_all(self):
161 nsPackageUri="13.csar",
168 nsPackageUri="14.csar",
170 resp = self.client.get("/api/parser/v1/nspackages")
171 self.assertEqual(resp.status_code, status.HTTP_200_OK)
172 expect_data = [{"csarId": "13",
173 "packageInfo": {"csarName": "13.csar",
177 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/13/13.csar",
180 "nsdInvariantId": None
183 "packageInfo": {"csarName": "14.csar",
187 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar",
190 "nsdInvariantId": None}}]
191 self.assertEqual(expect_data, resp.data)
193 def test_ns_pkg_get_one(self):
199 nsPackageUri="14.csar",
201 resp = self.client.get("/api/parser/v1/nspackages/14")
202 self.assertEqual(resp.status_code, status.HTTP_200_OK)
210 "csarName": "14.csar",
212 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar",
213 "nsdInvariantId": None}}
214 self.assertEqual(expect_data, resp.data)
216 def test_ns_pkg_get_one_not_found(self):
217 resp = self.client.get("/api/parser/v1/nspackages/22")
218 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
220 {"error": "Ns package[22] not Found."},
223 ##########################################################################
225 @mock.patch.object(toscaparsers, 'parse_nsd')
226 def test_nsd_parse_normal(self, mock_parse_nsd):
227 NSPackageModel(nsPackageId="18", nsdId="12").save()
228 mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
229 req_data = {"csarId": "18", "inputs": []}
230 resp = self.client.post(
231 "/api/parser/v1/parsernsd",
234 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
235 self.assertEqual({"model": '{"a": "b"}'}, resp.data)
237 def test_nsd_parse_when_csar_not_exist(self):
238 req_data = {"csarId": "1", "inputs": []}
239 resp = self.client.post(
240 "/api/parser/v1/parsernsd",
245 status.HTTP_500_INTERNAL_SERVER_ERROR)
246 self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."})