X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=catalog%2Fpackages%2Ftests.py;h=d52027577c4509052b01f1fe8236b4d48ef2d24e;hb=3a772d431e121dd03dd663eadc099c38557edb75;hp=b2deba66e0920796d81825ab229f60d6bf9e7c34;hpb=dd8ce89b4ac028cf19c86c3afd132fabffe5b703;p=vfc%2Fnfvo%2Fcatalog.git diff --git a/catalog/packages/tests.py b/catalog/packages/tests.py index b2deba66..d5202757 100644 --- a/catalog/packages/tests.py +++ b/catalog/packages/tests.py @@ -17,9 +17,11 @@ import mock import catalog.pub.utils.restcall import json from catalog.packages.ns_package import NsPackage -from catalog.packages import nf_package +from catalog.packages.nf_package import NfPackage +from catalog.packages.nf_package import NfDistributeThread +from catalog.packages.nf_package import NfPkgDeleteThread from django.test import Client -from catalog.pub.database.models import NSDModel, NfPackageModel +from catalog.pub.database.models import NSDModel, NfPackageModel, JobStatusModel, JobModel from rest_framework import status @@ -32,43 +34,45 @@ class PackageTest(unittest.TestCase): self.nf_csarId = 456 self.nsdata = { - "csarId": self.ns_csarId + "csarId": str(self.ns_csarId) } self.nfdata = { - "csarId": self.nf_csarId + "csarId": str(self.nf_csarId) } - self.nsd_json = { - "vnffgs": [ - { - "vnffg_id": "vnffg1", - "description": "", - "members": [ - "path1", - "path2" - ], - "properties": { + self.vnfd_json = { + "metadata": { + "id": "456", "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" - ] - } + "version": "5.16.10", + "vnfd_version": "1.1.0", + "name": "zte_xgw", + "domain_type": "CN", + "vnf_type": "XGW", + "is_shared": "false", + "cross_dc": "false", + "vmnumber_overquota_alarm": "false", + "description": "", + "vnf_extend_type": "driver&script", + "plugin_info": "zte_cn_plugin_v6.16.10", + "script_info": "script/cn.py", + "adjust_vnf_capacity": "true", + "custom_properties": "", + }, + "reserved_total": { + "vmnum": 10, + "vcpunum": 20, + "memorysize": 1000, + "portnum": 30, + "hdsize": 1024, + "shdsize": 2048, + "isreserve": 0, + }, } - ], + + self.nsd_json = { + "inputs": { "sfc_data_network": { "type": "string", @@ -207,7 +211,7 @@ class PackageTest(unittest.TestCase): "routers": [], "vnfs": [ { - "vnf_id": "VFW", + "vnf_id": "456", "description": "", "properties": { "plugin_info": "vbrasplugin_1.0", @@ -226,7 +230,7 @@ class PackageTest(unittest.TestCase): "vmnumber_overquota_alarm": "true", "vnfd_version": "1.0.0", "externalPluginManageNetworkName": "vlan_4007_plugin_net", - "id": "vcpe_vfw_zte_1_0", + "id": "456", "request_reclassification": "false" }, "dependencies": [ @@ -252,7 +256,7 @@ class PackageTest(unittest.TestCase): "is_shared": "false", "adjust_vnf_capacity": "true", "name": "VNAT", - "id": "vcpe_vnat_zte_1", + "id": "456", "vnf_extend_type": "driver", "csarVersion": "v1.0", "csarType": "NFAR", @@ -442,15 +446,16 @@ class PackageTest(unittest.TestCase): self.assertEquals([],response.data) @mock.patch.object(NsPackage,'get_nsd') - def test_ns_distribute(self, mock_get_nsd): + def test_ns_distribute_2(self, mock_get_nsd): local_file_name = "/url/local/filename" nsd = json.JSONEncoder().encode(self.nsd_json) mock_get_nsd.return_value = self.nsd_json,local_file_name,nsd - response = self.client.post("/api/nfvocatalog/v1/nspackages",self.nsdata) - + response = self.client.post("/api/catalog/v1/nspackages",self.nsdata) self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content) - self.assertIsNotNone(NSDModel.objects.filter(id=self.ns_csarId)) + self.assert_nsdmodel_result("VCPE_NS", 0) + self.assertEqual("VNF package(456) is not distributed.", response.data["statusDescription"], response.content) + NSDModel.objects.filter(id="VCPE_NS").delete() def test_nfpackages_get(self): response = self.client.get("/api/catalog/v1/vnfpackages") @@ -458,28 +463,80 @@ class PackageTest(unittest.TestCase): nsdModel = NSDModel.objects.filter(nsd_id="VCPE_NS") self.assertSequenceEqual(nsdModel,[]) - #nsd_id = nsdModel.nsd_id - #self.assertEqual(self.nsd_json["metadata"]["id"], nsdModel.nsd_id) - #self.assertEqual(self.nsd_json["metadata"]["name"], nsdModel.name) - #self.assertEqual(self.nsd_json["metadata"]["version"], nsdModel.version) - #self.assertEqual(self.nsd_json["metadata"]["description"], nsdModel.description) - #self.assertEqual(self.nsd_json["metadata"]["vendor"], nsdModel.vendor) - def test_ns_distribute(self): + + @mock.patch.object(NfDistributeThread, 'get_vnfd') + @mock.patch.object(NsPackage,'get_nsd') + def test_ns_distribute(self, mock_get_nsd,mock_get_vnfd): + # First distribute a VNF + local_file_name = "/url/local/filename" + vnfd = json.JSONEncoder().encode(self.vnfd_json) + mock_get_vnfd.return_value = self.vnfd_json,local_file_name,vnfd + NfDistributeThread(str(self.nf_csarId), ["1"], "1", "4").run() + + # Then distribute a NS associated with the below VNF + local_file_name = "/url/local/filename" + nsd = json.JSONEncoder().encode(self.nsd_json) + mock_get_nsd.return_value = self.nsd_json,local_file_name,nsd response = self.client.post("/api/catalog/v1/nspackages",self.nsdata) - #self.assertEqual(status.HTTP_200_OK, response.status_code, response.content) + self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content) + self.assertEqual("CSAR(123) distributed successfully.", response.data["statusDescription"], response.content) + self.assert_nsdmodel_result("VCPE_NS", 1) + NfPackageModel.objects.filter(vnfdid=str(self.nf_csarId)).delete() + NSDModel.objects.filter(nsd_id="VCPE_NS").delete() + @mock.patch.object(NfDistributeThread, 'get_vnfd') + def test_nf_distribute(self, mock_get_vnfd): + local_file_name = "/url/local/filename" + vnfd = json.JSONEncoder().encode(self.vnfd_json) + mock_get_vnfd.return_value = self.vnfd_json,local_file_name,vnfd + + NfDistributeThread("dd", ["1"], "1", "5").run() + self.assert_job_result("5", 100, "CSAR(dd) distribute successfully.") + NfPackageModel.objects.filter(nfpackageid="dd").delete() - def test_nf_distribute(self): - #response = self.client.post("/api/catalog/v1/vnfpackages",self.nfdata) - #self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content) - pass def test_ns_package_delete(self): response = self.client.delete("/api/catalog/v1/nspackages/" + str(self.ns_csarId)) self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content) - def test_nf_package_delete(self): - #response = self.client.delete("/api/catalog/v1/vnfpackages/" + str(self.nf_csarId)) - #self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content) - pass + def test_nf_package_delete_error(self): + # Delete it directly + NfPkgDeleteThread("bb", "6", False).run() + self.assert_nfmodel_result("bb",0) + + @mock.patch.object(NfDistributeThread, 'get_vnfd') + def test_nf_package_delete(self,mock_get_vnfd): + # First distribute a VNF + local_file_name = "/url/local/filename" + vnfd = json.JSONEncoder().encode(self.vnfd_json) + mock_get_vnfd.return_value = self.vnfd_json,local_file_name,vnfd + + NfDistributeThread("bb", ["1"], "1", "5").run() + self.assert_job_result("5", 100, "CSAR(bb) distribute successfully.") + self.assert_nfmodel_result("bb",1) + + # Then delete it + NfPkgDeleteThread("bb", "6", False).run() + self.assert_nfmodel_result("bb",0) + + 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)) + + def assert_nsdmodel_result(self,nsd_id,size): + nsdmodels = NSDModel.objects.filter( + nsd_id = nsd_id + ) + + self.assertEquals(size, len(nsdmodels)) + + def assert_nfmodel_result(self,csar_id,size): + vnfdmodels = NfPackageModel.objects.filter( + nfpackageid = csar_id + ) + + self.assertEquals(size, len(vnfdmodels))