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 get asset(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="b632bddc-bccd-4180-bd8d-4e8a9578eff7").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(b632bddc-bccd-4180-bd8d-4e8a9578eff7) 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(restcall, 'call_req')
123 def test_ns_when_not_distributed_by_sdc(self, mock_call_req):
124 mock_call_req.return_value = [0, json.JSONEncoder().encode({
126 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
127 "distributionStatus": "DISTRIBUTION_APPROVED"
129 resp = self.client.post(
130 "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
131 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
132 self.assertEqual("failed", resp.data["status"])
134 "The asset (services,1) is not distributed from sdc.",
135 resp.data["statusDescription"])
137 ##########################################################################
139 def test_ns_pkg_normal_delete(self):
140 NSPackageModel(nsPackageId="8", nsdId="2").save()
141 resp = self.client.delete("/api/catalog/v1/nspackages/8")
142 self.assertEqual(resp.status_code, status.HTTP_200_OK)
143 self.assertEqual("success", resp.data["status"])
145 "Delete CSAR(8) successfully.",
146 resp.data["statusDescription"])
148 def test_ns_pkg_get_all(self):
154 nsPackageUri="13.csar",
161 nsPackageUri="14.csar",
163 resp = self.client.get("/api/catalog/v1/nspackages")
164 self.assertEqual(resp.status_code, status.HTTP_200_OK)
165 expect_data = [{"csarId": "13",
166 "packageInfo": {"csarName": "13.csar",
170 "downloadUrl": "http://127.0.0.1:8806/static/catalog/13/13.csar",
173 "nsdInvariantId": None
176 "packageInfo": {"csarName": "14.csar",
180 "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
183 "nsdInvariantId": None}}]
184 self.assertEqual(expect_data, resp.data)
186 def test_ns_pkg_get_one(self):
192 nsPackageUri="14.csar",
194 resp = self.client.get("/api/catalog/v1/nspackages/14")
195 self.assertEqual(resp.status_code, status.HTTP_200_OK)
203 "csarName": "14.csar",
205 "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
206 "nsdInvariantId": None}}
207 self.assertEqual(expect_data, resp.data)
209 def test_ns_pkg_get_one_not_found(self):
210 resp = self.client.get("/api/catalog/v1/nspackages/22")
211 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
213 {"error": "Ns package[22] not Found."},
216 ##########################################################################
218 @mock.patch.object(toscaparser, 'parse_nsd')
219 def test_nsd_parse_normal(self, mock_parse_nsd):
220 NSPackageModel(nsPackageId="18", nsdId="12").save()
221 mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
222 req_data = {"csarId": "18", "inputs": []}
223 resp = self.client.post(
224 "/api/parser/v1/parsernsd",
227 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
228 self.assertEqual({"model": '{"a": "b"}'}, resp.data)
230 def test_nsd_parse_when_csar_not_exist(self):
231 req_data = {"csarId": "1", "inputs": []}
232 resp = self.client.post(
233 "/api/parser/v1/parsernsd",
238 status.HTTP_500_INTERNAL_SERVER_ERROR)
239 self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."})