diff options
Diffstat (limited to 'genericparser/packages/tests/test_nspackage.py')
| -rw-r--r-- | genericparser/packages/tests/test_nspackage.py | 466 |
1 files changed, 466 insertions, 0 deletions
diff --git a/genericparser/packages/tests/test_nspackage.py b/genericparser/packages/tests/test_nspackage.py new file mode 100644 index 0000000..f44488e --- /dev/null +++ b/genericparser/packages/tests/test_nspackage.py @@ -0,0 +1,466 @@ +# Copyright 2017 ZTE Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json +import mock +from rest_framework import status +from django.test import TestCase +from django.test import Client + +from genericparser.pub.utils import restcall, toscaparsers +from genericparser.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel +from genericparser.pub.msapi import sdc + + +class TestNsPackage(TestCase): + def setUp(self): + self.client = Client() + NSPackageModel.objects.filter().delete() + VnfPackageModel.objects.filter().delete() + self.nsd_data = {"vnffgs": [{"vnffg_id": "vnffg1", + "description": "", + "members": ["path1", + "path2"], + "properties": {"vendor": "zte", + "connection_point": ["m6000_data_in", + "m600_tunnel_cp", + "m6000_data_out"], + "version": "1.0", + "constituent_vnfs": ["VFW", + "VNAT"], + "number_of_endpoints": 3, + "dependent_virtual_link": ["sfc_data_network", + "ext_datanet_net", + "ext_mnet_net"]}}], + "inputs": {"sfc_data_network": {"type": "string", + "value": "sfc_data_network"}, + "externalDataNetworkName": {"type": "string", + "value": "vlan_4004_tunnel_net"}, + "externalManageNetworkName": {"type": "string", + "value": "vlan_4008_mng_net"}, + "NatIpRange": {"type": "string", + "value": "192.167.0.10-192.168.0.20"}, + "externalPluginManageNetworkName": {"type": "string", + "value": "vlan_4007_plugin_net"}}, + "pnfs": [{"pnf_id": "m6000_s", + "cps": [], + "description": "", + "properties": {"vendor": "zte", + "request_reclassification": False, + "pnf_type": "m6000s", + "version": "1.0", + "management_address": "111111", + "id": "m6000_s", + "nsh_aware": False}}], + "fps": [{"properties": {"symmetric": False, + "policy": {"type": "ACL", + "criteria": {"dest_port_range": "1-100", + "ip_protocol": "tcp", + "source_ip_range": ["119.1.1.1-119.1.1.10"], + "dest_ip_range": [{"get_input": "NatIpRange"}], + "dscp": 0, + "source_port_range": "1-100"}}}, + "forwarder_list": [{"capability": "", + "type": "cp", + "node_name": "m6000_data_out"}, + {"capability": "", + "type": "cp", + "node_name": "m600_tunnel_cp"}, + {"capability": "vnat_fw_inout", + "type": "vnf", + "node_name": "VNAT"}], + "description": "", + "fp_id": "path2"}, + {"properties": {"symmetric": True, + "policy": {"type": "ACL", + "criteria": {"dest_port_range": "1-100", + "ip_protocol": "tcp", + "source_ip_range": ["1-100"], + "dest_ip_range": ["1-100"], + "dscp": 4, + "source_port_range": "1-100"}}}, + "forwarder_list": [{"capability": "", + "type": "cp", + "node_name": "m6000_data_in"}, + {"capability": "", + "type": "cp", + "node_name": "m600_tunnel_cp"}, + {"capability": "vfw_fw_inout", + "type": "vnf", + "node_name": "VFW"}, + {"capability": "vnat_fw_inout", + "type": "vnf", + "node_name": "VNAT"}, + {"capability": "", + "type": "cp", + "node_name": "m600_tunnel_cp"}, + {"capability": "", + "type": "cp", + "node_name": "m6000_data_out"}], + "description": "", + "fp_id": "path1"}], + "routers": [], + "vnfs": [{"vnf_id": "VFW", + "description": "", + "properties": {"plugin_info": "vbrasplugin_1.0", + "vendor": "zte", + "is_shared": False, + "adjust_vnf_capacity": True, + "name": "VFW", + "vnf_extend_type": "driver", + "csarVersion": "v1.0", + "csarType": "NFAR", + "csarProvider": "ZTE", + "version": "1.0", + "nsh_aware": True, + "cross_dc": False, + "vnf_type": "VFW", + "vmnumber_overquota_alarm": True, + "vnfd_version": "1.0.0", + "externalPluginManageNetworkName": "vlan_4007_plugin_net", + "id": "vcpe_vfw_zte_1_0", + "request_reclassification": False}, + "dependencies": [{"key_name": "vfw_ctrl_by_manager_cp", + "vl_id": "ext_mnet_net"}, + {"key_name": "vfw_data_cp", + "vl_id": "sfc_data_network"}], + "type": "tosca.nodes.nfv.ext.zte.VNF.VFW", + "networks": []}], + "ns_exposed": {"external_cps": [], + "forward_cps": []}, + "policies": [{"file_url": "policies/abc.drl", + "name": "aaa"}], + "vls": [{"route_id": "", + "vl_id": "ext_mnet_net", + "route_external": False, + "description": "", + "properties": {"name": "vlan_4008_mng_net", + "mtu": 1500, + "location_info": {"tenant": "admin", + "vimid": 2, + "availability_zone": "nova"}, + "ip_version": 4, + "dhcp_enabled": True, + "network_name": "vlan_4008_mng_net", + "network_type": "vlan"}}, + {"route_id": "", + "vl_id": "ext_datanet_net", + "route_external": False, + "description": "", + "properties": {"name": "vlan_4004_tunnel_net", + "mtu": 1500, + "location_info": {"tenant": "admin", + "vimid": 2, + "availability_zone": "nova"}, + "ip_version": 4, + "dhcp_enabled": True, + "network_name": "vlan_4004_tunnel_net", + "network_type": "vlan"}}, + {"route_id": "", + "vl_id": "sfc_data_network", + "route_external": False, + "description": "", + "properties": {"name": "sfc_data_network", + "dhcp_enabled": True, + "is_predefined": False, + "location_info": {"tenant": "admin", + "vimid": 2, + "availability_zone": "nova"}, + "ip_version": 4, + "mtu": 1500, + "network_name": "sfc_data_network", + "network_type": "vlan"}}], + "cps": [{"pnf_id": "m6000_s", + "vl_id": "path2", + "description": "", + "cp_id": "m6000_data_out", + "properties": {"direction": "bidirectional", + "vnic_type": "normal", + "bandwidth": 0, + "mac_address": "11-22-33-22-11-44", + "interface_name": "xgei-0/4/1/5", + "ip_address": "176.1.1.2", + "order": 0, + "sfc_encapsulation": "mac"}}, + {"pnf_id": "m6000_s", + "vl_id": "ext_datanet_net", + "description": "", + "cp_id": "m600_tunnel_cp", + "properties": {"direction": "bidirectional", + "vnic_type": "normal", + "bandwidth": 0, + "mac_address": "00-11-00-22-33-00", + "interface_name": "gei-0/4/0/13", + "ip_address": "191.167.100.5", + "order": 0, + "sfc_encapsulation": "mac"}}, + {"pnf_id": "m6000_s", + "vl_id": "path2", + "description": "", + "cp_id": "m6000_data_in", + "properties": {"direction": "bidirectional", + "vnic_type": "normal", + "bandwidth": 0, + "mac_address": "11-22-33-22-11-41", + "interface_name": "gei-0/4/0/7", + "ip_address": "1.1.1.1", + "order": 0, + "sfc_encapsulation": "mac", + "bond": "none"}}, + {"pnf_id": "m6000_s", + "vl_id": "ext_mnet_net", + "description": "", + "cp_id": "m600_mnt_cp", + "properties": {"direction": "bidirectional", + "vnic_type": "normal", + "bandwidth": 0, + "mac_address": "00-11-00-22-33-11", + "interface_name": "gei-0/4/0/1", + "ip_address": "10.46.244.51", + "order": 0, + "sfc_encapsulation": "mac", + "bond": "none"}}], + "metadata": {"invariant_id": "vcpe_ns_sff_1", + "name": "VCPE_NS", + "csarVersion": "v1.0", + "csarType": "NSAR", + "csarProvider": "ZTE", + "version": 1, + "vendor": "ZTE", + "id": "VCPE_NS", + "description": "vcpe_ns"}, + "ns": { + "properties": { + "descriptor_id": "VCPE_NS", + "version": 1, + "name": "VCPE_NS", + "desginer": "ZTE", + "invariant_id": "vcpe_ns_sff_1" + } + } + } + + def tearDown(self): + pass + + def test_ns_pkg_distribute_when_ns_exists(self): + NSPackageModel(nsPackageId="1", nsdId="2").save() + resp = self.client.post( + "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json') + self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual("failed", resp.data["status"]) + self.assertEqual( + "NS CSAR(1) already exists.", + resp.data["statusDescription"]) + + @mock.patch.object(restcall, 'call_req') + def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req): + mock_call_req.return_value = [0, "[]", '200'] + resp = self.client.post( + "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json') + self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual("failed", resp.data["status"]) + self.assertEqual( + "Failed to query artifact(services,1) from sdc.", + resp.data["statusDescription"]) + + @mock.patch.object(restcall, 'call_req') + @mock.patch.object(sdc, 'download_artifacts') + @mock.patch.object(toscaparsers, 'parse_nsd') + def test_ns_pkg_distribute_when_nsd_already_exists( + self, mock_parse_nsd, mock_download_artifacts, mock_call_req): + mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data) + mock_download_artifacts.return_value = "/home/vcpe.csar" + mock_call_req.return_value = [0, json.JSONEncoder().encode([{ + "uuid": "1", + "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar", + "distributionStatus": "DISTRIBUTED" + }]), '200'] + NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save() + resp = self.client.post( + "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json') + self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual("failed", resp.data["status"]) + self.assertEqual( + "NSD(VCPE_NS) already exists.", + resp.data["statusDescription"]) + + @mock.patch.object(restcall, 'call_req') + @mock.patch.object(sdc, 'download_artifacts') + @mock.patch.object(toscaparsers, 'parse_nsd') + def test_ns_pkg_distribute_when_nf_not_distributed( + self, mock_parse_nsd, mock_download_artifacts, mock_call_req): + mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data) + mock_download_artifacts.return_value = "/home/vcpe.csar" + mock_call_req.return_value = [0, json.JSONEncoder().encode([{ + "uuid": "1", + "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar", + "distributionStatus": "DISTRIBUTED", + }]), '200'] + resp = self.client.post( + "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json') + self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual("failed", resp.data["status"]) + self.assertEqual( + "VNF package(vcpe_vfw_zte_1_0) is not distributed.", + resp.data["statusDescription"]) + + @mock.patch.object(restcall, 'call_req') + @mock.patch.object(sdc, 'download_artifacts') + @mock.patch.object(toscaparsers, 'parse_nsd') + def test_ns_pkg_distribute_when_successfully( + self, mock_parse_nsd, mock_download_artifacts, mock_call_req): + mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data) + mock_download_artifacts.return_value = "/home/vcpe.csar" + mock_call_req.return_value = [0, json.JSONEncoder().encode([{ + "uuid": "1", + "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar", + "distributionStatus": "DISTRIBUTED" + }]), '200'] + VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save() + PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save() + resp = self.client.post( + "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json') + self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual("success", resp.data["status"]) + self.assertEqual( + "CSAR(1) distributed successfully.", + resp.data["statusDescription"]) + + @mock.patch.object(sdc, 'get_artifacts') + def test_ns_when_not_distributed_by_sdc(self, mock_get_artifacts): + mock_get_artifacts.return_value = [{ + "uuid": "1", + "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb", + "name": "underlayvpn", + "version": "2.0", + "toscaModelURL": "/sdc/v1/genericparser/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel", + "category": "Volte", + "subCategory": "VolteVNF", + "resourceType": "VF", + "lifecycleState": "CERTIFIED", + "distributionStatus": "DISTRIBUTION_APPROVED", + "lastUpdaterUserId": "jh0003" + }] + resp = self.client.post( + "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json') + self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual("failed", resp.data["status"]) + self.assertEqual( + "The artifact (services,1) is not distributed from sdc.", + resp.data["statusDescription"]) + + ########################################################################## + + def test_ns_pkg_normal_delete(self): + NSPackageModel(nsPackageId="8", nsdId="2").save() + resp = self.client.delete("/api/genericparser/v1/nspackages/8") + self.assertEqual(resp.status_code, status.HTTP_200_OK) + self.assertEqual("success", resp.data["status"]) + self.assertEqual( + "Delete CSAR(8) successfully.", + resp.data["statusDescription"]) + + def test_ns_pkg_get_all(self): + NSPackageModel( + nsPackageId="13", + nsdId="2", + nsdDesginer="2", + nsdVersion="2", + nsPackageUri="13.csar", + nsdModel="").save() + NSPackageModel( + nsPackageId="14", + nsdId="3", + nsdDesginer="3", + nsdVersion="3", + nsPackageUri="14.csar", + nsdModel="").save() + resp = self.client.get("/api/genericparser/v1/nspackages") + self.assertEqual(resp.status_code, status.HTTP_200_OK) + expect_data = [{"csarId": "13", + "packageInfo": {"csarName": "13.csar", + "nsdProvider": "2", + "nsdId": "2", + "nsPackageId": "13", + "downloadUrl": "http://127.0.0.1:8806/static/genericparser/13/13.csar", + "nsdModel": "", + "nsdVersion": "2", + "nsdInvariantId": None + }}, + {"csarId": "14", + "packageInfo": {"csarName": "14.csar", + "nsdProvider": "3", + "nsdId": "3", + "nsPackageId": "14", + "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar", + "nsdModel": "", + "nsdVersion": "3", + "nsdInvariantId": None}}] + self.assertEqual(expect_data, resp.data) + + def test_ns_pkg_get_one(self): + NSPackageModel( + nsPackageId="14", + nsdId="2", + nsdDesginer="3", + nsdVersion="4", + nsPackageUri="14.csar", + nsdModel="").save() + resp = self.client.get("/api/genericparser/v1/nspackages/14") + self.assertEqual(resp.status_code, status.HTTP_200_OK) + expect_data = { + "csarId": "14", + "packageInfo": { + "nsdId": "2", + "nsPackageId": "14", + "nsdProvider": "3", + "nsdVersion": "4", + "csarName": "14.csar", + "nsdModel": "", + "downloadUrl": "http://127.0.0.1:8806/static/genericparser/14/14.csar", + "nsdInvariantId": None}} + self.assertEqual(expect_data, resp.data) + + def test_ns_pkg_get_one_not_found(self): + resp = self.client.get("/api/genericparser/v1/nspackages/22") + self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) + self.assertEqual( + {"error": "Ns package[22] not Found."}, + resp.data) + + ########################################################################## + + @mock.patch.object(toscaparsers, 'parse_nsd') + def test_nsd_parse_normal(self, mock_parse_nsd): + NSPackageModel(nsPackageId="18", nsdId="12").save() + mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"}) + req_data = {"csarId": "18", "inputs": []} + resp = self.client.post( + "/api/genericparser/v1/parsernsd", + req_data, + format='json') + self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual({"model": '{"a": "b"}'}, resp.data) + + def test_nsd_parse_when_csar_not_exist(self): + req_data = {"csarId": "1", "inputs": []} + resp = self.client.post( + "/api/genericparser/v1/parsernsd", + req_data, + format='json') + self.assertEqual( + resp.status_code, + status.HTTP_500_INTERNAL_SERVER_ERROR) + self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."}) |