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.
18 from django.test import Client
19 from django.test import TestCase
20 from rest_framework import status
22 from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
23 from catalog.pub.msapi import sdc
24 from catalog.pub.utils import restcall, toscaparser
25 from .const import nsd_data
28 class TestNsPackage(TestCase):
30 self.client = Client()
31 NSPackageModel.objects.filter().delete()
32 VnfPackageModel.objects.filter().delete()
33 self.nsd_data = nsd_data
38 def test_ns_pkg_distribute_when_ns_exists(self):
39 NSPackageModel(nsPackageId="1", nsdId="2").save()
40 resp = self.client.post(
41 "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
42 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
43 self.assertEqual("failed", resp.data["status"])
45 "NS CSAR(1) already exists.",
46 resp.data["statusDescription"])
48 @mock.patch.object(restcall, 'call_req')
49 def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req):
50 mock_call_req.return_value = [0, "[]", '200']
51 resp = self.client.post(
52 "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
53 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
54 self.assertEqual("failed", resp.data["status"])
56 "Failed to query artifact(services,1) from sdc.",
57 resp.data["statusDescription"])
59 @mock.patch.object(restcall, 'call_req')
60 @mock.patch.object(sdc, 'download_artifacts')
61 @mock.patch.object(toscaparser, 'parse_nsd')
62 def test_ns_pkg_distribute_when_nsd_already_exists(
63 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
64 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
65 mock_download_artifacts.return_value = "/home/vcpe.csar"
66 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
68 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
69 "distributionStatus": "DISTRIBUTED"
71 NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
72 resp = self.client.post(
73 "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
74 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
75 self.assertEqual("failed", resp.data["status"])
77 "NSD(VCPE_NS) already exists.",
78 resp.data["statusDescription"])
80 @mock.patch.object(restcall, 'call_req')
81 @mock.patch.object(sdc, 'download_artifacts')
82 @mock.patch.object(toscaparser, 'parse_nsd')
83 def test_ns_pkg_distribute_when_nf_not_distributed(
84 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
85 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
86 mock_download_artifacts.return_value = "/home/vcpe.csar"
87 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
89 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
90 "distributionStatus": "DISTRIBUTED",
92 resp = self.client.post(
93 "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
94 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
95 self.assertEqual("failed", resp.data["status"])
97 "VNF package(vcpe_vfw_zte_1_0) is not distributed.",
98 resp.data["statusDescription"])
100 @mock.patch.object(restcall, 'call_req')
101 @mock.patch.object(sdc, 'download_artifacts')
102 @mock.patch.object(toscaparser, 'parse_nsd')
103 def test_ns_pkg_distribute_when_successfully(
104 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
105 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
106 mock_download_artifacts.return_value = "/home/vcpe.csar"
107 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
109 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
110 "distributionStatus": "DISTRIBUTED"
112 VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
113 PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
114 resp = self.client.post(
115 "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
116 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
117 self.assertEqual("success", resp.data["status"])
119 "CSAR(1) distributed successfully.",
120 resp.data["statusDescription"])
122 @mock.patch.object(sdc, 'get_artifacts')
123 def test_ns_when_not_distributed_by_sdc(self, mock_get_artifacts):
124 mock_get_artifacts.return_value = [{
126 "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
127 "name": "underlayvpn",
129 "toscaModelURL": "/sdc/v1/catalog/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
131 "subCategory": "VolteVNF",
132 "resourceType": "VF",
133 "lifecycleState": "CERTIFIED",
134 "distributionStatus": "DISTRIBUTION_APPROVED",
135 "lastUpdaterUserId": "jh0003"
137 resp = self.client.post(
138 "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
139 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
140 self.assertEqual("failed", resp.data["status"])
142 "The artifact (services,1) is not distributed from sdc.",
143 resp.data["statusDescription"])
145 ##########################################################################
147 def test_ns_pkg_normal_delete(self):
148 NSPackageModel(nsPackageId="8", nsdId="2").save()
149 resp = self.client.delete("/api/catalog/v1/nspackages/8")
150 self.assertEqual(resp.status_code, status.HTTP_200_OK)
151 self.assertEqual("success", resp.data["status"])
153 "Delete CSAR(8) successfully.",
154 resp.data["statusDescription"])
156 def test_ns_pkg_get_all(self):
162 nsPackageUri="13.csar",
169 nsPackageUri="14.csar",
171 resp = self.client.get("/api/catalog/v1/nspackages")
172 self.assertEqual(resp.status_code, status.HTTP_200_OK)
173 expect_data = [{"csarId": "13",
174 "packageInfo": {"csarName": "13.csar",
178 "downloadUrl": "http://127.0.0.1:8806/static/catalog/13/13.csar",
181 "nsdInvariantId": None
184 "packageInfo": {"csarName": "14.csar",
188 "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
191 "nsdInvariantId": None}}]
192 self.assertEqual(expect_data, resp.data)
194 def test_ns_pkg_get_one(self):
200 nsPackageUri="14.csar",
202 resp = self.client.get("/api/catalog/v1/nspackages/14")
203 self.assertEqual(resp.status_code, status.HTTP_200_OK)
211 "csarName": "14.csar",
213 "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
214 "nsdInvariantId": None}}
215 self.assertEqual(expect_data, resp.data)
217 def test_ns_pkg_get_one_not_found(self):
218 resp = self.client.get("/api/catalog/v1/nspackages/22")
219 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
221 {"error": "Ns package[22] not Found."},
224 ##########################################################################
226 @mock.patch.object(toscaparser, 'parse_nsd')
227 def test_nsd_parse_normal(self, mock_parse_nsd):
228 NSPackageModel(nsPackageId="18", nsdId="12").save()
229 mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
230 req_data = {"csarId": "18", "inputs": []}
231 resp = self.client.post(
232 "/api/parser/v1/parsernsd",
235 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
236 self.assertEqual({"model": '{"a": "b"}'}, resp.data)
238 def test_nsd_parse_when_csar_not_exist(self):
239 req_data = {"csarId": "1", "inputs": []}
240 resp = self.client.post(
241 "/api/parser/v1/parsernsd",
246 status.HTTP_500_INTERNAL_SERVER_ERROR)
247 self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."})