from catalog.pub.msapi import sdc
from catalog.pub.utils import fileutil
from catalog.pub.utils import toscaparser
+from rest_framework import status
logger = logging.getLogger(__name__)
def ns_delete_csar(csar_id, force_delete):
ret = None
+ nsinstances = []
try:
- ret = NsPackage().delete_csar(csar_id, force_delete)
+ if force_delete:
+ ret = NsPackage().delete_csar(csar_id)
+ return fmt_ns_pkg_rsp(STATUS_SUCCESS, ret[1], "")
+ nsinstances = nfvolcm.get_nsInstances(csar_id)
+ if nsinstances:
+ if len(nsinstances) > 0:
+ return fmt_ns_pkg_rsp(STATUS_FAILED, "NS instances using the CSAR exists!",status.HTTP_412_PRECONDITION_FAILED)
+ ret = NsPackage().delete_csar(csar_id)
+ return fmt_ns_pkg_rsp(STATUS_SUCCESS, ret[1], "")
except CatalogException as e:
return fmt_ns_pkg_rsp(STATUS_FAILED, e.message)
except:
logger.error(traceback.format_exc())
return fmt_ns_pkg_rsp(STATUS_FAILED, str(sys.exc_info()))
- return fmt_ns_pkg_rsp(STATUS_SUCCESS, ret[1], "")
+
def ns_get_csars():
ret = None
return nsd,local_file_name,nsd_json
- def delete_csar(self, csar_id, force_delete):
+ def delete_csar(self, csar_id):
'''
if force_delete:
NSInstModel.objects.filter(nspackage_id=csar_id).delete()
if NSInstModel.objects.filter(nspackage_id=csar_id):
raise CatalogException("CSAR(%s) is in using, cannot be deleted." % csar_id)
'''
- nfvolcm.delete_ns_inst_mock()
+ #nfvolcm.delete_ns_inst_mock()
NSDModel.objects.filter(id=csar_id).delete()
return [0, "Delete CSAR(%s) successfully." % csar_id]
from django.test import Client
from catalog.pub.database.models import NSDModel, NfPackageModel, JobStatusModel, JobModel
from rest_framework import status
+from catalog.pub.msapi import nfvolcm
class PackageTest(unittest.TestCase):
@mock.patch.object(NfDistributeThread, 'get_vnfd')
@mock.patch.object(NsPackage,'get_nsd')
- def test_ns_package_delete(self, mock_get_nsd,mock_get_vnfd):
+ @mock.patch.object(nfvolcm,'get_nsInstances')
+ def test_ns_package_delete(self, mock_get_nsInstances,mock_get_nsd,mock_get_vnfd):
# First distribute a VNF
local_file_name = "/url/local/filename"
self.assert_nsdmodel_result("VCPE_NS", 1)
# Finally delete ns package
+ mock_get_nsInstances.return_value = []
response = self.client.delete("/api/catalog/v1/nspackages/" + str(self.ns_csarId))
self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
self.assertEqual("Delete CSAR(123) successfully.", response.data["statusDescription"], response.content)
self.assert_nsdmodel_result("VCPE_NS", 0)
+ @mock.patch.object(NfDistributeThread, 'get_vnfd')
+ @mock.patch.object(NsPackage,'get_nsd')
+ @mock.patch.object(nfvolcm,'get_nsInstances')
+ def test_ns_package_delete_force(self, mock_get_nsInstances,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()
+ self.assert_nfmodel_result(str(self.nf_csarId), 1)
+
+ # 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_202_ACCEPTED, response.status_code, response.content)
+ self.assertEqual("CSAR(123) distributed successfully.", response.data["statusDescription"], response.content)
+ self.assert_nfmodel_result(str(self.nf_csarId), 1)
+ self.assert_nsdmodel_result("VCPE_NS", 1)
+
+ # Finally delete ns package
+ mock_get_nsInstances.return_value = [{"csarid":"1"},{"csarid":"2"}]
+ response = self.client.delete("/api/catalog/v1/nspackages/%sforce"% str(self.ns_csarId))
+ self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
+ self.assertEqual("Delete CSAR(123) successfully.", response.data["statusDescription"], response.content)
+ self.assert_nsdmodel_result("VCPE_NS", 0)
+
def test_nf_package_delete_error(self):
# Delete it directly
self.assert_nfmodel_result("bb",0)
content=content)
def get_nsInstances(csarid):
- nsInstances=call_lcm("/nlcm/v1/ns ","get")
-
-
+ ret=call_lcm("/nlcm/v1/ns?csarId=%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])
# Mock code because the REST API from nfvolcm to delete ns instance is not implemented
def delete_ns_inst_mock():
import json
import logging
+import os
from catalog.pub.exceptions import CatalogException
from catalog.pub.utils import restcall
ASSETTYPE_SERVICES = "services"
def call_sdc(resource, method, content=''):
+ additional_headers = {
+ 'X-ECOMP-InstanceID': 'VFC',
+ }
return restcall.call_req(base_url=SDC_BASE_URL,
user=SDC_USER,
passwd=SDC_PASSWD,
auth_type=restcall.rest_no_auth,
resource=resource,
method=method,
- content=content)
+ content=content,
+ additional_headers=additional_headers)
+"""
+sample of return value
+[
+ {
+ "uuid": "c94490a0-f7ef-48be-b3f8-8d8662a37236",
+ "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
+ "name": "underlayvpn",
+ "version": "2.0",
+ "toscaModelURL": "/sdc/v1/catalog/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
+ "category": "Volte",
+ "subCategory": "VolteVF",
+ "resourceType": "VF",
+ "lifecycleState": "CERTIFIED",
+ "lastUpdaterUserId": "jh0003"
+ }
+]
+"""
def get_artifacts(asset_type):
resource = "/sdc/v1/catalog/{assetType}"
resource = resource.format(assetType=asset_type)
raise CatalogException("Failed to delete artifacts(%s) from sdc." % artifact_id)
return json.JSONDecoder().decode(ret[1])
-def download_artifacts(download_url, local_path):
- ret = restcall.call_req(base_url=download_url,
+def download_artifacts(download_url, local_path, file_name):
+ additional_headers = {
+ 'X-ECOMP-InstanceID': 'VFC',
+ 'accept': 'application/octet-stream'
+ }
+ ret = restcall.call_req(base_url=SDC_BASE_URL,
user=SDC_USER,
passwd=SDC_PASSWD,
- auth_type=rest_no_auth,
- resource="",
- method="GET")
- # TODO:
+ auth_type=restcall.rest_no_auth,
+ resource=download_url,
+ method="GET",
+ additional_headers=additional_headers)
+ if ret[0] != 0:
+ logger.error("Status code is %s, detail is %s.", ret[2], ret[1])
+ raise CatalogException("Failed to download %s from sdc." % download_url)
+ local_file_name = os.path.join(local_path, file_name)
+ local_file = open(local_file_name, 'wb')
+ local_file.write(ret[1])
+ local_file.close()
+ return local_file_name
+
+
+
+
+
+
+
+
+
+
logger = logging.getLogger(__name__)
-def call_req(base_url, user, passwd, auth_type, resource, method, content=''):
+def call_req(base_url, user, passwd, auth_type, resource, method,
+ content='', additional_headers={}):
callid = str(uuid.uuid1())
logger.debug("[%s]call_req('%s','%s','%s',%s,'%s','%s','%s')" % (
callid, base_url, user, passwd, auth_type, resource, method, content))
http.follow_all_redirects = True
try:
resp, resp_content = http.request(full_url, method=method.upper(), body=content, headers=headers)
- resp_status, resp_body = resp['status'], resp_content.decode('UTF-8')
- logger.debug("[%s][%d]status=%s,resp_body=%s)" % (callid, retry_times, resp_status, resp_body))
+ resp_status, resp_body = resp['status'], resp_content
+ logger.debug("[%s][%d]status=%s)" % (callid, retry_times, resp_status))
+ if headers['accept'] == 'application/json':
+ resp_body = resp_content.decode('UTF-8')
+ logger.debug("resp_body=%s", resp_body)
if resp_status in status_ok_list:
ret = [0, resp_body, resp_status]
else:
base_url = "http://%s:%s/" % (MSB_SERVICE_IP, MSB_SERVICE_PORT)
return call_req(base_url, "", "", rest_no_auth, resource, method, content)
+def upload_by_msb(resource, method, file_data={}):
+ headers = {'Content-Type': 'application/octet-stream'}
+ full_url = "http://%s:%s/%s" % (MSB_SERVICE_IP, MSB_SERVICE_PORT, resource)
+ http = httplib2.Http()
+ resp, resp_content = http.request(full_url,
+ method=method.upper(), body=file_data, headers=headers)
+ resp_status, resp_body = resp['status'], resp_content.decode('UTF-8')
+ if resp_status not in status_ok_list:
+ logger.error("Status code is %s, detail is %s.", resp_status, resp_body)
+ return [1, "Failed to upload file.", resp_status]
+ logger.debug("resp_body=%s", resp_body)
+ return [0, resp_body, resp_status]
def combine_url(base_url, resource):
full_url = None
"vnfVersion": {\r
"type": "string",\r
"description": "VNF Software version"\r
+ },\r
+ "downloadUri":{\r
+ "type": "string",\r
+ "description": "The download uri of VNF package"\r
}\r
}\r
},\r
Code Review / vfc / nfvo / catalog.git / commitdiff