X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=catalog%2Fpackages%2Ftests%2Ftest_vnf_package.py;h=9f0cef51d2dab2465df4cbbf4529a3e332c1fbd9;hb=79e83b3550bcaa3da0fa86da8925d2f1347a5e2b;hp=fb3a5dee7a6878d01e5867335866ab173ec74835;hpb=14e4e5d6d90f5545fc7fc7c24dc0175bb223380a;p=vfc%2Fnfvo%2Fcatalog.git diff --git a/catalog/packages/tests/test_vnf_package.py b/catalog/packages/tests/test_vnf_package.py index fb3a5dee..9f0cef51 100644 --- a/catalog/packages/tests/test_vnf_package.py +++ b/catalog/packages/tests/test_vnf_package.py @@ -12,16 +12,19 @@ # See the License for the specific language governing permissions and # limitations under the License. -import os import json -import mock +import os import urllib2 +import mock -from rest_framework.test import APIClient 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.packages.biz.vnf_package import VnfPkgUploadThread from catalog.pub.database.models import VnfPackageModel from catalog.pub.utils import toscaparser @@ -37,210 +40,6 @@ class MockReq(): class TestVnfPackage(TestCase): def setUp(self): self.client = APIClient() - self.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" - } - } def tearDown(self): pass @@ -252,14 +51,20 @@ class TestVnfPackage(TestCase): vnfPackageId="222", onboardingState="CREATED" ) - mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data) + 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_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222") - self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId) + 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) - os.remove(vnf_pkg1[0].localFilePath) - os.removedirs(os.path.join(CATALOG_ROOT_PATH, vnf_pkg1[0].vnfPackageId)) + 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(urllib2, 'urlopen') @@ -268,7 +73,7 @@ class TestVnfPackage(TestCase): vnfPackageId="222", onboardingState="CREATED" ) - mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data) + 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 @@ -276,8 +81,10 @@ class TestVnfPackage(TestCase): vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222") self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId) - os.remove(vnf_pkg1[0].localFilePath) - os.removedirs(os.path.join(CATALOG_ROOT_PATH, vnf_pkg1[0].vnfPackageId)) + 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 = { @@ -329,6 +136,10 @@ class TestVnfPackage(TestCase): 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", @@ -410,6 +221,11 @@ class TestVnfPackage(TestCase): 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", "wb") as fp: fp.writelines("AAAABBBBCCCCDDDD") @@ -419,11 +235,11 @@ class TestVnfPackage(TestCase): localFilePath="vnfPackage.csar" ) response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content") - partial_file_content = '' + file_content = '' for data in response.streaming_content: - partial_file_content = partial_file_content + data + file_content = file_content + data self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual('AAAABBBBCCCCDDDD', partial_file_content) + self.assertEqual('AAAABBBBCCCCDDDD', file_content) os.remove("vnfPackage.csar") def test_fetch_partical_vnf_pkg(self): @@ -434,10 +250,114 @@ class TestVnfPackage(TestCase): onboardingState="ONBOARDED", localFilePath="vnfPackage.csar" ) - response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", RANGE="4-7") + 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 self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual('BBBB', partial_file_content) + self.assertEqual('BBB', 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(), "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)