import json
import logging
import os
-
import threading
import traceback
import urllib
from catalog.packages import const
from catalog.packages.biz.common import parse_file_range, read, save
-from catalog.packages.biz.notificationsutil import prepare_vnfpkg_notification, NotificationsUtil
-from catalog.pub.config.config import CATALOG_ROOT_PATH
+from catalog.packages.biz.notificationsutil import PkgNotifications
+from catalog.pub.config.config import CATALOG_ROOT_PATH, MSB_BASE_URL
from catalog.pub.database.models import VnfPackageModel, NSPackageModel
from catalog.pub.exceptions import CatalogException, ResourceNotFoundException
from catalog.pub.utils import fileutil, toscaparser
# logger.error("VNF package(%s) is not CREATED" % vnf_pkg_id)
# raise CatalogException("VNF package(%s) is not CREATED" % vnf_pkg_id)
vnf_pkg.update(onboardingState=const.PKG_STATUS.UPLOADING)
- send_notification(vnf_pkg_id, const.PKG_NOTIFICATION_TYPE.ONBOARDING,
- const.PKG_CHANGE_TYPE.OP_STATE_CHANGE)
local_file_name = save(remote_file, vnf_pkg_id)
logger.info('VNF package(%s) has been uploaded.' % vnf_pkg_id)
logger.info('VNF packge(%s) has been uploaded.' % self.vnf_pkg_id)
+def get_mfile_data(path):
+ logger.debug('get_mfile_data path %s' % path)
+ files = fileutil.filter_files(path, '.mf')
+ if files:
+ src_file = os.path.join(path, files[0])
+ src_dict_list = []
+ with open(src_file, 'r') as f:
+ data = f.readlines()
+ for line in data:
+ if line.strip() == "":
+ continue
+ src_dict = {}
+ k, v = line.split(':', maxsplit=1)
+ if k.strip() in ["Source", "Algorithm", "Hash"]:
+ if k.strip() == "Source" and src_dict:
+ src_dict_list.extend(src_dict)
+ src_dict = {}
+ src_dict[k.strip()] = v.strip()
+ print("src_dict:%s" % src_dict)
+ if src_dict:
+ src_dict_list.append(src_dict)
+
+ logger.debug('get_mfile_data: %s' % src_dict_list)
+ return src_dict_list
+
+
+def fill_artifacts_data(vnf_pkg_id):
+ vnf_pkg_path = os.path.join(CATALOG_ROOT_PATH, vnf_pkg_id)
+ if os.path.exists(vnf_pkg_path) is False:
+ return None
+ files = fileutil.filter_files(vnf_pkg_path, '.csar')
+ for filename in files:
+ logger.info('fill_artifacts_data filename (%s)...' % filename)
+ dst_file_path = os.path.join(vnf_pkg_path, "tmp")
+ src_file = os.path.join(vnf_pkg_path, filename)
+ dst_file = os.path.join(dst_file_path, filename)
+ fileutil.recreate_dir(dst_file_path)
+ fileutil.copy(src_file, vnf_pkg_path, dst_file)
+ artifact_vnf_file = fileutil.unzip_file(dst_file, dst_file_path, "")
+ artifacts = get_mfile_data(artifact_vnf_file)
+ if artifacts:
+ return [{
+ "artifactPath": artifact.get("Source", ""),
+ "checksum": {
+ "algorithm": artifact.get("Hash", "Null"),
+ "hash": artifact.get("Algorithm", "Null")
+ }
+ } for artifact in artifacts]
+
+
+def fill_links(pkg_id, is_onboarded=False):
+ self_href = "%s/api/vnfpkgm/v1/vnf_packages/%s" % (MSB_BASE_URL, pkg_id)
+ links = {
+ "self": {"href": self_href},
+ "vnfd": {"href": "%s/%s" % (self_href, "vnfd")},
+ "packageContent": {"href": "%s/%s" % (self_href, "package_content")}
+ }
+ if not is_onboarded:
+ links.pop("vnfd")
+ return links
+
+
def fill_response_data(nf_pkg):
pkg_info = {}
pkg_info["id"] = nf_pkg.vnfPackageId
if nf_pkg.checksum:
pkg_info["checksum"] = json.JSONDecoder().decode(nf_pkg.checksum)
pkg_info["softwareImages"] = None # TODO
- pkg_info["additionalArtifacts"] = None # TODO
+ pkg_info["additionalArtifacts"] = fill_artifacts_data(nf_pkg.vnfPackageId)
pkg_info["onboardingState"] = nf_pkg.onboardingState
pkg_info["operationalState"] = nf_pkg.operationalState
pkg_info["usageState"] = nf_pkg.usageState
if nf_pkg.userDefinedData:
pkg_info["userDefinedData"] = json.JSONDecoder().decode(nf_pkg.userDefinedData)
- pkg_info["_links"] = None # TODO
+ pkg_info["_links"] = fill_links(nf_pkg.vnfPackageId, True)
return pkg_info
localFilePath=vnf_pkg_path,
vnfPackageUri=os.path.split(vnf_pkg_path)[-1]
)
+ send_notification(vnf_pkg_id, const.PKG_NOTIFICATION_TYPE.ONBOARDING,
+ const.PKG_CHANGE_TYPE.OP_STATE_CHANGE)
else:
raise CatalogException("VNF propeties and metadata in VNF Package(id=%s) are empty." % vnf_pkg_id)
logger.info('VNF package(%s) has been processed(done).' % vnf_pkg_id)
def send_notification(pkg_id, type, pkg_change_type, operational_state=None):
- data = prepare_vnfpkg_notification(vnf_pkg_id=pkg_id,
- notification_type=type,
- pkg_change_type=pkg_change_type,
- operational_state=operational_state)
- filters = {
- 'vnfdId': 'vnfd_id',
- 'vnfPkgId': 'vnf_pkg_id'
- }
- logger.debug('Notify request data = %s' % data)
- logger.debug('Notify request filters = %s' % filters)
- NotificationsUtil().send_notification(data, filters, True)
+ notify = PkgNotifications(type, pkg_id, change_type=pkg_change_type,
+ operational_state=operational_state)
+ notify.send_notification()