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
26 class TestNsPackage(TestCase):
28 self.client = Client()
29 NSPackageModel.objects.filter().delete()
30 VnfPackageModel.objects.filter().delete()
31 self.nsd_data = {"vnffgs": [{"vnffg_id": "vnffg1",
35 "properties": {"vendor": "zte",
36 "connection_point": ["m6000_data_in",
40 "constituent_vnfs": ["VFW",
42 "number_of_endpoints": 3,
43 "dependent_virtual_link": ["sfc_data_network",
46 "inputs": {"sfc_data_network": {"type": "string",
47 "value": "sfc_data_network"},
48 "externalDataNetworkName": {"type": "string",
49 "value": "vlan_4004_tunnel_net"},
50 "externalManageNetworkName": {"type": "string",
51 "value": "vlan_4008_mng_net"},
52 "NatIpRange": {"type": "string",
53 "value": "192.167.0.10-192.168.0.20"},
54 "externalPluginManageNetworkName": {"type": "string",
55 "value": "vlan_4007_plugin_net"}},
56 "pnfs": [{"pnf_id": "m6000_s",
59 "properties": {"vendor": "zte",
60 "request_reclassification": False,
63 "management_address": "111111",
65 "nsh_aware": False}}],
66 "fps": [{"properties": {"symmetric": False,
67 "policy": {"type": "ACL",
68 "criteria": {"dest_port_range": "1-100",
70 "source_ip_range": ["119.1.1.1-119.1.1.10"],
71 "dest_ip_range": [{"get_input": "NatIpRange"}],
73 "source_port_range": "1-100"}}},
74 "forwarder_list": [{"capability": "",
76 "node_name": "m6000_data_out"},
79 "node_name": "m600_tunnel_cp"},
80 {"capability": "vnat_fw_inout",
82 "node_name": "VNAT"}],
85 {"properties": {"symmetric": True,
86 "policy": {"type": "ACL",
87 "criteria": {"dest_port_range": "1-100",
89 "source_ip_range": ["1-100"],
90 "dest_ip_range": ["1-100"],
92 "source_port_range": "1-100"}}},
93 "forwarder_list": [{"capability": "",
95 "node_name": "m6000_data_in"},
98 "node_name": "m600_tunnel_cp"},
99 {"capability": "vfw_fw_inout",
102 {"capability": "vnat_fw_inout",
104 "node_name": "VNAT"},
107 "node_name": "m600_tunnel_cp"},
110 "node_name": "m6000_data_out"}],
114 "vnfs": [{"vnf_id": "VFW",
116 "properties": {"plugin_info": "vbrasplugin_1.0",
119 "adjust_vnf_capacity": True,
121 "vnf_extend_type": "driver",
122 "csarVersion": "v1.0",
124 "csarProvider": "ZTE",
129 "vmnumber_overquota_alarm": True,
130 "vnfd_version": "1.0.0",
131 "externalPluginManageNetworkName": "vlan_4007_plugin_net",
132 "id": "vcpe_vfw_zte_1_0",
133 "request_reclassification": False},
134 "dependencies": [{"key_name": "vfw_ctrl_by_manager_cp",
135 "vl_id": "ext_mnet_net"},
136 {"key_name": "vfw_data_cp",
137 "vl_id": "sfc_data_network"}],
138 "type": "tosca.nodes.nfv.ext.zte.VNF.VFW",
140 "ns_exposed": {"external_cps": [],
142 "policies": [{"file_url": "policies/abc.drl",
144 "vls": [{"route_id": "",
145 "vl_id": "ext_mnet_net",
146 "route_external": False,
148 "properties": {"name": "vlan_4008_mng_net",
150 "location_info": {"tenant": "admin",
152 "availability_zone": "nova"},
154 "dhcp_enabled": True,
155 "network_name": "vlan_4008_mng_net",
156 "network_type": "vlan"}},
158 "vl_id": "ext_datanet_net",
159 "route_external": False,
161 "properties": {"name": "vlan_4004_tunnel_net",
163 "location_info": {"tenant": "admin",
165 "availability_zone": "nova"},
167 "dhcp_enabled": True,
168 "network_name": "vlan_4004_tunnel_net",
169 "network_type": "vlan"}},
171 "vl_id": "sfc_data_network",
172 "route_external": False,
174 "properties": {"name": "sfc_data_network",
175 "dhcp_enabled": True,
176 "is_predefined": False,
177 "location_info": {"tenant": "admin",
179 "availability_zone": "nova"},
182 "network_name": "sfc_data_network",
183 "network_type": "vlan"}}],
184 "cps": [{"pnf_id": "m6000_s",
187 "cp_id": "m6000_data_out",
188 "properties": {"direction": "bidirectional",
189 "vnic_type": "normal",
191 "mac_address": "11-22-33-22-11-44",
192 "interface_name": "xgei-0/4/1/5",
193 "ip_address": "176.1.1.2",
195 "sfc_encapsulation": "mac"}},
196 {"pnf_id": "m6000_s",
197 "vl_id": "ext_datanet_net",
199 "cp_id": "m600_tunnel_cp",
200 "properties": {"direction": "bidirectional",
201 "vnic_type": "normal",
203 "mac_address": "00-11-00-22-33-00",
204 "interface_name": "gei-0/4/0/13",
205 "ip_address": "191.167.100.5",
207 "sfc_encapsulation": "mac"}},
208 {"pnf_id": "m6000_s",
211 "cp_id": "m6000_data_in",
212 "properties": {"direction": "bidirectional",
213 "vnic_type": "normal",
215 "mac_address": "11-22-33-22-11-41",
216 "interface_name": "gei-0/4/0/7",
217 "ip_address": "1.1.1.1",
219 "sfc_encapsulation": "mac",
221 {"pnf_id": "m6000_s",
222 "vl_id": "ext_mnet_net",
224 "cp_id": "m600_mnt_cp",
225 "properties": {"direction": "bidirectional",
226 "vnic_type": "normal",
228 "mac_address": "00-11-00-22-33-11",
229 "interface_name": "gei-0/4/0/1",
230 "ip_address": "10.46.244.51",
232 "sfc_encapsulation": "mac",
234 "metadata": {"invariant_id": "vcpe_ns_sff_1",
236 "csarVersion": "v1.0",
238 "csarProvider": "ZTE",
242 "description": "vcpe_ns"},
245 "descriptor_id": "VCPE_NS",
249 "invariant_id": "vcpe_ns_sff_1"
257 def test_ns_pkg_distribute_when_ns_exists(self):
258 NSPackageModel(nsPackageId="1", nsdId="2").save()
259 resp = self.client.post(
260 "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
261 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
262 self.assertEqual("failed", resp.data["status"])
264 "NS CSAR(1) already exists.",
265 resp.data["statusDescription"])
267 @mock.patch.object(restcall, 'call_req')
268 def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req):
269 mock_call_req.return_value = [0, "[]", '200']
270 resp = self.client.post(
271 "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
272 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
273 self.assertEqual("failed", resp.data["status"])
275 "Failed to query artifact(services,1) from sdc.",
276 resp.data["statusDescription"])
278 @mock.patch.object(restcall, 'call_req')
279 @mock.patch.object(sdc, 'download_artifacts')
280 @mock.patch.object(toscaparsers, 'parse_nsd')
281 def test_ns_pkg_distribute_when_nsd_already_exists(
282 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
283 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
284 mock_download_artifacts.return_value = "/home/vcpe.csar"
285 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
287 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
288 "distributionStatus": "DISTRIBUTED"
290 NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
291 resp = self.client.post(
292 "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
293 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
294 self.assertEqual("failed", resp.data["status"])
296 "NSD(VCPE_NS) already exists.",
297 resp.data["statusDescription"])
299 @mock.patch.object(restcall, 'call_req')
300 @mock.patch.object(sdc, 'download_artifacts')
301 @mock.patch.object(toscaparsers, 'parse_nsd')
302 def test_ns_pkg_distribute_when_nf_not_distributed(
303 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
304 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
305 mock_download_artifacts.return_value = "/home/vcpe.csar"
306 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
308 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
309 "distributionStatus": "DISTRIBUTED",
311 resp = self.client.post(
312 "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
313 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
314 self.assertEqual("failed", resp.data["status"])
316 "VNF package(vcpe_vfw_zte_1_0) is not distributed.",
317 resp.data["statusDescription"])
319 @mock.patch.object(restcall, 'call_req')
320 @mock.patch.object(sdc, 'download_artifacts')
321 @mock.patch.object(toscaparsers, 'parse_nsd')
322 def test_ns_pkg_distribute_when_successfully(
323 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
324 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
325 mock_download_artifacts.return_value = "/home/vcpe.csar"
326 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
328 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
329 "distributionStatus": "DISTRIBUTED"
331 VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
332 PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
333 resp = self.client.post(
334 "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
335 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
336 self.assertEqual("success", resp.data["status"])
338 "CSAR(1) distributed successfully.",
339 resp.data["statusDescription"])
341 @mock.patch.object(sdc, 'get_artifacts')
342 def test_ns_when_not_distributed_by_sdc(self, mock_get_artifacts):
343 mock_get_artifacts.return_value = [{
345 "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
346 "name": "underlayvpn",
348 "toscaModelURL": "/sdc/v1/genericparser/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
350 "subCategory": "VolteVNF",
351 "resourceType": "VF",
352 "lifecycleState": "CERTIFIED",
353 "distributionStatus": "DISTRIBUTION_APPROVED",
354 "lastUpdaterUserId": "jh0003"
356 resp = self.client.post(
357 "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
358 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
359 self.assertEqual("failed", resp.data["status"])
361 "The artifact (services,1) is not distributed from sdc.",
362 resp.data["statusDescription"])
364 ##########################################################################
366 def test_ns_pkg_normal_delete(self):
367 NSPackageModel(nsPackageId="8", nsdId="2").save()
368 resp = self.client.delete("/api/genericparser/v1/nspackages/8")
369 self.assertEqual(resp.status_code, status.HTTP_200_OK)
370 self.assertEqual("success", resp.data["status"])
372 "Delete CSAR(8) successfully.",
373 resp.data["statusDescription"])
375 def test_ns_pkg_get_all(self):
381 nsPackageUri="13.csar",
388 nsPackageUri="14.csar",
390 resp = self.client.get("/api/genericparser/v1/nspackages")
391 self.assertEqual(resp.status_code, status.HTTP_200_OK)
392 expect_data = [{"csarId": "13",
393 "packageInfo": {"csarName": "13.csar",
397 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/13/13.csar",
400 "nsdInvariantId": None
403 "packageInfo": {"csarName": "14.csar",
407 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar",
410 "nsdInvariantId": None}}]
411 self.assertEqual(expect_data, resp.data)
413 def test_ns_pkg_get_one(self):
419 nsPackageUri="14.csar",
421 resp = self.client.get("/api/genericparser/v1/nspackages/14")
422 self.assertEqual(resp.status_code, status.HTTP_200_OK)
430 "csarName": "14.csar",
432 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar",
433 "nsdInvariantId": None}}
434 self.assertEqual(expect_data, resp.data)
436 def test_ns_pkg_get_one_not_found(self):
437 resp = self.client.get("/api/genericparser/v1/nspackages/22")
438 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
440 {"error": "Ns package[22] not Found."},
443 ##########################################################################
445 @mock.patch.object(toscaparsers, 'parse_nsd')
446 def test_nsd_parse_normal(self, mock_parse_nsd):
447 NSPackageModel(nsPackageId="18", nsdId="12").save()
448 mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
449 req_data = {"csarId": "18", "inputs": []}
450 resp = self.client.post(
451 "/api/genericparser/v1/parsernsd",
454 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
455 self.assertEqual({"model": '{"a": "b"}'}, resp.data)
457 def test_nsd_parse_when_csar_not_exist(self):
458 req_data = {"csarId": "1", "inputs": []}
459 resp = self.client.post(
460 "/api/genericparser/v1/parsernsd",
465 status.HTTP_500_INTERNAL_SERVER_ERROR)
466 self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."})