aboutsummaryrefslogtreecommitdiffstats
path: root/catalog/packages/tests
diff options
context:
space:
mode:
authordyh <dengyuanhong@chinamobile.com>2019-09-04 09:52:48 +0800
committerdyh <dengyuanhong@chinamobile.com>2019-09-04 16:09:26 +0800
commita32c2b20207885d895bd96204cc166fca14db97b (patch)
tree1edd33368158dc5f057a0a9475dced3df6c3b24c /catalog/packages/tests
parent431a5a35a8e0a26d21c663167303696db8a7a2a6 (diff)
update for change to etsicatalog
Change-Id: Idc2a6950960a324964500a8c4701be422de2b782 Issue-ID: MODELING-216 Signed-off-by: dyh <dengyuanhong@chinamobile.com>
Diffstat (limited to 'catalog/packages/tests')
-rw-r--r--catalog/packages/tests/__init__.py13
-rw-r--r--catalog/packages/tests/const.py596
-rw-r--r--catalog/packages/tests/test_health_check.py50
-rw-r--r--catalog/packages/tests/test_ns_descriptor.py300
-rw-r--r--catalog/packages/tests/test_nsdm_subscription.py521
-rw-r--r--catalog/packages/tests/test_nspackage.py246
-rw-r--r--catalog/packages/tests/test_pnf_descriptor.py286
-rw-r--r--catalog/packages/tests/test_service_descriptor.py95
-rw-r--r--catalog/packages/tests/test_servicepackage.py481
-rw-r--r--catalog/packages/tests/test_vnf_package.py382
-rw-r--r--catalog/packages/tests/test_vnf_pkg_subscription.py183
-rw-r--r--catalog/packages/tests/test_vnfpackage.py258
12 files changed, 3411 insertions, 0 deletions
diff --git a/catalog/packages/tests/__init__.py b/catalog/packages/tests/__init__.py
new file mode 100644
index 0000000..c7b6818
--- /dev/null
+++ b/catalog/packages/tests/__init__.py
@@ -0,0 +1,13 @@
+# 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.
diff --git a/catalog/packages/tests/const.py b/catalog/packages/tests/const.py
new file mode 100644
index 0000000..991c87c
--- /dev/null
+++ b/catalog/packages/tests/const.py
@@ -0,0 +1,596 @@
+# Copyright 2018 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.
+
+vnfd_data = {
+ "volume_storages": [
+ {
+ "properties": {
+ "size_of_storage": {
+ "factor": 10,
+ "value": 10000000000,
+ "unit": "GB",
+ "unit_size": 1000000000
+ },
+ "type_of_storage": "volume",
+ "rdma_enabled": False,
+ "size": "10 GB"
+ },
+ "volume_storage_id": "vNAT_Storage_6wdgwzedlb6sq18uzrr41sof7",
+ "description": ""
+ }
+ ],
+ "inputs": {},
+ "vdus": [
+ {
+ "volume_storages": [
+ "vNAT_Storage_6wdgwzedlb6sq18uzrr41sof7"
+ ],
+ "description": "",
+ "dependencies": [],
+ "vls": [],
+ "properties": {
+ "name": "vNat",
+ "configurable_properties": {
+ "test": {
+ "additional_vnfc_configurable_properties": {
+ "aaa": "1",
+ "bbb": "2",
+ "ccc": "3"
+ }
+ }
+ },
+ "description": "the virtual machine of vNat",
+ "nfvi_constraints": [
+ "test"
+ ],
+ "boot_order": [
+ "vNAT_Storage"
+ ]
+ },
+ "vdu_id": "vdu_vNat",
+ "artifacts": [
+ {
+ "artifact_name": "vNatVNFImage",
+ "type": "tosca.artifacts.nfv.SwImage",
+ "properties": {
+ "operating_system": "linux",
+ "sw_image": "/swimages/vRouterVNF_ControlPlane.qcow2",
+ "name": "vNatVNFImage",
+ "container_format": "bare",
+ "min_ram": "1 GB",
+ "disk_format": "qcow2",
+ "supported_virtualisation_environments": [
+ "test_0"
+ ],
+ "version": "1.0",
+ "checksum": "5000",
+ "min_disk": "10 GB",
+ "size": "10 GB"
+ },
+ "file": "/swimages/vRouterVNF_ControlPlane.qcow2"
+ }
+ ],
+ "nfv_compute": {
+ "flavor_extra_specs": {
+ "hw:cpu_sockets": "2",
+ "sw:ovs_dpdk": "true",
+ "hw:cpu_threads": "2",
+ "hw:numa_mem.1": "3072",
+ "hw:numa_mem.0": "1024",
+ "hw:numa_nodes": "2",
+ "hw:numa_cpus.0": "0,1",
+ "hw:numa_cpus.1": "2,3,4,5",
+ "hw:cpu_cores": "2",
+ "hw:cpu_threads_policy": "isolate"
+ },
+ "cpu_frequency": "2.4 GHz",
+ "num_cpus": 2,
+ "mem_size": "10 GB"
+ },
+ "local_storages": [],
+ "image_file": "vNatVNFImage",
+ "cps": []
+ }
+ ],
+ "image_files": [
+ {
+ "properties": {
+ "operating_system": "linux",
+ "sw_image": "/swimages/vRouterVNF_ControlPlane.qcow2",
+ "name": "vNatVNFImage",
+ "container_format": "bare",
+ "min_ram": "1 GB",
+ "disk_format": "qcow2",
+ "supported_virtualisation_environments": [
+ "test_0"
+ ],
+ "version": "1.0",
+ "checksum": "5000",
+ "min_disk": "10 GB",
+ "size": "10 GB"
+ },
+ "image_file_id": "vNatVNFImage",
+ "description": ""
+ }
+ ],
+ "routers": [],
+ "local_storages": [],
+ "vnf_exposed": {
+ "external_cps": [
+ {
+ "key_name": "sriov_plane",
+ "cp_id": "SRIOV_Port"
+ }
+ ],
+ "forward_cps": []
+ },
+ "vls": [
+ {
+ "route_id": "",
+ "vl_id": "sriov_link",
+ "route_external": False,
+ "description": "",
+ "properties": {
+ "vl_flavours": {
+ "vl_id": "aaaa"
+ },
+ "connectivity_type": {
+ "layer_protocol": "ipv4",
+ "flow_pattern": "flat"
+ },
+ "description": "sriov_link",
+ "test_access": [
+ "test"
+ ]
+ }
+ }
+ ],
+ "cps": [
+ {
+ "vl_id": "sriov_link",
+ "vdu_id": "vdu_vNat",
+ "description": "",
+ "cp_id": "SRIOV_Port",
+ "properties": {
+ "address_data": [
+ {
+ "address_type": "ip_address",
+ "l3_address_data": {
+ "ip_address_type": "ipv4",
+ "floating_ip_activated": False,
+ "number_of_ip_address": 1,
+ "ip_address_assignment": True
+ }
+ }
+ ],
+ "description": "sriov port",
+ "layer_protocol": "ipv4",
+ "virtual_network_interface_requirements": [
+ {
+ "requirement": {
+ "SRIOV": "true"
+ },
+ "support_mandatory": False,
+ "name": "sriov",
+ "description": "sriov"
+ },
+ {
+ "requirement": {
+ "SRIOV": "False"
+ },
+ "support_mandatory": False,
+ "name": "normal",
+ "description": "normal"
+ }
+ ],
+ "role": "root",
+ "bitrate_requirement": 10
+ }
+ }
+ ],
+ "metadata": {
+ "vnfSoftwareVersion": "1.0.0",
+ "vnfProductName": "zte",
+ "localizationLanguage": [
+ "english",
+ "chinese"
+ ],
+ "vnfProvider": "zte",
+ "vnfmInfo": "zte",
+ "defaultLocalizationLanguage": "english",
+ "vnfdId": "zte-hss-1.0",
+ "id": "zte-hss-1.0",
+ "vnfProductInfoDescription": "hss",
+ "vnfdVersion": "1.0.0",
+ "vnfProductInfoName": "hss"
+ },
+ "vnf": {
+ "properties": {
+ "descriptor_id": "zte-hss-1.0",
+ "descriptor_version": "1.0.0",
+ "software_version": "1.0.0",
+ "provider": "zte"
+ },
+ "metadata": {
+ }
+ }
+}
+
+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"
+ }
+}
+}
+
+pnfd_data = {
+ "metadata": {
+ "id": "zte-1.0",
+ }
+}
+
+sd_data = {
+ "inputs": {
+ "sdwanvpnresource_list": [
+ {
+ "sdwanvpn_topology": "",
+ "required": True,
+ "type": "string"
+ }
+ ]
+ },
+ "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
+ }
+ }
+ ],
+ "description": "",
+ "vnfs": [
+ {
+ "vnf_id": "sdwansiteresource",
+ "description": "",
+ "properties": {
+ "sdwandevice_type": "",
+ "sdwandevice_class": "PNF",
+ "multi_stage_design": "false",
+ "min_instances": "1",
+ "sdwansite_controlPoint": "",
+ "id": "cd557883-ac4b-462d-aa01-421b5fa606b1",
+ "sdwansite_longitude": "",
+ "sdwansite_latitude": "",
+ "sdwansite_postcode": "",
+ "sdwansite_type": "",
+ "nf_naming": {
+ "ecomp_generated_naming": True
+ },
+ "sdwansite_emails": "",
+ "sdwansite_role": "",
+ "vnfm_info": "",
+ "sdwansite_address": "",
+ "sdwansite_description": "",
+ "availability_zone_max_count": "1",
+ "sdwansite_name": ""
+ }
+ }
+ ],
+ "service": {
+ "type": "org.openecomp.service.EnhanceService",
+ "properties": {
+ "descriptor_id": "49ee73f4-1e31-4054-b871-eb9b1c29999b",
+ "designer": "",
+ "invariant_id": "5de07996-7ff0-4ec1-b93c-e3a00bb3f207",
+ "name": "Enhance_Service",
+ "verison": ""
+ },
+ "metadata": {
+ "category": "E2E Service",
+ "serviceType": "",
+ "description": "Enhance_Service",
+ "instantiationType": "A-la-carte",
+ "type": "Service",
+ "environmentContext": "General_Revenue-Bearing",
+ "serviceEcompNaming": True,
+ "UUID": "49ee73f4-1e31-4054-b871-eb9b1c29999b",
+ "ecompGeneratedNaming": True,
+ "serviceRole": "",
+ "invariantUUID": "5de07996-7ff0-4ec1-b93c-e3a00bb3f207",
+ "namingPolicy": "",
+ "name": "Enhance_Service"
+ }
+ },
+ "metadata": {
+ "category": "E2E Service",
+ "serviceType": "",
+ "description": "Enhance_Service",
+ "instantiationType": "A-la-carte",
+ "type": "Service",
+ "environmentContext": "General_Revenue-Bearing",
+ "serviceEcompNaming": True,
+ "UUID": "49ee73f4-1e31-4054-b871-eb9b1c29999b",
+ "ecompGeneratedNaming": True,
+ "serviceRole": "",
+ "invariantUUID": "5de07996-7ff0-4ec1-b93c-e3a00bb3f207",
+ "namingPolicy": "",
+ "name": "Enhance_Service"
+ }
+}
+
+vnf_subscription_data = {
+ "filters": {
+ "notificationTypes": [
+ "VnfPackageOnboardingNotification"
+ ],
+ "vnfProductsFromProviders": {
+ "vnfProvider": "string",
+ "vnfProducts": {
+ "vnfProductName": "string",
+ "versions": {
+ "vnfSoftwareVersion": "string",
+ "vnfdVersions": [
+ "string"
+ ]
+ }
+ }
+ },
+ "vnfdId": [
+ "3fa85f64-5717-4562-b3fc-2c963f66afa6"
+ ],
+ "vnfPkgId": [
+ "3fa85f64-5717-4562-b3fc-2c963f66afa6"
+ ],
+ "operationalState": [
+ "ENABLED"
+ ],
+ "usageState": [
+ "IN_USE"
+ ]
+ },
+ "callbackUri": "http://www.vnf1.com/notification",
+ "authentication": {
+ "authType": [
+ "BASIC"
+ ],
+ "paramsBasic": {
+ "userName": "string",
+ "password": "string"
+ }
+ }
+}
diff --git a/catalog/packages/tests/test_health_check.py b/catalog/packages/tests/test_health_check.py
new file mode 100644
index 0000000..f8c3b56
--- /dev/null
+++ b/catalog/packages/tests/test_health_check.py
@@ -0,0 +1,50 @@
+# Copyright (c) 2019, CMCC Technologies Co., Ltd.
+
+# 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
+
+from django.test import TestCase, Client
+from rest_framework import status
+
+
+class TestHealthCheck(TestCase):
+ def setUp(self):
+ self.client = Client()
+
+ def tearDown(self):
+ pass
+
+ def test_vnfpkgm_health_check(self):
+ response = self.client.get("/api/vnfpkgm/v1/health_check")
+ self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
+ resp_data = json.loads(response.content)
+ self.assertEqual({"status": "active"}, resp_data)
+
+ def test_nsd_health_check(self):
+ response = self.client.get("/api/nsd/v1/health_check")
+ self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
+ resp_data = json.loads(response.content)
+ self.assertEqual({"status": "active"}, resp_data)
+
+ def test_catalog_health_check(self):
+ response = self.client.get("/api/catalog/v1/health_check")
+ self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
+ resp_data = json.loads(response.content)
+ self.assertEqual({"status": "active"}, resp_data)
+
+ def test_parser_health_check(self):
+ response = self.client.get("/api/parser/v1/health_check")
+ self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
+ resp_data = json.loads(response.content)
+ self.assertEqual({"status": "active"}, resp_data)
diff --git a/catalog/packages/tests/test_ns_descriptor.py b/catalog/packages/tests/test_ns_descriptor.py
new file mode 100644
index 0000000..473786e
--- /dev/null
+++ b/catalog/packages/tests/test_ns_descriptor.py
@@ -0,0 +1,300 @@
+# Copyright 2018 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 copy
+import json
+import mock
+import os
+
+from django.test import TestCase
+from rest_framework import status
+from rest_framework.test import APIClient
+from catalog.packages.biz.ns_descriptor import NsDescriptor
+from catalog.packages.const import PKG_STATUS
+from catalog.packages.tests.const import nsd_data
+from catalog.pub.config.config import CATALOG_ROOT_PATH
+from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
+from catalog.pub.utils import toscaparser
+
+
+class TestNsDescriptor(TestCase):
+ def setUp(self):
+ self.client = APIClient()
+ self.user_defined_data = {
+ 'key1': 'value1',
+ 'key2': 'value2',
+ 'key3': 'value3',
+ }
+ self.expected_nsd_info = {
+ 'id': None,
+ 'nsdId': None,
+ 'nsdName': None,
+ 'nsdVersion': None,
+ 'nsdDesigner': None,
+ 'nsdInvariantId': None,
+ 'vnfPkgIds': [],
+ 'pnfdInfoIds': [],
+ 'nestedNsdInfoIds': [],
+ 'nsdOnboardingState': 'CREATED',
+ 'onboardingFailureDetails': None,
+ 'nsdOperationalState': 'DISABLED',
+ 'nsdUsageState': 'NOT_IN_USE',
+ 'userDefinedData': self.user_defined_data,
+ '_links': None
+ }
+ self.nsdModel = {
+ "pnfs": [{"properties": {"id": "m6000_s"}}],
+ "vnfs": [{"properties": {"id": "123"}}]
+ }
+
+ def tearDown(self):
+ pass
+
+ def test_nsd_create_normal(self):
+ reqest_data = {'userDefinedData': self.user_defined_data}
+ expected_reponse_data = {
+ 'nsdOnboardingState': 'CREATED',
+ 'nsdOperationalState': 'DISABLED',
+ 'nsdUsageState': 'NOT_IN_USE',
+ 'userDefinedData': self.user_defined_data,
+ '_links': None
+ }
+
+ response = self.client.post(
+ '/api/nsd/v1/ns_descriptors',
+ data=reqest_data,
+ format='json'
+ )
+ response.data.pop('id')
+ self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+ self.assertEqual(expected_reponse_data, response.data)
+
+ def test_query_multiple_nsds_normal(self):
+ expected_reponse_data = [
+ copy.deepcopy(self.expected_nsd_info),
+ copy.deepcopy(self.expected_nsd_info)
+ ]
+ expected_reponse_data[0]['id'] = '0'
+ expected_reponse_data[0]['nsdId'] = '0'
+ expected_reponse_data[1]['id'] = '1'
+ expected_reponse_data[1]['nsdId'] = '1'
+
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ for i in range(2):
+ NSPackageModel(
+ nsPackageId=str(i),
+ onboardingState='CREATED',
+ operationalState='DISABLED',
+ usageState='NOT_IN_USE',
+ userDefinedData=user_defined_data,
+ nsdId=str(i)
+ ).save()
+
+ response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(expected_reponse_data, response.data)
+
+ expected_reponse_data = [
+ copy.deepcopy(self.expected_nsd_info)
+ ]
+ expected_reponse_data[0]['id'] = '1'
+ expected_reponse_data[0]['nsdId'] = '1'
+ response = self.client.get('/api/nsd/v1/ns_descriptors?nsdId=1', format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(expected_reponse_data, response.data)
+
+ def test_query_single_nsd_normal(self):
+ expected_reponse_data = copy.deepcopy(self.expected_nsd_info)
+ expected_reponse_data['id'] = '22'
+
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ NSPackageModel(
+ nsPackageId='22',
+ onboardingState='CREATED',
+ operationalState='DISABLED',
+ usageState='NOT_IN_USE',
+ userDefinedData=user_defined_data,
+ nsdModel=json.JSONEncoder().encode(self.nsdModel)
+ ).save()
+
+ response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(expected_reponse_data, response.data)
+
+ def test_query_single_when_ns_not_exist(self):
+ response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ def test_delete_single_nsd_normal(self):
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ NSPackageModel(
+ nsPackageId='21',
+ operationalState='DISABLED',
+ usageState='NOT_IN_USE',
+ userDefinedData=user_defined_data,
+ nsdModel='test'
+ ).save()
+
+ response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
+ self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(None, response.data)
+
+ def test_delete_when_ns_not_exist(self):
+ response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
+ self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+
+ @mock.patch.object(toscaparser, 'parse_nsd')
+ def test_nsd_content_upload_normal(self, mock_parse_nsd):
+ user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
+ mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
+ VnfPackageModel(
+ vnfPackageId="111",
+ vnfdId="vcpe_vfw_zte_1_0"
+ ).save()
+
+ PnfPackageModel(
+ pnfPackageId="112",
+ pnfdId="m6000_s"
+ ).save()
+
+ NSPackageModel(
+ nsPackageId='22',
+ operationalState='DISABLED',
+ usageState='NOT_IN_USE',
+ userDefinedData=user_defined_data_json,
+ ).save()
+
+ with open('nsd_content.txt', 'wt') as fp:
+ fp.write('test')
+ with open('nsd_content.txt', 'rt') as fp:
+ resp = self.client.put(
+ "/api/nsd/v1/ns_descriptors/22/nsd_content",
+ {'file': fp},
+ )
+ file_content = ''
+ with open(os.path.join(CATALOG_ROOT_PATH, '22/nsd_content.txt')) as fp:
+ data = fp.read()
+ file_content = '%s%s' % (file_content, data)
+ ns_pkg = NSPackageModel.objects.filter(nsPackageId="22")
+ self.assertEqual("VCPE_NS", ns_pkg[0].nsdId)
+ self.assertEqual(PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
+ self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(None, resp.data)
+ self.assertEqual(file_content, 'test')
+ os.remove('nsd_content.txt')
+
+ def test_nsd_content_upload_failure(self):
+ with open('nsd_content.txt', 'wt') as fp:
+ fp.write('test')
+ with open('nsd_content.txt', 'rt') as fp:
+ response = self.client.put(
+ "/api/nsd/v1/ns_descriptors/22/nsd_content",
+ {'file': fp},
+ )
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ def test_nsd_content_download_normal(self):
+ with open('nsd_content.txt', 'wt') as fp:
+ fp.writelines('test1')
+ fp.writelines('test2')
+ NSPackageModel.objects.create(
+ nsPackageId='23',
+ onboardingState='ONBOARDED',
+ localFilePath='nsd_content.txt'
+ )
+ response = self.client.get(
+ "/api/nsd/v1/ns_descriptors/23/nsd_content", format='json'
+ )
+ file_content = ""
+ for data in response.streaming_content:
+ file_content = '%s%s' % (file_content, data.decode())
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual("test1test2", file_content)
+ os.remove('nsd_content.txt')
+
+ def test_nsd_content_download_when_ns_not_exist(self):
+ response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ def test_nsd_content_download_failed(self):
+ NSPackageModel.objects.create(
+ nsPackageId='23',
+ onboardingState='CREATED',
+ localFilePath='nsd_content.txt'
+ )
+ response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ def test_nsd_content_partial_download_normal(self):
+ with open('nsd_content.txt', 'wt') as fp:
+ fp.writelines('test1')
+ fp.writelines('test2')
+ NSPackageModel(
+ nsPackageId='23',
+ onboardingState='ONBOARDED',
+ localFilePath='nsd_content.txt'
+ ).save()
+
+ response = self.client.get(
+ "/api/nsd/v1/ns_descriptors/23/nsd_content",
+ HTTP_RANGE='5-10',
+ format='json'
+ )
+ partial_file_content = ''
+ for data in response.streaming_content:
+ partial_file_content = '%s%s' % (partial_file_content, data.decode())
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual("test2", partial_file_content)
+ os.remove('nsd_content.txt')
+
+ @mock.patch.object(NsDescriptor, 'create')
+ def test_nsd_create_when_catch_exception(self, mock_create):
+ reqest_data = {'userDefinedData': self.user_defined_data}
+ mock_create.side_effect = TypeError("integer type")
+ response = self.client.post('/api/nsd/v1/ns_descriptors', data=reqest_data, format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(NsDescriptor, 'query_single')
+ def test_query_single_when_catch_exception(self, mock_query_single):
+ mock_query_single.side_effect = TypeError("integer type")
+ response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(NsDescriptor, 'query_multiple')
+ def test_query_multiple_when_catch_exception(self, mock_query_multipe):
+ mock_query_multipe.side_effect = TypeError("integer type")
+ response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(NsDescriptor, 'delete_single')
+ def test_delete_when_catch_exception(self, mock_delete_single):
+ mock_delete_single.side_effect = TypeError("integer type")
+ response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(NsDescriptor, 'upload')
+ def test_upload_when_catch_exception(self, mock_upload):
+ mock_upload.side_effect = TypeError("integer type")
+ with open('nsd_content.txt', 'wt') as fp:
+ fp.write('test')
+ with open('nsd_content.txt', 'rt') as fp:
+ response = self.client.put("/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp})
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+ os.remove('nsd_content.txt')
+
+ @mock.patch.object(NsDescriptor, 'download')
+ def test_download_when_catch_exception(self, mock_download):
+ mock_download.side_effect = TypeError("integer type")
+ response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
diff --git a/catalog/packages/tests/test_nsdm_subscription.py b/catalog/packages/tests/test_nsdm_subscription.py
new file mode 100644
index 0000000..f73c416
--- /dev/null
+++ b/catalog/packages/tests/test_nsdm_subscription.py
@@ -0,0 +1,521 @@
+# Copyright (C) 2019 Verizon. All Rights Reserved
+#
+# 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
+import uuid
+from django.test import TestCase
+from rest_framework.test import APIClient
+from rest_framework import status
+
+from catalog.packages.biz.nsdm_subscription import NsdmSubscription
+from catalog.pub.database.models import NsdmSubscriptionModel
+
+
+class TestNsdmSubscription(TestCase):
+
+ def setUp(self):
+ self.client = APIClient()
+ NsdmSubscriptionModel.objects.all().delete()
+ self.subscription_id = str(uuid.uuid4())
+ self.subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsBasic": {
+ "userName": "username",
+ "password": "password"
+ }
+ },
+ "filter": {
+ "nsdId": ["b632bddc-abcd-4180-bd8d-4e8a9578eff7"],
+ }
+ }
+ self.links = {
+ "self": {
+ "href": "/api/v1/subscriptions/" + self.subscription_id
+ }
+ }
+ self.test_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "id": self.subscription_id,
+ "filter": {
+ "notificationTypes": [
+ "NsdOnBoardingNotification"
+ ],
+ "nsdInfoId": [],
+ "nsdId": [],
+ "nsdName": [],
+ "nsdVersion": [],
+ "nsdInvariantId": [],
+ "vnfPkgIds": [],
+ "nestedNsdInfoIds": [],
+ "nsdOnboardingState": [],
+ "nsdOperationalState": [],
+ "nsdUsageState": [],
+ "pnfdInfoIds": [],
+ "pnfdId": [],
+ "pnfdName": [],
+ "pnfdVersion": [],
+ "pnfdProvider": [],
+ "pnfdInvariantId": [],
+ "pnfdOnboardingState": [],
+ "pnfdUsageState": []
+ },
+ "_links": self.links,
+ }
+
+ def tearDown(self):
+ pass
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
+ temp_uuid = str(uuid.uuid4())
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=self.subscription, format='json')
+ self.assertEqual(201, response.status_code)
+ self.assertEqual(self.subscription["callbackUri"],
+ response.data["callbackUri"])
+ self.assertEqual(temp_uuid, response.data["id"])
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
+ temp_uuid = str(uuid.uuid4())
+ mock_requests.return_value.status_code = 500
+ mock_requests.get.return_value.status_code = 500
+ mock_uuid4.return_value = temp_uuid
+ expected_data = {
+ 'status': 500,
+ 'detail': "callbackUri http://callbackuri.com didn't"
+ " return 204 statuscode."
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=self.subscription, format='json')
+ self.assertEqual(500, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch("requests.get")
+ def test_nsdm_second_subscription(self, mock_requests):
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=self.subscription, format='json')
+ self.assertEqual(201, response.status_code)
+ self.assertEqual(self.subscription["callbackUri"],
+ response.data["callbackUri"])
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsBasic": {
+ "userName": "username",
+ "password": "password"
+ }
+ },
+ "filter": {
+ "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
+ }
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(201, response.status_code)
+ self.assertEqual(dummy_subscription["callbackUri"],
+ response.data["callbackUri"])
+
+ @mock.patch("requests.get")
+ def test_nsdm_duplicate_subscription(self, mock_requests):
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=self.subscription, format='json')
+ self.assertEqual(201, response.status_code)
+ self.assertEqual(self.subscription["callbackUri"],
+ response.data["callbackUri"])
+ expected_data = {
+ 'status': 303,
+ 'detail': 'Already Subscription exists with'
+ ' the same callbackUri and filter'
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=self.subscription, format='json')
+ self.assertEqual(303, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch("requests.get")
+ def test_nsdm_bad_request(self, mock_requests):
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsBasic": {
+ "userName": "username",
+ "password": "password"
+ }
+ },
+ "filter": {
+ "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
+ }
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(400, response.status_code)
+
+ @mock.patch("requests.get")
+ def test_nsdm_invalid_authtype_subscription(self, mock_requests):
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
+ "paramsBasic": {
+ "userName": "username",
+ "password": "password"
+ }
+ }
+ }
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ expected_data = {
+ 'status': 400,
+ 'detail': 'Auth type should be BASIC'
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(400, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch("requests.get")
+ def test_nsdm_invalid_authtype_oauthclient_subscription(
+ self, mock_requests):
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsOauth2ClientCredentials": {
+ "clientId": "clientId",
+ "clientPassword": "password",
+ "tokenEndpoint": "http://tokenEndpoint"
+ }
+ }
+ }
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ expected_data = {
+ 'status': 400,
+ 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS'
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(400, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch("requests.get")
+ def test_nsdm_invalid_authparams_subscription(self, mock_requests):
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsBasic": {
+ "userName": "username"
+ }
+ }
+ }
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ expected_data = {
+ 'status': 400,
+ 'detail': 'userName and password needed for BASIC'
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(400, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch("requests.get")
+ def test_nsdm_invalid_authparams_oauthclient_subscription(
+ self, mock_requests):
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
+ "paramsOauth2ClientCredentials": {
+ "clientPassword": "password",
+ "tokenEndpoint": "http://tokenEndpoint"
+ }
+ }
+ }
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ expected_data = {
+ 'status': 400,
+ 'detail': 'clientId, clientPassword and tokenEndpoint'
+ ' required for OAUTH2_CLIENT_CREDENTIALS'
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(400, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch("requests.get")
+ def test_nsdm_invalid_filter_subscription(self, mock_requests):
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsBasic": {
+ "userName": "username",
+ "password": "password"
+ }
+ },
+ "filter": {
+ "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
+ "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
+ }
+ }
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ expected_data = {
+ 'status': 400,
+ 'detail': 'Notification Filter should contain'
+ ' either nsdId or nsdInfoId'
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(400, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch("requests.get")
+ def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
+ dummy_subscription = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsBasic": {
+ "userName": "username",
+ "password": "password"
+ }
+ },
+ "filter": {
+ "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
+ "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
+ }
+ }
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ expected_data = {
+ 'status': 400,
+ 'detail': 'Notification Filter should contain'
+ ' either pnfdId or pnfdInfoIds'
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=dummy_subscription, format='json')
+ self.assertEqual(400, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ @mock.patch.object(NsdmSubscription, 'create')
+ def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
+ mock_create.side_effect = TypeError("Unicode type")
+ response = self.client.post('/api/nsd/v1/subscriptions',
+ data=self.subscription, format='json')
+ self.assertEqual(response.status_code,
+ status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ def test_nsdm_get_subscriptions(self):
+ NsdmSubscriptionModel(subscriptionid=self.subscription_id,
+ callback_uri="http://callbackuri.com",
+ auth_info={},
+ notificationTypes=json.dumps(
+ ["NsdOnBoardingNotification"]),
+ nsdId=[], nsdVersion=[],
+ nsdInfoId=[], nsdDesigner=[],
+ nsdName=[], nsdInvariantId=[],
+ vnfPkgIds=[], pnfdInfoIds=[],
+ nestedNsdInfoIds=[], nsdOnboardingState=[],
+ nsdOperationalState=[], nsdUsageState=[],
+ pnfdId=[], pnfdVersion=[], pnfdProvider=[],
+ pnfdName=[], pnfdInvariantId=[],
+ pnfdOnboardingState=[], pnfdUsageState=[],
+ links=json.dumps(self.links)).save()
+ response = self.client.get("/api/nsd/v1/subscriptions",
+ format='json')
+ self.assertEqual(status.HTTP_200_OK, response.status_code)
+ self.assertEqual([self.test_subscription], response.data)
+
+ def test_nsdm_get_subscriptions_filter(self):
+ NsdmSubscriptionModel(subscriptionid=self.subscription_id,
+ callback_uri="http://callbackuri.com",
+ auth_info={},
+ notificationTypes=json.dumps(
+ ["NsdOnBoardingNotification"]),
+ nsdId=[], nsdVersion=[],
+ nsdInfoId=[], nsdDesigner=[],
+ nsdName=[], nsdInvariantId=[],
+ vnfPkgIds=[], pnfdInfoIds=[],
+ nestedNsdInfoIds=[], nsdOnboardingState=[],
+ nsdOperationalState=[], nsdUsageState=[],
+ pnfdId=[], pnfdVersion=[], pnfdProvider=[],
+ pnfdName=[], pnfdInvariantId=[],
+ pnfdOnboardingState=[], pnfdUsageState=[],
+ links=json.dumps(self.links)).save()
+ response = self.client.get("/api/nsd/v1/subscriptions"
+ "?notificationTypes"
+ "=NsdOnBoardingNotification",
+ format='json')
+ self.assertEqual(status.HTTP_200_OK, response.status_code)
+ self.assertEqual([self.test_subscription], response.data)
+
+ def test_nsdm_get_subscriptions_filter_failure(self):
+ NsdmSubscriptionModel(subscriptionid=self.subscription_id,
+ callback_uri="http://callbackuri.com",
+ auth_info={},
+ notificationTypes=json.dumps(
+ ["NsdOnBoardingNotification"]),
+ nsdId=[], nsdVersion=[],
+ nsdInfoId=[], nsdDesigner=[],
+ nsdName=[], nsdInvariantId=[],
+ vnfPkgIds=[], pnfdInfoIds=[],
+ nestedNsdInfoIds=[], nsdOnboardingState=[],
+ nsdOperationalState=[], nsdUsageState=[],
+ pnfdId=[], pnfdVersion=[], pnfdProvider=[],
+ pnfdName=[], pnfdInvariantId=[],
+ pnfdOnboardingState=[], pnfdUsageState=[],
+ links=json.dumps(self.links)).save()
+ response = self.client.get("/api/nsd/v1/subscriptions"
+ "?notificationTypes="
+ "PnfdOnBoardingFailureNotification",
+ format='json')
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
+
+ def test_nsdm_get_subscriptions_invalid_filter(self):
+ NsdmSubscriptionModel(subscriptionid=self.subscription_id,
+ callback_uri="http://callbackuri.com",
+ auth_info={},
+ notificationTypes=json.dumps(
+ ["NsdOnBoardingNotification"]),
+ nsdId=[], nsdVersion=[],
+ nsdInfoId=[], nsdDesigner=[],
+ nsdName=[], nsdInvariantId=[],
+ vnfPkgIds=[], pnfdInfoIds=[],
+ nestedNsdInfoIds=[], nsdOnboardingState=[],
+ nsdOperationalState=[], nsdUsageState=[],
+ pnfdId=[], pnfdVersion=[], pnfdProvider=[],
+ pnfdName=[], pnfdInvariantId=[],
+ pnfdOnboardingState=[], pnfdUsageState=[],
+ links=json.dumps(self.links)).save()
+ response = self.client.get("/api/nsd/v1/subscriptions"
+ "?notificationTypes="
+ "PnfdOnBoardingFailureNotificati",
+ format='json')
+ self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+
+ @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
+ def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
+ mock_create.side_effect = TypeError("Unicode type")
+ response = self.client.get('/api/nsd/v1/subscriptions', format='json')
+ self.assertEqual(response.status_code,
+ status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ def test_nsdm_get_subscription(self):
+ NsdmSubscriptionModel(subscriptionid=self.subscription_id,
+ callback_uri="http://callbackuri.com",
+ auth_info={},
+ notificationTypes=json.dumps(
+ ["NsdOnBoardingNotification"]),
+ nsdId=[], nsdVersion=[],
+ nsdInfoId=[], nsdDesigner=[],
+ nsdName=[], nsdInvariantId=[],
+ vnfPkgIds=[], pnfdInfoIds=[],
+ nestedNsdInfoIds=[], nsdOnboardingState=[],
+ nsdOperationalState=[], nsdUsageState=[],
+ pnfdId=[], pnfdVersion=[], pnfdProvider=[],
+ pnfdName=[], pnfdInvariantId=[],
+ pnfdOnboardingState=[], pnfdUsageState=[],
+ links=json.dumps(self.links)).save()
+ response = self.client.get('/api/nsd/v1/'
+ 'subscriptions/' + self.subscription_id,
+ format='json')
+ self.assertEqual(status.HTTP_200_OK, response.status_code)
+ self.assertEqual(self.test_subscription, response.data)
+
+ def test_nsdm_get_subscription_failure(self):
+ expected_data = {
+ "status": 404,
+ "detail": "Subscription(" + self.subscription_id + ") "
+ "doesn't exists"
+ }
+ response = self.client.get('/api/nsd/v1/'
+ 'subscriptions/' + self.subscription_id,
+ format='json')
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
+ self.assertEqual(expected_data, response.data)
+
+ def test_nsdm_get_subscription_failure_bad_request(self):
+ response = self.client.get("/api/nsd/v1/subscriptions/123",
+ format='json')
+ self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+
+ @mock.patch.object(NsdmSubscription, 'query_single_subscription')
+ def test_nsdmsubscription_getsingle_when_catch_exception(
+ self, mock_create):
+ mock_create.side_effect = TypeError("Unicode type")
+ response = self.client.get('/api/nsd/v1/'
+ 'subscriptions/' + self.subscription_id,
+ format='json')
+ self.assertEqual(response.status_code,
+ status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ def test_ndsm_delete_subscription(self):
+ NsdmSubscriptionModel(subscriptionid=self.subscription_id,
+ callback_uri="http://callbackuri.com",
+ auth_info={},
+ notificationTypes=json.dumps(
+ ["NsdOnBoardingNotification"]),
+ nsdId=[], nsdVersion=[],
+ nsdInfoId=[], nsdDesigner=[],
+ nsdName=[], nsdInvariantId=[],
+ vnfPkgIds=[], pnfdInfoIds=[],
+ nestedNsdInfoIds=[], nsdOnboardingState=[],
+ nsdOperationalState=[], nsdUsageState=[],
+ pnfdId=[], pnfdVersion=[], pnfdProvider=[],
+ pnfdName=[], pnfdInvariantId=[],
+ pnfdOnboardingState=[], pnfdUsageState=[],
+ links=json.dumps(self.links)).save()
+ response = self.client.delete('/api/nsd/v1/'
+ 'subscriptions/' + self.subscription_id,
+ format='json')
+ self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
+
+ def test_ndsm_delete_subscription_failure(self):
+ response = self.client.delete('/api/nsd/v1/'
+ 'subscriptions/' + self.subscription_id,
+ format='json')
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
+
+ def test_nsdm_delete_subscription_failure_bad_request(self):
+ response = self.client.delete("/api/nsd/v1/subscriptions/123",
+ format='json')
+ self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+
+ @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
+ def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
+ mock_create.side_effect = TypeError("Unicode type")
+ response = self.client.delete('/api/nsd/v1/'
+ 'subscriptions/' + self.subscription_id,
+ format='json')
+ self.assertEqual(response.status_code,
+ status.HTTP_500_INTERNAL_SERVER_ERROR)
diff --git a/catalog/packages/tests/test_nspackage.py b/catalog/packages/tests/test_nspackage.py
new file mode 100644
index 0000000..91f3503
--- /dev/null
+++ b/catalog/packages/tests/test_nspackage.py
@@ -0,0 +1,246 @@
+# 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 catalog.pub.utils import restcall, toscaparser
+from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
+from catalog.pub.msapi import sdc
+from .const import nsd_data
+
+
+class TestNsPackage(TestCase):
+ def setUp(self):
+ self.client = Client()
+ NSPackageModel.objects.filter().delete()
+ VnfPackageModel.objects.filter().delete()
+ self.nsd_data = nsd_data
+
+ def tearDown(self):
+ pass
+
+ def test_ns_pkg_distribute_when_ns_exists(self):
+ NSPackageModel(nsPackageId="1", nsdId="2").save()
+ resp = self.client.post(
+ "/api/catalog/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/catalog/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(toscaparser, '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/catalog/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(toscaparser, '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/catalog/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(toscaparser, '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/catalog/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/catalog/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/catalog/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/catalog/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/catalog/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/catalog/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/catalog/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/catalog/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/catalog/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/catalog/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(toscaparser, '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/catalog/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/catalog/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."})
diff --git a/catalog/packages/tests/test_pnf_descriptor.py b/catalog/packages/tests/test_pnf_descriptor.py
new file mode 100644
index 0000000..8af8614
--- /dev/null
+++ b/catalog/packages/tests/test_pnf_descriptor.py
@@ -0,0 +1,286 @@
+# Copyright 2018 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 copy
+import json
+import mock
+import os
+import shutil
+
+
+from django.test import TestCase
+from rest_framework import status
+from rest_framework.test import APIClient
+from catalog.packages.biz.pnf_descriptor import PnfDescriptor
+from catalog.packages.const import PKG_STATUS
+from catalog.packages.tests.const import pnfd_data
+from catalog.pub.config.config import CATALOG_ROOT_PATH
+from catalog.pub.database.models import PnfPackageModel, NSPackageModel
+from catalog.pub.utils import toscaparser
+
+
+class TestPnfDescriptor(TestCase):
+ def setUp(self):
+ self.client = APIClient()
+ self.user_defined_data = {
+ 'key1': 'value1',
+ 'key2': 'value2',
+ 'key3': 'value3',
+ }
+ self.expected_pnfd_info = {
+ 'id': None,
+ 'pnfdId': None,
+ 'pnfdName': None,
+ 'pnfdVersion': None,
+ 'pnfdProvider': None,
+ 'pnfdInvariantId': None,
+ 'pnfdOnboardingState': 'CREATED',
+ 'onboardingFailureDetails': None,
+ 'pnfdUsageState': 'NOT_IN_USE',
+ 'userDefinedData': self.user_defined_data,
+ '_links': None
+ }
+ self.nsdModel = {
+ "pnfs": [{"properties": {"id": "m6000_s"}}]
+ }
+
+ def tearDown(self):
+ file_path = os.path.join(CATALOG_ROOT_PATH, "22")
+ if os.path.exists(file_path):
+ shutil.rmtree(file_path)
+
+ def test_pnfd_create_normal(self):
+ request_data = {'userDefinedData': self.user_defined_data}
+ expected_reponse_data = {
+ 'pnfdOnboardingState': 'CREATED',
+ 'pnfdUsageState': 'NOT_IN_USE',
+ 'userDefinedData': self.user_defined_data,
+ '_links': None
+ }
+
+ response = self.client.post(
+ '/api/nsd/v1/pnf_descriptors',
+ data=request_data,
+ format='json'
+ )
+ response.data.pop('id')
+ self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+ self.assertEqual(expected_reponse_data, response.data)
+
+ def test_query_multiple_pnfds_normal(self):
+ expected_reponse_data = [
+ copy.deepcopy(self.expected_pnfd_info),
+ copy.deepcopy(self.expected_pnfd_info)
+ ]
+ expected_reponse_data[0]['id'] = '0'
+ expected_reponse_data[1]['id'] = '1'
+
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ for i in range(2):
+ PnfPackageModel(
+ pnfPackageId=str(i),
+ onboardingState='CREATED',
+ usageState='NOT_IN_USE',
+ userDefinedData=user_defined_data
+ ).save()
+ response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(expected_reponse_data, response.data)
+
+ def test_query_single_pnfd_normal(self):
+ expected_reponse_data = copy.deepcopy(self.expected_pnfd_info)
+ expected_reponse_data['id'] = '22'
+
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ PnfPackageModel(
+ pnfPackageId='22',
+ onboardingState='CREATED',
+ usageState='NOT_IN_USE',
+ userDefinedData=user_defined_data
+ ).save()
+
+ response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(expected_reponse_data, response.data)
+
+ def test_query_single_pnfd_failed(self):
+ response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ def test_delete_single_pnfd_normal(self):
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ PnfPackageModel(
+ pnfPackageId='22',
+ usageState=PKG_STATUS.NOT_IN_USE,
+ userDefinedData=user_defined_data,
+ pnfdModel='test'
+ ).save()
+ NSPackageModel.objects.create(
+ nsPackageId="111",
+ nsdModel=json.JSONEncoder().encode(self.nsdModel)
+ )
+ resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
+ self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(None, resp.data)
+
+ def test_delete_single_pnfd_when_not_exist(self):
+ resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
+ self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(None, resp.data)
+
+ @mock.patch.object(toscaparser, "parse_pnfd")
+ def test_pnfd_content_upload_normal(self, mock_parse_pnfd):
+ user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
+ PnfPackageModel(
+ pnfPackageId='22',
+ usageState=PKG_STATUS.NOT_IN_USE,
+ userDefinedData=user_defined_data_json,
+ ).save()
+ mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
+ with open('pnfd_content.txt', 'wt') as fp:
+ fp.write('test')
+
+ with open('pnfd_content.txt', 'rt') as fp:
+ resp = self.client.put(
+ "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
+ {'file': fp},
+ )
+ pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId="22")
+ self.assertEqual(pnf_pkg[0].pnfdId, "zte-1.0")
+ self.assertEqual(pnf_pkg[0].onboardingState, PKG_STATUS.ONBOARDED)
+ self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(None, resp.data)
+ os.remove('pnfd_content.txt')
+
+ def test_pnfd_content_upload_when_pnf_not_exist(self):
+ with open('pnfd_content.txt', 'wt') as fp:
+ fp.write('test')
+
+ with open('pnfd_content.txt', 'rt') as fp:
+ resp = self.client.put(
+ "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
+ {'file': fp},
+ )
+ self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, "parse_pnfd")
+ def test_pnfd_content_upload_when_pnfd_exist(self, mock_parse_pnfd):
+ with open('pnfd_content.txt', 'wt') as fp:
+ fp.write('test')
+ PnfPackageModel(
+ pnfPackageId='22',
+ usageState=PKG_STATUS.NOT_IN_USE,
+ pnfdId="zte-1.1"
+ ).save()
+ PnfPackageModel(
+ pnfPackageId='23',
+ usageState=PKG_STATUS.NOT_IN_USE,
+ pnfdId="zte-1.0"
+ ).save()
+ mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
+ with open('pnfd_content.txt', 'rt') as fp:
+ resp = self.client.put(
+ "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
+ {'file': fp},
+ )
+ self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ def test_pnfd_download_normal(self):
+ with open('pnfd_content.txt', 'wt') as fp:
+ fp.writelines('test1')
+ fp.writelines('test2')
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ PnfPackageModel(
+ pnfPackageId='22',
+ usageState=PKG_STATUS.NOT_IN_USE,
+ onboardingState=PKG_STATUS.ONBOARDED,
+ userDefinedData=user_defined_data,
+ localFilePath="pnfd_content.txt",
+ pnfdModel='test'
+ ).save()
+ resp = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
+ file_content = ""
+ for data in resp.streaming_content:
+ file_content = '%s%s' % (file_content, data.decode())
+ self.assertEqual(resp.status_code, status.HTTP_200_OK)
+ self.assertEqual("test1test2", file_content)
+ os.remove('pnfd_content.txt')
+
+ def test_pnfd_download_failed(self):
+ response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ def test_pnfd_download_when_not_on_boarded(self):
+ with open('pnfd_content.txt', 'wt') as fp:
+ fp.writelines('test1')
+ fp.writelines('test2')
+ user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
+ PnfPackageModel(
+ pnfPackageId='22',
+ usageState=PKG_STATUS.NOT_IN_USE,
+ onboardingState=PKG_STATUS.CREATED,
+ userDefinedData=user_defined_data,
+ localFilePath="pnfd_content.txt",
+ pnfdModel='test'
+ ).save()
+ response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+ os.remove('pnfd_content.txt')
+
+ @mock.patch.object(PnfDescriptor, "create")
+ def test_pnfd_create_when_catch_exception(self, mock_create):
+ request_data = {'userDefinedData': self.user_defined_data}
+ mock_create.side_effect = TypeError('integer type')
+ response = self.client.post('/api/nsd/v1/pnf_descriptors', data=request_data, format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(PnfDescriptor, "delete_single")
+ def test_delete_single_when_catch_exception(self, mock_delete_single):
+ mock_delete_single.side_effect = TypeError("integer type")
+ response = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(PnfDescriptor, "query_single")
+ def test_query_single_when_catch_exception(self, mock_query_single):
+ mock_query_single.side_effect = TypeError("integer type")
+ response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(PnfDescriptor, "query_multiple")
+ def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
+ mock_query_muitiple.side_effect = TypeError("integer type")
+ response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(PnfDescriptor, "upload")
+ def test_upload_when_catch_exception(self, mock_upload):
+ mock_upload.side_effect = TypeError("integer type")
+ response = self.client.put("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(PnfDescriptor, "download")
+ def test_download_when_catch_exception(self, mock_download):
+ mock_download.side_effect = TypeError("integer type")
+ response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, 'parse_pnfd')
+ def test_pnfd_parse_normal(self, mock_parse_pnfd):
+ PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
+ mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
+ req_data = {"csarId": "8", "inputs": []}
+ resp = self.client.post("/api/catalog/v1/parserpnfd", req_data, format='json')
+ self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
+ self.assertEqual({"model": '{"c": "d"}'}, resp.data)
diff --git a/catalog/packages/tests/test_service_descriptor.py b/catalog/packages/tests/test_service_descriptor.py
new file mode 100644
index 0000000..08a6f03
--- /dev/null
+++ b/catalog/packages/tests/test_service_descriptor.py
@@ -0,0 +1,95 @@
+# Copyright (c) 2019, CMCC Technologies. Co., Ltd.
+#
+# 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 logging
+
+from django.test import TestCase
+from mock import mock
+
+from catalog.packages.biz.service_descriptor import ServiceDescriptor
+from catalog.packages.const import PKG_STATUS
+from catalog.pub.database.models import ServicePackageModel, VnfPackageModel, PnfPackageModel
+from catalog.pub.exceptions import PackageNotFoundException
+from catalog.pub.utils import toscaparser
+from .const import sd_data
+
+logger = logging.getLogger(__name__)
+
+
+class TestServiceDescription(TestCase):
+
+ def setUp(self):
+ self.user_defined_data = {
+ 'key1': 'value1',
+ 'key2': 'value2',
+ 'key3': 'value3',
+ }
+ self.data = {
+ 'userDefinedData': self.user_defined_data,
+ }
+ self.sd_data = sd_data
+ ServicePackageModel.objects.filter().delete()
+
+ def tearDown(self):
+ pass
+
+ def test_create(self):
+ result_data = ServiceDescriptor().create(self.data)
+ self.assertIsNotNone(result_data['id'])
+ service_package = ServicePackageModel.objects.filter(servicePackageId=result_data['id'])[0]
+ self.assertIsNotNone(service_package)
+ self.assertEqual(PKG_STATUS.DISABLED, service_package.operationalState)
+ self.assertEqual(PKG_STATUS.CREATED, service_package.onboardingState)
+ self.assertEqual(PKG_STATUS.NOT_IN_USE, service_package.usageState)
+
+ def test_create_with_csarid(self):
+ csar_id = '0b667470-e6b3-4ee8-8f08-186317a04dc2'
+ result_data = ServiceDescriptor().create(self.data, csar_id)
+ self.assertEqual(csar_id, result_data['id'])
+ service_package = ServicePackageModel.objects.filter(servicePackageId=csar_id)[0]
+ self.assertIsNotNone(service_package)
+ self.assertEqual(PKG_STATUS.DISABLED, service_package.operationalState)
+ self.assertEqual(PKG_STATUS.CREATED, service_package.onboardingState)
+ self.assertEqual(PKG_STATUS.NOT_IN_USE, service_package.usageState)
+
+ @mock.patch.object(toscaparser, 'parse_sd')
+ def test_parse_serviced_and_save(self, mock_parse_sd):
+ mock_parse_sd.return_value = json.JSONEncoder().encode(self.sd_data)
+ servcie_desc = ServiceDescriptor()
+ csar_id = '0b667470-e6b3-4ee8-8f08-186317a04dc2'
+ servcie_desc.create(self.data, csar_id)
+ VnfPackageModel(vnfPackageId="1", vnfdId="cd557883-ac4b-462d-aa01-421b5fa606b1").save()
+ PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
+ local_file_name = "/test.csar"
+ servcie_desc.parse_serviced_and_save(csar_id, local_file_name)
+
+ service_package = ServicePackageModel.objects.filter(servicePackageId=csar_id)[0]
+ self.assertIsNotNone(service_package)
+
+ def test_delete_single(self):
+ servcie_desc = ServiceDescriptor()
+ csar_id = '0b667470-e6b3-4ee8-8f08-186317a04dc2'
+ servcie_desc.create(self.data, csar_id)
+
+ servcie_desc.delete_single(csar_id)
+ self.assertTrue(len(ServicePackageModel.objects.filter(servicePackageId=csar_id)) == 0)
+ self.assertFalse(ServicePackageModel.objects.filter(servicePackageId=csar_id).exists())
+
+ def test_delete_single_not_exists(self):
+ csar_id = "8000"
+ try:
+ ServiceDescriptor().delete_single(csar_id)
+ except Exception as e:
+ self.assertTrue(isinstance(e, PackageNotFoundException))
+ self.assertEqual("Service package[8000] not Found.", e.args[0])
diff --git a/catalog/packages/tests/test_servicepackage.py b/catalog/packages/tests/test_servicepackage.py
new file mode 100644
index 0000000..241d80d
--- /dev/null
+++ b/catalog/packages/tests/test_servicepackage.py
@@ -0,0 +1,481 @@
+# Copyright (c) 2019, CMCC Technologies. Co., Ltd.
+#
+# 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
+
+from django.test import TestCase, Client
+from mock import mock
+from rest_framework import status
+
+from catalog.packages.biz.sdc_service_package import ServicePackage
+from catalog.packages.const import PKG_STATUS
+from catalog.pub.database.models import ServicePackageModel, VnfPackageModel, PnfPackageModel
+from catalog.pub.exceptions import PackageNotFoundException, PackageHasExistsException, CatalogException
+from catalog.pub.msapi import sdc
+from catalog.pub.utils import toscaparser
+
+PARSER_BASE_URL = "/api/parser/v1"
+
+
+class TestServicePackage(TestCase):
+ """ Test case for Service Package operations"""
+
+ def setUp(self):
+ self.client = Client()
+ ServicePackageModel.objects.filter().delete()
+ self.sd_data = {
+ "inputs": {
+ "sdwanvpnresource_list": [
+ {
+ "sdwanvpn_topology": "",
+ "required": True,
+ "type": "string"
+ },
+ {
+ "sdwansitelan_list": [
+ {
+ "deviceName": "",
+ "required": True,
+ "type": "string",
+ "description": "The device name in the site"
+ }
+ ]
+ }
+ ],
+ "sdwansiteresource_list": [
+ {
+ "sdwansite_controlPoint": "",
+ "required": False,
+ "type": "string",
+ "description": "The control point of the site,only for sd-wan-edge"
+ },
+ {
+ "sdwandevice_list": [
+ {
+ "systemIp": "",
+ "required": False,
+ "type": "string",
+ "description": "The system ip of the device"
+ }
+ ]
+ }
+ ]
+ },
+ "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
+ }
+ }
+ ],
+ "description": "",
+ "graph": {
+ "sdwansiteresource": [
+ "sdwanvpnresource"
+ ],
+ "sdwanvpnresource": []
+ },
+ "basepath": "c:\\users\\cmcc\\appdata\\local\\temp\\tmpn79jwc\\Definitions",
+ "vnfs": [
+ {
+ "vnf_id": "sdwansiteresource",
+ "description": "",
+ "properties": {
+ "sdwandevice_type": "",
+ "sdwandevice_class": "PNF",
+ "multi_stage_design": "False",
+ "min_instances": "1",
+ "sdwansite_controlPoint": "",
+ "id": "cd557883-ac4b-462d-aa01-421b5fa606b1",
+ "sdwansite_longitude": "",
+ "sdwansite_latitude": "",
+ "sdwansite_postcode": "",
+ "sdwansite_type": "",
+ "nf_naming": {
+ "ecomp_generated_naming": True
+ },
+ "sdwansite_emails": "",
+ "sdwansite_role": "",
+ "vnfm_info": "",
+ "sdwansite_address": "",
+ "sdwansite_description": "",
+ "availability_zone_max_count": "1",
+ "sdwansite_name": ""
+ },
+ "dependencies": [],
+ "networks": [],
+ "metadata": {
+ "category": "Configuration",
+ "subcategory": "Configuration",
+ "UUID": "cd557883-ac4b-462d-aa01-421b5fa606b1",
+ "invariantUUID": "c83b621e-e267-4910-a75a-a2a5957296e4",
+ "name": "sdwansiteresource",
+ "customizationUUID": "673dd6b3-3a06-4ef0-8ad0-8c26224b08f7",
+ "resourceVendorRelease": "1.0",
+ "version": "1.0",
+ "resourceVendor": "onap",
+ "resourceVendorModelNumber": "",
+ "type": "VF",
+ "description": "sdwansiteresource"
+ }
+ }
+ ],
+ "vls": [],
+ "service": {
+ "type": "org.openecomp.service.EnhanceService",
+ "requirements": {
+ "sdwanvpnresource.sdwanvpn.dependency": [
+ "sdwanvpnresource",
+ "sdwanvpn.dependency"
+ ],
+ "sdwansiteresource.sdwansitewan.dependency": [
+ "sdwansiteresource",
+ "sdwansitewan.dependency"
+ ],
+ "sdwansiteresource.sdwandevice.dependency": [
+ "sdwansiteresource",
+ "sdwandevice.dependency"
+ ],
+ "sdwanvpnresource.sdwansitelan.dependency": [
+ "sdwanvpnresource",
+ "sdwansitelan.dependency"
+ ],
+ "sdwanvpnresource.sdwanvpn.device": [
+ "sdwanvpnresource",
+ "sdwanvpn.device"
+ ],
+ "sdwansiteresource.sdwansite.device": [
+ "sdwansiteresource",
+ "sdwansite.device"
+ ],
+ "sdwansiteresource.sdwansite.dependency": [
+ "sdwansiteresource",
+ "sdwansite.dependency"
+ ],
+ "sdwanvpnresource.sdwansitelan.device": [
+ "sdwanvpnresource",
+ "sdwansitelan.device"
+ ],
+ "sdwansiteresource.sdwansitewan.device": [
+ "sdwansiteresource",
+ "sdwansitewan.device"
+ ],
+ "sdwansiteresource.sdwandevice.device": [
+ "sdwansiteresource",
+ "sdwandevice.device"
+ ]
+ },
+ "properties": {
+ "descriptor_id": "49ee73f4-1e31-4054-b871-eb9b1c29999b",
+ "designer": "",
+ "invariant_id": "5de07996-7ff0-4ec1-b93c-e3a00bb3f207",
+ "name": "Enhance_Service",
+ "verison": ""
+ },
+ "capabilities": {
+ "sdwansiteresource.sdwandevice.feature": [
+ "sdwansiteresource",
+ "sdwandevice.feature"
+ ],
+ "sdwanvpnresource.sdwanvpn.feature": [
+ "sdwanvpnresource",
+ "sdwanvpn.feature"
+ ],
+ "sdwanvpnresource.sdwanvpn.link": [
+ "sdwanvpnresource",
+ "sdwanvpn.link"
+ ],
+ "sdwansiteresource.sdwansite.feature": [
+ "sdwansiteresource",
+ "sdwansite.feature"
+ ],
+ "sdwansiteresource.sdwansitewan.feature": [
+ "sdwansiteresource",
+ "sdwansitewan.feature"
+ ],
+ "sdwanvpnresource.sdwansitelan.feature": [
+ "sdwanvpnresource",
+ "sdwansitelan.feature"
+ ]
+ },
+ "metadata": {
+ "category": "E2E Service",
+ "serviceType": "",
+ "description": "Enhance_Service",
+ "instantiationType": "A-la-carte",
+ "type": "Service",
+ "environmentContext": "General_Revenue-Bearing",
+ "serviceEcompNaming": True,
+ "UUID": "49ee73f4-1e31-4054-b871-eb9b1c29999b",
+ "ecompGeneratedNaming": True,
+ "serviceRole": "",
+ "invariantUUID": "5de07996-7ff0-4ec1-b93c-e3a00bb3f207",
+ "namingPolicy": "",
+ "name": "Enhance_Service"
+ }
+ },
+ "metadata": {
+ "category": "E2E Service",
+ "serviceType": "",
+ "description": "Enhance_Service",
+ "instantiationType": "A-la-carte",
+ "type": "Service",
+ "environmentContext": "General_Revenue-Bearing",
+ "serviceEcompNaming": True,
+ "UUID": "49ee73f4-1e31-4054-b871-eb9b1c29999b",
+ "ecompGeneratedNaming": True,
+ "serviceRole": "",
+ "invariantUUID": "5de07996-7ff0-4ec1-b93c-e3a00bb3f207",
+ "namingPolicy": "",
+ "name": "Enhance_Service"
+ }
+ }
+ self.asset_data = {
+ "uuid": "1",
+ "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
+ "name": "underlayvpn",
+ "version": "2.0",
+ "toscaModelURL": "/sdc/v1/catalog/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
+ "category": "Volte",
+ "subCategory": "VolteVNF",
+ "resourceType": "VF",
+ "lifecycleState": "CERTIFIED",
+ "distributionStatus": "DISTRIBUTION_APPROVED",
+ "lastUpdaterUserId": "jh0003",
+ "resources": [
+ {
+ "resourceInstanceName": "contrailV2VLANSubInterface 0",
+ "resourceName": "contrailV2VLANSubInterface",
+ "resourceInvariantUUID": "4d31b775-af63-491d-89f1-254e218e7140",
+ "resourceVersion": "1.0",
+ "resoucreType": "CP",
+ "resourceUUID": "cd557883-ac4b-462d-aa01-421b5fa606b1"
+ },
+ {
+ "resourceInstanceName": "Network 0",
+ "resourceName": "Network",
+ "resourceInvariantUUID": "f90f567e-7d7d-4216-af38-6bca0637c59f",
+ "resourceVersion": "1.0",
+ "resoucreType": "VL",
+ "resourceUUID": "m6000_s"
+ }
+ ]
+ }
+
+ def tearDown(self):
+ pass
+
+ ###############################################################
+
+ def test_service_pkg_distribute_when_pkg_exists(self):
+ ServicePackageModel(servicePackageId="1", servicedId="2").save()
+ csar_id = "1"
+ try:
+ ServicePackage().on_distribute(csar_id)
+ except PackageHasExistsException as e:
+ self.assertEqual("Service CSAR(1) already exists.", e.args[0])
+
+ @mock.patch.object(sdc, 'get_asset')
+ def test_service_pkg_distribute_when_fail_get_artifacts(self, mock_get_asset):
+ mock_get_asset.side_effect = CatalogException("Failed to query artifact(services,1) from sdc.")
+ csar_id = "1"
+ try:
+ ServicePackage().on_distribute(csar_id)
+ except Exception as e:
+ self.assertTrue(isinstance(e, CatalogException))
+ self.assertEqual("Failed to query artifact(services,1) from sdc.", e.args[0])
+
+ @mock.patch.object(sdc, 'get_asset')
+ def test_service_pkg_distribute_when_resource_not_distribute(self, mock_get_asset):
+ mock_get_asset.return_value = self.asset_data
+ csar_id = "1"
+ try:
+ ServicePackage().on_distribute(csar_id)
+ except Exception as e:
+ self.assertTrue(isinstance(e, CatalogException))
+ self.assertEqual("Resource (cd557883-ac4b-462d-aa01-421b5fa606b1) is not distributed.", e.args[0])
+
+ @mock.patch.object(sdc, 'get_asset')
+ @mock.patch.object(sdc, 'download_artifacts')
+ def test_service_pkg_distribute_when_fail_download_artifacts(self, mock_get_asset, mock_download_artifacts):
+ mock_get_asset.return_value = self.asset_data
+ mock_download_artifacts.side_effect = CatalogException("Failed to download 1 from sdc.")
+ csar_id = "1"
+ VnfPackageModel(vnfPackageId="cd557883-ac4b-462d-aa01-421b5fa606b1",
+ vnfdId="cd557883-ac4b-462d-aa01-421b5fa606b1").save()
+ PnfPackageModel(pnfPackageId="m6000_s", pnfdId="m6000_s").save()
+
+ try:
+ ServicePackage().on_distribute(csar_id)
+ except Exception as e:
+ self.assertTrue(isinstance(e, CatalogException))
+ self.assertEqual("Failed to download 1 from sdc.", e.args[0])
+
+ @mock.patch.object(sdc, 'get_asset')
+ @mock.patch.object(sdc, 'download_artifacts')
+ @mock.patch.object(toscaparser, 'parse_sd')
+ def test_service_pkg_distribute(self, mock_parse_sd, mock_download_artifacts, mock_get_asset):
+ mock_parse_sd.return_value = json.JSONEncoder().encode(self.sd_data)
+ mock_download_artifacts.return_value = "/test.csar"
+ mock_get_asset.return_value = self.asset_data
+ VnfPackageModel(vnfPackageId="cd557883-ac4b-462d-aa01-421b5fa606b1",
+ vnfdId="cd557883-ac4b-462d-aa01-421b5fa606b1").save()
+ PnfPackageModel(pnfPackageId="m6000_s", pnfdId="m6000_s").save()
+ ServicePackage().on_distribute(csar_id="1")
+
+ service_package = ServicePackageModel.objects.filter(servicePackageId="1").first()
+ self.assertEqual("5de07996-7ff0-4ec1-b93c-e3a00bb3f207", service_package.invariantId)
+ self.assertEqual("Enhance_Service", service_package.servicedName)
+ self.assertEqual(PKG_STATUS.ONBOARDED, service_package.onboardingState)
+ self.assertEqual(PKG_STATUS.ENABLED, service_package.operationalState)
+ self.assertEqual(PKG_STATUS.NOT_IN_USE, service_package.usageState)
+
+ def test_api_service_pkg_distribute_when_pkg_exists(self):
+ ServicePackageModel(servicePackageId="1", servicedId="2").save()
+ resp = self.client.post(
+ PARSER_BASE_URL + "/service_packages", {"csarId": "1"}, format='json')
+ self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+ self.assertEqual("Service CSAR(1) already exists.", resp.data["errorMessage"])
+
+ ###############################################################
+
+ def test_service_pkg_get_all(self):
+ ServicePackageModel(
+ servicePackageId="13",
+ servicedId="2",
+ servicedDesigner="2",
+ servicedVersion="2",
+ servicePackageUri="13.csar",
+ servicedModel="").save()
+ ServicePackageModel(
+ servicePackageId="14",
+ servicedId="3",
+ servicedDesigner="3",
+ servicedVersion="3",
+ servicePackageUri="14.csar",
+ servicedModel="").save()
+ csars = ServicePackage().get_csars()
+ self.assertEqual(2, len(csars))
+
+ def test_api_service_pkg_get_all(self):
+ ServicePackageModel(
+ servicePackageId="13",
+ servicedId="2",
+ servicedDesigner="2",
+ servicedVersion="2",
+ servicePackageUri="13.csar",
+ servicedModel="").save()
+ ServicePackageModel(
+ servicePackageId="14",
+ servicedId="3",
+ servicedDesigner="3",
+ servicedVersion="3",
+ servicePackageUri="14.csar",
+ servicedModel="").save()
+ resp = self.client.get(PARSER_BASE_URL + "/service_packages")
+ self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+ ###############################################################
+
+ def test_service_pkg_get_one(self):
+ ServicePackageModel(
+ servicePackageId="14",
+ servicedId="2",
+ servicedDesigner="3",
+ servicedVersion="4",
+ servicePackageUri="14.csar",
+ servicedModel="").save()
+ csar = ServicePackage().get_csar(14)
+ self.assertEqual(14, csar['csarId'])
+
+ def test_service_pkg_get_one_not_found(self):
+ try:
+ ServicePackage().get_csar(1000)
+ except PackageNotFoundException as e:
+ self.assertEqual("Service package[1000] not Found.", e.args[0])
+
+ def test_api_service_pkg_get_one(self):
+ ServicePackageModel(
+ servicePackageId="14",
+ servicedId="2",
+ servicedDesigner="3",
+ servicedVersion="4",
+ servicePackageUri="14.csar",
+ servicedModel="").save()
+ resp = self.client.get(PARSER_BASE_URL + "/service_packages/14")
+ self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+ def test_api_service_pkg_get_one_not_found(self):
+ resp = self.client.get(PARSER_BASE_URL + "/service_packages/22")
+ self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+ self.assertEqual(
+ {"errorMessage": "Service package[22] not Found.", 'error': 404},
+ resp.data)
+
+ ###############################################################
+
+ def test_service_pkg_normal_delete(self):
+ ServicePackageModel(servicePackageId="8", servicedId="2").save()
+ sp = ServicePackageModel.objects.filter(servicePackageId=8)
+ self.assertEqual(1, len(sp))
+ ServicePackage().delete_csar("8")
+ sp = ServicePackageModel.objects.filter(servicePackageId=8)
+ self.assertEqual(0, len(sp))
+
+ def test_service_pkg_normal_delete_not_found(self):
+ try:
+ ServicePackage().delete_csar("8000")
+ except PackageNotFoundException as e:
+ self.assertEqual("Service package[8000] not Found.", e.args[0])
+
+ def test_api_service_pkg_normal_delete(self):
+ ServicePackageModel(servicePackageId="8", servicedId="2").save()
+ resp = self.client.delete(PARSER_BASE_URL + "/service_packages/8")
+ self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
+
+ ###############################################################
+
+ @mock.patch.object(toscaparser, 'parse_sd')
+ def test_service_pkg_parser(self, mock_parse_sd):
+ ServicePackageModel(servicePackageId="8", servicedId="2").save()
+ mock_parse_sd.return_value = json.JSONEncoder().encode({"a": "b"})
+
+ inputs = []
+ ret = ServicePackage().parse_serviced("8", inputs)
+ self.assertTrue({"model": '{"c": "d"}'}, ret)
+
+ def test_service_pkg_parser_not_found(self):
+ try:
+ csar_id = "8000"
+ inputs = []
+ ServicePackage().parse_serviced(csar_id, inputs)
+ except PackageNotFoundException as e:
+ self.assertEqual("Service CSAR(8000) does not exist.", e.args[0])
+
+ def test_api_service_pkg_parser_not_found(self):
+ query_data = {
+ "csarId": "1",
+ "packageType": "Service",
+ "inputs": "string"
+ }
+ resp = self.client.post(PARSER_BASE_URL + "/parser", query_data, format='json')
+ self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
diff --git a/catalog/packages/tests/test_vnf_package.py b/catalog/packages/tests/test_vnf_package.py
new file mode 100644
index 0000000..b83268a
--- /dev/null
+++ b/catalog/packages/tests/test_vnf_package.py
@@ -0,0 +1,382 @@
+# Copyright 2018 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 os
+import urllib
+import mock
+import shutil
+
+from django.test import TestCase
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from catalog.packages.biz.vnf_package import VnfPackage, VnfPkgUploadThread
+from catalog.packages.const import PKG_STATUS
+from catalog.packages.tests.const import vnfd_data
+from catalog.pub.config.config import CATALOG_ROOT_PATH
+from catalog.pub.database.models import VnfPackageModel
+from catalog.pub.utils import toscaparser
+
+
+class MockReq():
+ def read(self):
+ return "1"
+
+ def close(self):
+ pass
+
+
+class TestVnfPackage(TestCase):
+ def setUp(self):
+ self.client = APIClient()
+
+ def tearDown(self):
+ file_path = os.path.join(CATALOG_ROOT_PATH, "222")
+ if os.path.exists(file_path):
+ shutil.rmtree(file_path)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_upload_vnf_pkg(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rt")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
+ self.assertEqual("zte-hss-1.0", vnf_pkg[0].vnfdId)
+ self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+ def test_upload_vnf_pkg_failed(self):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ )
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ @mock.patch.object(urllib.request, 'urlopen')
+ def test_upload_nf_pkg_from_uri(self, mock_urlopen, mock_parse_vnfd):
+ vnf_pkg = VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
+ mock_urlopen.return_value = MockReq()
+ vnf_pkg_id = vnf_pkg.vnfPackageId
+ VnfPkgUploadThread(req_data, vnf_pkg_id).run()
+ vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222")
+ self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId)
+
+ def test_upload_from_uri_failed(self):
+ req_data = {"username": "123"}
+ response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ def test_create_vnf_pkg(self):
+ req_data = {
+ "userDefinedData": {"a": "A"}
+ }
+ response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
+ resp_data = json.loads(response.content)
+ expect_resp_data = {
+ "id": resp_data.get("id"),
+ "onboardingState": "CREATED",
+ "operationalState": "DISABLED",
+ "usageState": "NOT_IN_USE",
+ "userDefinedData": {"a": "A"},
+ "_links": None # TODO
+ }
+ self.assertEqual(expect_resp_data, resp_data)
+ self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
+ def test_query_single_vnf(self):
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ vnfdId="zte-hss-1.0",
+ vnfVendor="zte",
+ vnfdProductName="hss",
+ vnfSoftwareVersion="1.0.0",
+ vnfdVersion="1.0.0",
+ checksum='{"algorithm":"111", "hash": "11"}',
+ onboardingState="CREATED",
+ operationalState="DISABLED",
+ usageState="NOT_IN_USE",
+ userDefinedData='{"a": "A"}'
+ )
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
+ expect_data = {
+ "id": "222",
+ "vnfdId": "zte-hss-1.0",
+ "vnfProductName": "hss",
+ "vnfSoftwareVersion": "1.0.0",
+ "vnfdVersion": "1.0.0",
+ "checksum": {"algorithm": "111", "hash": "11"},
+ "softwareImages": None,
+ "additionalArtifacts": None,
+ "onboardingState": "CREATED",
+ "operationalState": "DISABLED",
+ "usageState": "NOT_IN_USE",
+ "userDefinedData": {"a": "A"},
+ "_links": None
+ }
+ self.assertEqual(response.data, expect_data)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+ def test_query_single_vnf_failed(self):
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ def test_query_multiple_vnf(self):
+ VnfPackageModel.objects.create(
+ vnfPackageId="111",
+ vnfdId="zte-hss-1.0",
+ vnfVendor="zte",
+ vnfdProductName="hss",
+ vnfSoftwareVersion="1.0.0",
+ vnfdVersion="1.0.0",
+ checksum='{"algorithm":"111", "hash": "11"}',
+ onboardingState="CREATED",
+ operationalState="DISABLED",
+ usageState="NOT_IN_USE",
+ userDefinedData='{"a": "A"}'
+ )
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ vnfdId="zte-hss-1.0",
+ vnfVendor="zte",
+ vnfdProductName="hss",
+ vnfSoftwareVersion="1.0.0",
+ vnfdVersion="1.0.0",
+ checksum='{"algorithm":"111", "hash": "11"}',
+ onboardingState="CREATED",
+ operationalState="DISABLED",
+ usageState="NOT_IN_USE",
+ userDefinedData='{"a": "A"}'
+ )
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
+ expect_data = [
+ {
+ "id": "111",
+ "vnfdId": "zte-hss-1.0",
+ "vnfProductName": "hss",
+ "vnfSoftwareVersion": "1.0.0",
+ "vnfdVersion": "1.0.0",
+ "checksum": {"algorithm": "111", "hash": "11"},
+ "softwareImages": None,
+ "additionalArtifacts": None,
+ "onboardingState": "CREATED",
+ "operationalState": "DISABLED",
+ "usageState": "NOT_IN_USE",
+ "userDefinedData": {"a": "A"},
+ "_links": None
+ },
+ {
+ "id": "222",
+ "vnfdId": "zte-hss-1.0",
+ "vnfProductName": "hss",
+ "vnfSoftwareVersion": "1.0.0",
+ "vnfdVersion": "1.0.0",
+ "checksum": {"algorithm": "111", "hash": "11"},
+ "softwareImages": None,
+ "additionalArtifacts": None,
+ "onboardingState": "CREATED",
+ "operationalState": "DISABLED",
+ "usageState": "NOT_IN_USE",
+ "userDefinedData": {"a": "A"},
+ "_links": None
+ }
+ ]
+ self.assertEqual(response.data, expect_data)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+ def test_delete_single_vnf_pkg(self):
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ vnfdId="zte-hss-1.0",
+ vnfVendor="zte",
+ vnfdProductName="hss",
+ vnfSoftwareVersion="1.0.0",
+ vnfdVersion="1.0.0",
+ checksum='{"algorithm":"111", "hash": "11"}',
+ onboardingState="CREATED",
+ operationalState="DISABLED",
+ usageState="NOT_IN_USE",
+ userDefinedData='{"a": "A"}'
+ )
+ response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(response.data, None)
+
+ def test_delete_when_vnf_pkg_not_exist(self):
+ response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(response.data, None)
+
+ def test_fetch_vnf_pkg(self):
+ with open("vnfPackage.csar", "wt") as fp:
+ fp.writelines("AAAABBBBCCCCDDDD")
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="ONBOARDED",
+ localFilePath="vnfPackage.csar"
+ )
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
+ file_content = ''
+ for data in response.streaming_content:
+ file_content = file_content + data.decode()
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual('AAAABBBBCCCCDDDD', file_content)
+ os.remove("vnfPackage.csar")
+
+ def test_fetch_partical_vnf_pkg(self):
+ with open("vnfPackage.csar", "wt") as fp:
+ fp.writelines("AAAABBBBCCCCDDDD")
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="ONBOARDED",
+ localFilePath="vnfPackage.csar"
+ )
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE="4-7")
+ partial_file_content = ''
+ for data in response.streaming_content:
+ partial_file_content = partial_file_content + data.decode()
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual('BBB', partial_file_content)
+ os.remove("vnfPackage.csar")
+
+ def test_fetch_last_partical_vnf_pkg(self):
+ with open("vnfPackage.csar", "wt") as fp:
+ fp.writelines("AAAABBBBCCCCDDDD")
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="ONBOARDED",
+ localFilePath="vnfPackage.csar"
+ )
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE=" 4-")
+ partial_file_content = ''
+ for data in response.streaming_content:
+ partial_file_content = partial_file_content + data.decode()
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual('BBBBCCCCDDDD', partial_file_content)
+ os.remove("vnfPackage.csar")
+
+ def test_fetch_vnf_pkg_when_pkg_not_exist(self):
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED",
+ localFilePath="vnfPackage.csar"
+ )
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, "create_vnf_pkg")
+ def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
+ mock_create_vnf_pkg.side_effect = TypeError('integer type')
+ req_data = {
+ "userDefinedData": {"a": "A"}
+ }
+ response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, "delete_vnf_pkg")
+ def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
+ mock_delete_vnf_pkg.side_effect = TypeError("integer type")
+ response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, "query_single")
+ def test_query_single_when_catch_exception(self, mock_query_single):
+ mock_query_single.side_effect = TypeError("integer type")
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, "query_multiple")
+ def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
+ mock_query_muitiple.side_effect = TypeError("integer type")
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_upload_when_catch_exception(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.side_effect = TypeError("integer type")
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPkgUploadThread, 'start')
+ def test_upload_from_uri_when_catch_exception(self, mock_start):
+ req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
+ mock_start.side_effect = TypeError("integer type")
+ response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, 'download')
+ def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
+ mock_download.side_effect = TypeError("integer type")
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_fetch_vnf_artifact(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image")
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response.getvalue(), b"ubuntu_16.04\n")
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/1451/artifacts/image")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image1")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
diff --git a/catalog/packages/tests/test_vnf_pkg_subscription.py b/catalog/packages/tests/test_vnf_pkg_subscription.py
new file mode 100644
index 0000000..635b137
--- /dev/null
+++ b/catalog/packages/tests/test_vnf_pkg_subscription.py
@@ -0,0 +1,183 @@
+# Copyright (C) 2019 Verizon. All Rights Reserved
+#
+# 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 uuid
+import mock
+
+from rest_framework.test import APIClient
+from django.test import TestCase
+
+from catalog.pub.database.models import VnfPkgSubscriptionModel
+from .const import vnf_subscription_data
+
+
+class TestNfPackageSubscription(TestCase):
+ def setUp(self):
+ self.client = APIClient()
+ VnfPkgSubscriptionModel.objects.filter().delete()
+ self.vnf_subscription_data = vnf_subscription_data
+
+ def tearDown(self):
+ pass
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_create_vnf_subscription(self, mock_uuid4, mock_requests):
+ temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ response = self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ self.assertEqual(201, response.status_code)
+ self.assertEqual(
+ self.vnf_subscription_data["callbackUri"],
+ response.data["callbackUri"]
+ )
+ self.assertEqual(temp_uuid, response.data["id"])
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_duplicate_subscriptions(self, mock_uuid4, mock_requests):
+ temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ temp1_uuid = "00342b18-a5c7-11e8-998c-bf1755941f12"
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.side_effect = [temp_uuid, temp1_uuid]
+ response = self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ self.assertEqual(201, response.status_code)
+ self.assertEqual(
+ self.vnf_subscription_data["callbackUri"],
+ response.data["callbackUri"]
+ )
+ self.assertEqual(temp_uuid, response.data["id"])
+ temp_uuid = "00442b18-a5c7-11e8-998c-bf1755941f12"
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ response = self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ self.assertEqual(303, response.status_code)
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_get_subscriptions(self, mock_uuid4, mock_requests):
+ temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ response = self.client.get(
+ "/api/vnfpkgm/v1/subscriptions?usageState=IN_USE",
+ format='json'
+ )
+ self.assertEqual(200, response.status_code)
+ self.assertEqual(1, len(response.data))
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_get_subscriptions_with_invalid_params(self, mock_uuid4, mock_requests):
+ temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ response = self.client.get(
+ "/api/vnfpkgm/v1/subscriptions?dummy=dummy",
+ format='json'
+ )
+ self.assertEqual(400, response.status_code)
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_get_subscription_with_id(self, mock_uuid4, mock_requests):
+ temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ response = self.client.get(
+ "/api/vnfpkgm/v1/subscriptions/%s" % temp_uuid,
+ format='json'
+ )
+ self.assertEqual(200, response.status_code)
+ self.assertEqual(temp_uuid, response.data["id"])
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_get_subscription_with_id_not_exists(self, mock_uuid4, mock_requests):
+ temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ dummy_uuid = str(uuid.uuid4())
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ response = self.client.get(
+ "/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid,
+ format='json'
+ )
+ self.assertEqual(404, response.status_code)
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_delete_subscription_with_id(self, mock_uuid4, mock_requests):
+ temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ dummy_uuid = str(uuid.uuid4())
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.status_code = 204
+ mock_uuid4.return_value = temp_uuid
+ self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=self.vnf_subscription_data,
+ format='json'
+ )
+ self.client.get(
+ "/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid,
+ format='json'
+ )
+ response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % temp_uuid)
+ self.assertEqual(204, response.status_code)
+
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_delete_subscription_with_id_not_exists(self, mock_uuid4, mock_requests):
+ dummy_uuid = str(uuid.uuid4())
+ response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid)
+ self.assertEqual(404, response.status_code)
diff --git a/catalog/packages/tests/test_vnfpackage.py b/catalog/packages/tests/test_vnfpackage.py
new file mode 100644
index 0000000..0d8cbad
--- /dev/null
+++ b/catalog/packages/tests/test_vnfpackage.py
@@ -0,0 +1,258 @@
+# 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.test import APIClient
+from django.test import TestCase
+from rest_framework import status
+from catalog.packages.biz.sdc_vnf_package import NfDistributeThread, NfPkgDeleteThread
+from catalog.pub.database.models import JobStatusModel, JobModel
+from catalog.pub.database.models import VnfPackageModel
+from catalog.pub.msapi import sdc
+from catalog.pub.utils import restcall, toscaparser
+from .const import vnfd_data
+
+
+class TestNfPackage(TestCase):
+ def setUp(self):
+ self.client = APIClient()
+ VnfPackageModel.objects.filter().delete()
+ JobModel.objects.filter().delete()
+ JobStatusModel.objects.filter().delete()
+ self.vnfd_data = vnfd_data
+
+ def tearDown(self):
+ pass
+
+ def assert_job_result(self, job_id, job_progress, job_detail):
+ jobs = JobStatusModel.objects.filter(
+ jobid=job_id,
+ progress=job_progress,
+ descp=job_detail)
+ self.assertEqual(1, len(jobs))
+
+ @mock.patch.object(NfDistributeThread, 'run')
+ def test_nf_pkg_distribute_normal(self, mock_run):
+ resp = self.client.post(
+ "/api/catalog/v1/vnfpackages",
+ {
+ "csarId": "1",
+ "vimIds": ["1"]
+ },
+ format='json'
+ )
+ self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
+
+ def test_nf_pkg_distribute_when_csar_already_exist(self):
+ VnfPackageModel(
+ vnfPackageId="1",
+ vnfdId="vcpe_vfw_zte_1_0"
+ ).save()
+ NfDistributeThread(
+ csar_id="1",
+ vim_ids=["1"],
+ lab_vim_id="",
+ job_id="2"
+ ).run()
+ self.assert_job_result("2", 255, "NF CSAR(1) already exists.")
+
+ @mock.patch.object(restcall, 'call_req')
+ @mock.patch.object(sdc, 'download_artifacts')
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_nf_pkg_distribute_when_vnfd_already_exist(self,
+ mock_parse_vnfd,
+ mock_download_artifacts,
+ mock_call_req):
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
+ mock_download_artifacts.return_value = "/home/hss.csar"
+ mock_call_req.return_value = [0, json.JSONEncoder().encode([{
+ "uuid": "1",
+ "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
+ }]), '200']
+ VnfPackageModel(vnfPackageId="2", vnfdId="zte-hss-1.0").save()
+ NfDistributeThread(
+ csar_id="1",
+ vim_ids=["1"],
+ lab_vim_id="",
+ job_id="2"
+ ).run()
+ self.assert_job_result("2", 255, "VNF package(zte-hss-1.0) already exists.")
+
+ @mock.patch.object(restcall, 'call_req')
+ @mock.patch.object(sdc, 'download_artifacts')
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_nf_pkg_distribute_successfully(self,
+ mock_parse_vnfd,
+ mock_download_artifacts,
+ mock_call_req):
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
+ mock_download_artifacts.return_value = "/home/hss.csar"
+ mock_call_req.return_value = [0, json.JSONEncoder().encode([{
+ "uuid": "1",
+ "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
+ }]), '200']
+ NfDistributeThread(
+ csar_id="1",
+ vim_ids=["1"],
+ lab_vim_id="",
+ job_id="4"
+ ).run()
+ self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
+
+ ###############################################################################################################
+
+ @mock.patch.object(NfPkgDeleteThread, 'run')
+ def test_nf_pkg_delete_normal(self, mock_run):
+ resp = self.client.delete("/api/catalog/v1/vnfpackages/1")
+ self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
+
+ def test_nf_pkg_normal_delete(self):
+ VnfPackageModel(
+ vnfPackageId="2",
+ vnfdId="vcpe_vfw_zte_1_0"
+ ).save()
+ NfPkgDeleteThread(
+ csar_id="2",
+ job_id="2"
+ ).run()
+ self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
+
+ def test_nf_pkg_get_all(self):
+ VnfPackageModel(
+ vnfPackageId="3",
+ vnfdId="3",
+ vnfVendor='3',
+ vnfdVersion='3',
+ vnfSoftwareVersion='',
+ vnfPackageUri='',
+ vnfdModel=''
+ ).save()
+ VnfPackageModel(
+ vnfPackageId="4",
+ vnfdId="4",
+ vnfVendor='4',
+ vnfdVersion='4',
+ vnfSoftwareVersion='',
+ vnfPackageUri='',
+ vnfdModel=''
+ ).save()
+ resp = self.client.get("/api/catalog/v1/vnfpackages")
+ self.assertEqual(resp.status_code, status.HTTP_200_OK)
+ expect_data = [
+ {
+ "imageInfo": [],
+ "csarId": "3",
+ "packageInfo": {
+ "csarName": "",
+ "vnfdModel": "",
+ "vnfdProvider": "3",
+ "vnfdId": "3",
+ "downloadUrl": "http://127.0.0.1:8806/static/catalog/3/",
+ "vnfVersion": "",
+ "vnfdVersion": "3",
+ "vnfPackageId": "3"
+ }
+ },
+ {
+ "imageInfo": [],
+ "csarId": "4",
+ "packageInfo": {
+ "csarName": "",
+ "vnfdModel": "",
+ "vnfdProvider": "4",
+ "vnfdId": "4",
+ "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
+ "vnfVersion": "",
+ "vnfdVersion": "4",
+ "vnfPackageId": "4"
+ }
+ }
+ ]
+ self.assertEqual(expect_data, resp.data)
+
+ def test_nf_pkg_get_one(self):
+ VnfPackageModel(
+ vnfPackageId="4",
+ vnfdId="4",
+ vnfVendor='4',
+ vnfdVersion='4',
+ vnfSoftwareVersion='',
+ vnfPackageUri='',
+ vnfdModel=''
+ ).save()
+
+ resp = self.client.get("/api/catalog/v1/vnfpackages/4")
+ self.assertEqual(resp.status_code, status.HTTP_200_OK)
+ expect_data = {
+ "imageInfo": [],
+ "csarId": "4",
+ "packageInfo": {
+ "csarName": "",
+ "vnfdModel": "",
+ "vnfdProvider": "4",
+ "vnfdId": "4",
+ "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
+ "vnfVersion": "",
+ "vnfdVersion": "4",
+ "vnfPackageId": "4"
+ }
+ }
+ self.assertEqual(expect_data, resp.data)
+
+ def test_nf_pkg_get_one_failed(self):
+ VnfPackageModel(
+ vnfPackageId="4",
+ vnfdId="4",
+ vnfVendor='4',
+ vnfdVersion='4',
+ vnfSoftwareVersion='',
+ vnfPackageUri='',
+ vnfdModel=''
+ ).save()
+
+ resp = self.client.get("/api/catalog/v1/vnfpackages/2")
+ self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+ self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
+
+ ###############################################################################################################
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_vnfd_parse_normal(self, mock_parse_vnfd):
+ VnfPackageModel(
+ vnfPackageId="8",
+ vnfdId="10"
+ ).save()
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
+ req_data = {
+ "csarId": "8",
+ "inputs": []
+ }
+ resp = self.client.post(
+ "/api/catalog/v1/parservnfd",
+ req_data,
+ format='json'
+ )
+ self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
+ self.assertEqual({"model": '{"c": "d"}'}, resp.data)
+
+ def test_vnfd_parse_when_csar_not_exist(self):
+ req_data = {"csarId": "1", "inputs": []}
+ resp = self.client.post(
+ "/api/catalog/v1/parservnfd",
+ req_data,
+ format='json'
+ )
+ self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+ self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})