+ 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))