- for vdu in get_iterable(indata.get("vdu")):
- icp_refs = []
- ecp_refs = []
- for interface in get_iterable(vdu.get("interface")):
- if interface.get("external-connection-point-ref"):
- if interface.get("external-connection-point-ref") in ecp_refs:
- raise EngineException("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
- "is referenced by other interface"
- .format(vdu["id"], interface["name"],
- interface["external-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- ecp_refs.append(interface.get("external-connection-point-ref"))
- for cp in get_iterable(indata.get("connection-point")):
- if cp["name"] == interface["external-connection-point-ref"]:
- break
- else:
- raise EngineException("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
- "must match an existing connection-point"
- .format(vdu["id"], interface["name"],
- interface["external-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- elif interface.get("internal-connection-point-ref"):
- if interface.get("internal-connection-point-ref") in icp_refs:
- raise EngineException("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
- "is referenced by other interface"
- .format(vdu["id"], interface["name"],
- interface["internal-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- icp_refs.append(interface.get("internal-connection-point-ref"))
- for internal_cp in get_iterable(vdu.get("internal-connection-point")):
- if interface["internal-connection-point-ref"] == internal_cp.get("id"):
- break
- else:
- raise EngineException("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
- "must match an existing vdu:internal-connection-point"
- .format(vdu["id"], interface["name"],
- interface["internal-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- # Validate that if descriptor contains charms, artifacts _admin.storage."pkg-dir" is not none
- if vdu.get("vdu-configuration"):
- if vdu["vdu-configuration"].get("juju"):
- if not self._validate_package_folders(storage_params, 'charms'):
- raise EngineException("Charm defined in vnf[id={}]:vdu[id={}] but not present in "
- "package".format(indata["id"], vdu["id"]))
- # Validate that if descriptor contains cloud-init, artifacts _admin.storage."pkg-dir" is not none
- if vdu.get("cloud-init-file"):
- if not self._validate_package_folders(storage_params, 'cloud_init', vdu["cloud-init-file"]):
- raise EngineException("Cloud-init defined in vnf[id={}]:vdu[id={}] but not present in "
- "package".format(indata["id"], vdu["id"]))
- # Validate that if descriptor contains charms, artifacts _admin.storage."pkg-dir" is not none
- if indata.get("vnf-configuration"):
- if indata["vnf-configuration"].get("juju"):
- if not self._validate_package_folders(storage_params, 'charms'):
- raise EngineException("Charm defined in vnf[id={}] but not present in "
- "package".format(indata["id"]))
+ def _validate_vdu_charms_in_package(self, storage_params, vdu, indata):
+ if not vdu.get("vdu-configuration"):
+ return
+ if vdu["vdu-configuration"].get("juju"):
+ if not self._validate_package_folders(storage_params, 'charms'):
+ raise EngineException("Charm defined in vnf[id={}]:vdu[id={}] but not present in "
+ "package".format(indata["id"], vdu["id"]))
+
+ def _validate_vdu_cloud_init_in_package(self, storage_params, vdu, indata):
+ if not vdu.get("cloud-init-file"):
+ return
+ if not self._validate_package_folders(storage_params, 'cloud_init', vdu["cloud-init-file"]):
+ raise EngineException("Cloud-init defined in vnf[id={}]:vdu[id={}] but not present in "
+ "package".format(indata["id"], vdu["id"]))
+
+ def _validate_vnf_charms_in_package(self, storage_params, indata):
+ if not indata.get("vnf-configuration"):
+ return
+ if indata["vnf-configuration"].get("juju"):
+ if not self._validate_package_folders(storage_params, 'charms'):
+ raise EngineException("Charm defined in vnf[id={}] but not present in "
+ "package".format(indata["id"]))
+
+ def _validate_package_folders(self, storage_params, folder, file=None):
+ if not storage_params or not storage_params.get("pkg-dir"):
+ return False
+ else:
+ if self.fs.file_exists("{}_".format(storage_params["folder"]), 'dir'):
+ f = "{}_/{}/{}".format(storage_params["folder"], storage_params["pkg-dir"], folder)
+ else:
+ f = "{}/{}/{}".format(storage_params["folder"], storage_params["pkg-dir"], folder)
+ if file:
+ return self.fs.file_exists("{}/{}".format(f, file), 'file')
+ else:
+ if self.fs.file_exists(f, 'dir'):
+ if self.fs.dir_ls(f):
+ return True
+ return False
+
+ @staticmethod
+ def validate_internal_vlds(indata):