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"}}
255 def test_ns_pkg_distribute_when_ns_exists(self):
256 NSPackageModel(nsPackageId="1", nsdId="2").save()
257 resp = self.client.post(
258 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
259 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
260 self.assertEqual("failed", resp.data["status"])
262 "NS CSAR(1) already exists.",
263 resp.data["statusDescription"])
265 @mock.patch.object(restcall, 'call_req')
266 def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req):
267 mock_call_req.return_value = [0, "[]", '200']
268 resp = self.client.post(
269 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
270 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
271 self.assertEqual("failed", resp.data["status"])
273 "Failed to query artifact(services,1) from sdc.",
274 resp.data["statusDescription"])
276 @mock.patch.object(restcall, 'call_req')
277 @mock.patch.object(sdc, 'download_artifacts')
278 @mock.patch.object(toscaparsers, 'parse_nsd')
279 def test_ns_pkg_distribute_when_nsd_already_exists(
280 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
281 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
282 mock_download_artifacts.return_value = "/home/vcpe.csar"
283 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
285 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
286 "distributionStatus": "DISTRIBUTED"
288 NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
289 resp = self.client.post(
290 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
291 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
292 self.assertEqual("failed", resp.data["status"])
294 "NSD(VCPE_NS) already exists.",
295 resp.data["statusDescription"])
297 @mock.patch.object(restcall, 'call_req')
298 @mock.patch.object(sdc, 'download_artifacts')
299 @mock.patch.object(toscaparsers, 'parse_nsd')
300 def test_ns_pkg_distribute_when_nf_not_distributed(
301 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
302 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
303 mock_download_artifacts.return_value = "/home/vcpe.csar"
304 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
306 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
307 "distributionStatus": "DISTRIBUTED",
309 resp = self.client.post(
310 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
311 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
312 self.assertEqual("failed", resp.data["status"])
314 "VNF package(vcpe_vfw_zte_1_0) is not distributed.",
315 resp.data["statusDescription"])
317 @mock.patch.object(restcall, 'call_req')
318 @mock.patch.object(sdc, 'download_artifacts')
319 @mock.patch.object(toscaparsers, 'parse_nsd')
320 def test_ns_pkg_distribute_when_successfully(
321 self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
322 mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
323 mock_download_artifacts.return_value = "/home/vcpe.csar"
324 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
326 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
327 "distributionStatus": "DISTRIBUTED"
329 VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
330 PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
331 resp = self.client.post(
332 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
333 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
334 self.assertEqual("success", resp.data["status"])
336 "CSAR(1) distributed successfully.",
337 resp.data["statusDescription"])
339 @mock.patch.object(sdc, 'get_artifacts')
340 def test_ns_when_not_distributed_by_sdc(self, mock_get_artifacts):
341 mock_get_artifacts.return_value = [{
343 "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
344 "name": "underlayvpn",
346 "toscaModelURL": "/sdc/v1/genericparser/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
348 "subCategory": "VolteVNF",
349 "resourceType": "VF",
350 "lifecycleState": "CERTIFIED",
351 "distributionStatus": "DISTRIBUTION_APPROVED",
352 "lastUpdaterUserId": "jh0003"
354 resp = self.client.post(
355 "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
356 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
357 self.assertEqual("failed", resp.data["status"])
359 "The artifact (services,1) is not distributed from sdc.",
360 resp.data["statusDescription"])
362 ##########################################################################
364 def test_ns_pkg_normal_delete(self):
365 NSPackageModel(nsPackageId="8", nsdId="2").save()
366 resp = self.client.delete("/api/parser/v1/nspackages/8")
367 self.assertEqual(resp.status_code, status.HTTP_200_OK)
368 self.assertEqual("success", resp.data["status"])
370 "Delete CSAR(8) successfully.",
371 resp.data["statusDescription"])
373 def test_ns_pkg_get_all(self):
379 nsPackageUri="13.csar",
386 nsPackageUri="14.csar",
388 resp = self.client.get("/api/parser/v1/nspackages")
389 self.assertEqual(resp.status_code, status.HTTP_200_OK)
390 expect_data = [{"csarId": "13",
391 "packageInfo": {"csarName": "13.csar",
395 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/13/13.csar",
398 "nsdInvariantId": None
401 "packageInfo": {"csarName": "14.csar",
405 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar",
408 "nsdInvariantId": None}}]
409 self.assertEqual(expect_data, resp.data)
411 def test_ns_pkg_get_one(self):
417 nsPackageUri="14.csar",
419 resp = self.client.get("/api/parser/v1/nspackages/14")
420 self.assertEqual(resp.status_code, status.HTTP_200_OK)
428 "csarName": "14.csar",
430 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar",
431 "nsdInvariantId": None}}
432 self.assertEqual(expect_data, resp.data)
434 def test_ns_pkg_get_one_not_found(self):
435 resp = self.client.get("/api/parser/v1/nspackages/22")
436 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
438 {"error": "Ns package[22] not Found."},
441 ##########################################################################
443 @mock.patch.object(toscaparsers, 'parse_nsd')
444 def test_nsd_parse_normal(self, mock_parse_nsd):
445 NSPackageModel(nsPackageId="18", nsdId="12").save()
446 mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
447 req_data = {"csarId": "18", "inputs": []}
448 resp = self.client.post(
449 "/api/parser/v1/parsernsd",
452 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
453 self.assertEqual({"model": '{"a": "b"}'}, resp.data)
455 def test_nsd_parse_when_csar_not_exist(self):
456 req_data = {"csarId": "1", "inputs": []}
457 resp = self.client.post(
458 "/api/parser/v1/parsernsd",
463 status.HTTP_500_INTERNAL_SERVER_ERROR)
464 self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."})