from catalog.pub.config.config import CATALOG_ROOT_PATH
from catalog.pub.database.models import VnfPackageModel
from catalog.pub.exceptions import CatalogException
-from catalog.pub.msapi import nslcm
from catalog.pub.msapi import sdc
from catalog.pub.utils import fileutil
from catalog.pub.utils import toscaparser
job_id=self.job_id)
JobUtil.add_job_status(self.job_id, 5, "Start to delete CSAR(%s)." % self.csar_id)
-
- if self.force_delete:
- nslcm.delete_nf_inst(self.csar_id)
- else:
- nfinstances = nslcm.get_vnfInstances(self.csar_id)
- if nfinstances and len(nfinstances) > 0:
- raise CatalogException("NfInst by csar(%s) exists, cannot delete." % self.csar_id)
-
- JobUtil.add_job_status(self.job_id, 50, "Delete CSAR(%s) from Database." % self.csar_id)
-
- if not VnfPackageModel.objects.filter(vnfPackageId=self.csar_id):
- JobUtil.add_job_status(self.job_id, 100, "Error! CSAR(%s) does not exist." % self.csar_id)
- return
-
VnfPackageModel.objects.filter(vnfPackageId=self.csar_id).delete()
- JobUtil.add_job_status(self.job_id, 80, "Delete local CSAR(%s) file." % self.csar_id)
+ JobUtil.add_job_status(self.job_id, 50, "Delete local CSAR(%s) file." % self.csar_id)
csar_save_path = os.path.join(CATALOG_ROOT_PATH, self.csar_id)
fileutil.delete_dirs(csar_save_path)
pkg_info["vnfdVersion"] = nf_pkg[0].vnfdVersion
pkg_info["vnfVersion"] = nf_pkg[0].vnfSoftwareVersion
-
- #vnf_insts = NfInstModel.objects.filter(package_id=csar_id)
- vnf_insts = nslcm.get_vnfInstances()
- vnf_inst_info = [{"vnfInstanceId": vnf_inst["vnfInstanceId"],
- "vnfInstanceName": vnf_inst["vnfInstanceName"]} for vnf_inst in vnf_insts]
-
return [0, {"csarId": csar_id,
"packageInfo": pkg_info,
- "imageInfo": [],
- "vnfInstanceInfo": vnf_inst_info}]
+ "imageInfo": []}]
from catalog.packages.nf_package import NfPkgDeleteThread
from catalog.packages.ns_package import NsPackage
from catalog.pub.database.models import NSPackageModel, VnfPackageModel, JobStatusModel
-from catalog.pub.msapi import nslcm
class PackageTest(unittest.TestCase):
NSPackageModel.objects.all().delete()
JobStatusModel.objects.all().delete()
-
- def test_nfpackages_get(self):
- response = self.client.get("/api/catalog/v1/vnfpackages")
- self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
-
- nsdModel = NSPackageModel.objects.filter(nsdId="VCPE_NS")
- self.assertEqual(len(nsdModel),0)
-
-
- @mock.patch.object(NfDistributeThread, 'get_vnfd')
- def test_nf_distribute(self, mock_get_vnfd):
- local_file_name = "/url/local/filename"
- vnfd = json.JSONEncoder().encode(vnfd_json)
- mock_get_vnfd.return_value = vnfd_json,local_file_name,vnfd
-
- NfDistributeThread("dd", ["1"], "1", "5").run()
- self.assert_job_result("5", 100, "CSAR(dd) distribute successfully.")
- VnfPackageModel.objects.filter(vnfPackageId="dd").delete()
-
-
- @mock.patch.object(NfDistributeThread, 'get_vnfd')
- @mock.patch.object(nslcm,'get_vnfInstances')
- def test_nf_package_delete(self,mock_get_vnfInstances,mock_get_vnfd):
- # First distribute a VNF
- local_file_name = "/url/local/filename"
- vnfd = json.JSONEncoder().encode(vnfd_json)
- mock_get_vnfd.return_value = 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 the vnf
- NfPkgDeleteThread("bb", "6", False).run()
- self.assert_nfmodel_result("bb",0)
- self.assert_job_result("6", 100, "Delete CSAR(bb) successfully.")
-
- @mock.patch.object(NfDistributeThread, 'get_vnfd')
- @mock.patch.object(nslcm,'get_vnfInstances')
- def test_nf_package_delete_force(self,mock_get_vnfInstances,mock_get_vnfd):
- # First distribute a VNF
- local_file_name = "/url/local/filename"
- vnfd = json.JSONEncoder().encode(vnfd_json)
- mock_get_vnfd.return_value = 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 the package by force
- NfPkgDeleteThread("bb", "6", True).run()
- self.assert_nfmodel_result("bb",0)
- self.assert_job_result("6", 100, "Delete CSAR(bb) successfully.")
-
- 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 = NSPackageModel.objects.filter(nsdId = nsd_id)
- self.assertEquals(size, len(nsdmodels))
-
- def assert_nfmodel_result(self,csar_id,size):
- vnfdmodels = VnfPackageModel.objects.filter(vnfPackageId = csar_id)
- self.assertEquals(size, len(vnfdmodels))
nsd_json = {
"inputs": {
+++ /dev/null
-# Copyright 2017 ZTE Corporation.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import json
-import logging
-from catalog.pub.exceptions import CatalogException
-from catalog.pub.utils import restcall
-from catalog.pub.utils.restcall import req_by_msb
-
-logger = logging.getLogger(__name__)
-
-
-def get_nsInstances(csarid):
- ret=req_by_msb("/nslcm/v1/ns?nsPackageId=%s"% csarid, "GET")
- if ret[0] != 0:
- logger.error("Status code is %s, detail is %s.", ret[2], ret[1])
- raise CatalogException("Failed to query NS Instances(%s) from NSLCM." % csarid)
- return json.JSONDecoder().decode(ret[1])
-
-
-def get_vnfInstances(csarid):
- ret=req_by_msb("/nslcm/v1/vnfs?vnfPackageId=%s"% csarid, "GET")
- if ret[0] != 0:
- logger.error("Status code is %s, detail is %s.", ret[2], ret[1])
- raise CatalogException("Failed to query VNF Instances(%s) from NSLCM." % csarid)
- return json.JSONDecoder().decode(ret[1])
-
-def delete_all_nsinst(csarid):
- nsinstances = get_nsInstances(csarid)
- for ns in nsinstances:
- nsInstanceId = ns["nsInstanceId"]
- ret=req_by_msb("/nslcm/v1/ns/%s" % nsInstanceId,"delete")
- if ret[0] != 0:
- logger.error("Status code is %s, detail is %s.", ret[2], ret[1])
- raise CatalogException("Failed to delete NS Instances(%s) from NSLCM." % nsInstanceId)
-
- return [0,'success']
-
-def delete_nf_inst(csar_id):
- #vnf_instance = get_vnfInstances(csar_id)
- # REST API from nslcm to delete nf instance is not implemented
- # ret=req_by_msb("/nslcm/v1/nf/%s" % csar_id,"delete")
- return [0,'success']
-
-# def delete_ns(asset_type):
-# resource = "/nfvolcm/v1/ns/"
-# resource = resource.format(assetType=asset_type)
-# ret = req_by_msb(resource, "DELETE")
-# if ret[0] != 0:
-# logger.error("Status code is %s, detail is %s.", ret[2], ret[1])
-# raise CatalogException("Failed to query artifacts(%s) from sdc." % asset_type)
-# return json.JSONDecoder().decode(ret[1])
Code Review / vfc / nfvo / catalog.git / commitdiff