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
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",
"routers": [],
"vnfs": [
{
- "vnf_id": "VFW",
+ "vnf_id": "456",
"description": "",
"properties": {
"plugin_info": "vbrasplugin_1.0",
"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": [
"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",
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")
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))