import functools
import logging
import os
+import base64
logger = logging.getLogger(__name__)
vnf = self.model.get_substitution_mappings(tosca)
properties = vnf.get("properties", {})
metadata = vnf.get("metadata", {})
- if properties.get("descriptor_id", "") == "":
- descriptor_id = metadata.get("descriptor_id", "")
- if descriptor_id == "":
- descriptor_id = metadata.get("id", "")
- if descriptor_id == "":
- descriptor_id = metadata.get("UUID", "")
- properties["descriptor_id"] = descriptor_id
-
- if properties.get("descriptor_version", "") == "":
- version = metadata.get("template_version", "")
- if version == "":
- version = metadata.get("version", "")
- properties["descriptor_version"] = version
-
- if properties.get("provider", "") == "":
- provider = metadata.get("template_author", "")
- if provider == "":
- provider = metadata.get("provider", "")
- properties["provider"] = provider
-
- if properties.get("template_name", "") == "":
- template_name = metadata.get("template_name", "")
- if template_name == "":
- template_name = metadata.get("template_name", "")
- properties["template_name"] = template_name
+
+ for key, value in list(properties.items()):
+ if isinstance(value, dict):
+ if value["type"] == "string":
+ properties[key] = value.get("default", "")
+ elif value["type"] == "list":
+ properties[key] = value.get("default", {})
+ else:
+ properties[key] = value.get("default", "")
+ ptype = "descriptor_id"
+ meta_types = ["descriptor_id", "id", "UUID"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "descriptor_version"
+ meta_types = ["template_version", "version"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "provider"
+ meta_types = ["template_author", "provider"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "template_name"
+ meta_types = ["template_name"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "software_version"
+ meta_types = ["software_version"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "product_name"
+ meta_types = ["product_name"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "flavour_description"
+ meta_types = ["flavour_description"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "vnfm_info"
+ meta_types = ["vnfm_info"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ ptype = "flavour_id"
+ meta_types = ["flavour_id"]
+ self._get_property(properties, metadata, ptype, meta_types)
+
+ logger.debug("vnf:%s", vnf)
return vnf
source_path = os.path.join(self.model.basepath, inject_file['source_path'])
with open(source_path, "rb") as f:
source_data = f.read()
- source_data_base64 = source_data.encode("base64")
- inject_file["source_data_base64"] = source_data_base64
+ source_data_base64 = base64.b64encode(source_data)
+ inject_file["source_data_base64"] = source_data_base64.decode()
if isinstance(inject_files, dict):
source_path = os.path.join(self.model.basepath, inject_files['source_path'])
with open(source_path, "rb") as f:
source_data = f.read()
- source_data_base64 = source_data.encode("base64")
- inject_files["source_data_base64"] = source_data_base64
+ source_data_base64 = base64.b64encode(source_data)
+ inject_files["source_data_base64"] = source_data_base64.decode()
virtual_storages = self.model.getRequirementByName(node, 'virtual_storage')
- ret['virtual_storages'] = map(functools.partial(self._trans_virtual_storage), virtual_storages)
- ret['dependencies'] = map(lambda x: self.model.get_requirement_node_name(x), self.model.getNodeDependencys(node))
+ ret['virtual_storages'] = list(map(functools.partial(self._trans_virtual_storage), virtual_storages))
+ ret['dependencies'] = [self.model.get_requirement_node_name(x) for x in self.model.getNodeDependencys(node)]
virtual_compute = self.model.getCapabilityByName(node, 'virtual_compute')
if virtual_compute is not None and 'properties' in virtual_compute:
ret['virtual_compute'] = virtual_compute['properties']
return {"external_cps": external_cps, "forward_cps": forward_cps}
return {}
+ def _get_property(self, properties, metadata, ptype, meta_types):
+ if ptype not in properties or properties[ptype] == "":
+ for mtype in meta_types:
+ data = metadata.get(mtype, "")
+ if data != "":
+ properties[ptype] = data
+
def _trans_virtual_storage(self, virtual_storage):
if isinstance(virtual_storage, str):
return {"virtual_storage_id": virtual_storage}
return vl_ids
def _get_virtal_binding_cp_ids(self, node, nodeTemplates):
- return map(lambda x: x['name'], self._get_virtal_binding_cps(node, nodeTemplates))
+ return [x['name'] for x in self._get_virtal_binding_cps(node, nodeTemplates)]
def _get_virtal_binding_cps(self, node, nodeTemplates):
cps = []
for tmpnode in nodeTemplates:
if 'requirements' in tmpnode:
for item in tmpnode['requirements']:
- for key, value in item.items():
+ for key, value in list(item.items()):
if key.upper().startswith('VIRTUAL_BINDING'):
req_node_name = self.model.get_requirement_node_name(value)
if req_node_name is not None and req_node_name == node['name']:
return cps
def _get_node_vdu_id(self, node):
- vdu_ids = map(lambda x: self.model.get_requirement_node_name(x), self.model.getRequirementByName(node, 'virtual_binding'))
+ vdu_ids = [self.model.get_requirement_node_name(x) for x in self.model.getRequirementByName(node, 'virtual_binding')]
if len(vdu_ids) > 0:
return vdu_ids[0]
return ""
def _get_node_vl_id(self, node):
- vl_ids = map(lambda x: self.model.get_requirement_node_name(x), self.model.getRequirementByName(node, 'virtual_link'))
+ vl_ids = [self.model.get_requirement_node_name(x) for x in self.model.getRequirementByName(node, 'virtual_link')]
if len(vl_ids) > 0:
return vl_ids[0]
return ""
def _buil_cp_vls(self, node):
- return map(lambda x: self._build_cp_vl(x), self.model.getRequirementByName(node, 'virtual_link'))
+ return [self._build_cp_vl(x) for x in self.model.getRequirementByName(node, 'virtual_link')]
def _build_cp_vl(self, req):
cp_vl = {}
if relationship is not None:
properties = self.model.get_prop_from_obj(relationship, 'properties')
if properties is not None and isinstance(properties, dict):
- for key, value in properties.items():
+ for key, value in list(properties.items()):
cp_vl[key] = value
return cp_vl
external_cps = []
if vnf_requirements:
if isinstance(vnf_requirements, dict):
- for key, value in vnf_requirements.items():
+ for key, value in list(vnf_requirements.items()):
if isinstance(value, list) and len(value) > 0:
external_cps.append({"key_name": key, "cpd_id": value[0]})
else:
external_cps.append({"key_name": key, "cpd_id": value})
elif isinstance(vnf_requirements, list):
for vnf_requirement in vnf_requirements:
- for key, value in vnf_requirement.items():
+ for key, value in list(vnf_requirement.items()):
if isinstance(value, list) and len(value) > 0:
external_cps.append({"key_name": key, "cpd_id": value[0]})
else:
def _get_forward_cps(self, vnf_capabilities):
forward_cps = []
if vnf_capabilities:
- for key, value in vnf_capabilities.items():
+ for key, value in list(vnf_capabilities.items()):
if isinstance(value, list) and len(value) > 0:
forward_cps.append({"key_name": key, "cpd_id": value[0]})
else:
Code Review / modeling / etsicatalog.git / blobdiff