Rename method of delete ns/nf
[vfc/nfvo/catalog.git] / catalog / packages / tests.py
index 15c77bd..d520275 100644 (file)
@@ -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,17 +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")
@@ -460,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))