validate_input,
vnfpkgop_new_schema,
)
-from osm_nbi.base_topic import BaseTopic, EngineException, get_iterable
+from osm_nbi.base_topic import (
+ BaseTopic,
+ EngineException,
+ get_iterable,
+ detect_descriptor_usage,
+)
from osm_im import etsi_nfv_vnfd, etsi_nfv_nsd
from osm_im.nst import nst as nst_im
from pyangbind.lib.serialise import pybindJSONDecoder
if revision > 1:
try:
self._validate_descriptor_changes(
+ _id,
descriptor_file_name,
current_revision_path,
- proposed_revision_path)
+ proposed_revision_path,
+ )
except Exception as e:
shutil.rmtree(self.fs.path + current_revision_path, ignore_errors=True)
shutil.rmtree(self.fs.path + proposed_revision_path, ignore_errors=True)
)
storage = content["_admin"]["storage"]
if path is not None and path != "$DESCRIPTOR": # artifacts
- if not storage.get("pkg-dir"):
+ if not storage.get("pkg-dir") and not storage.get("folder"):
raise EngineException(
"Packages does not contains artifacts",
http_code=HTTPStatus.BAD_REQUEST,
return indata
- def _validate_descriptor_changes(self,
+ def _validate_descriptor_changes(
+ self,
+ descriptor_id,
descriptor_file_name,
old_descriptor_directory,
- new_descriptor_directory):
- # Todo: compare changes and throw a meaningful exception for the user to understand
+ new_descriptor_directory
+ ):
# Example:
# raise EngineException(
# "Error in validating new descriptor: <NODE> cannot be modified",
def _validate_descriptor_changes(
self,
+ descriptor_id: str,
descriptor_file_name: str,
old_descriptor_directory: str,
new_descriptor_directory: str,
Args:
old_descriptor_directory (str): Directory of descriptor which is in-use
- new_descriptor_directory (str): Directory of directory which is proposed to update (new revision)
+ new_descriptor_directory (str): Directory of descriptor which is proposed to update (new revision)
Returns:
None
EngineException: In case of error when there are unallowed changes
"""
try:
+ # If VNFD does not exist in DB or it is not in use by any NS,
+ # validation is not required.
+ vnfd = self.db.get_one("vnfds", {"_id": descriptor_id})
+ if not vnfd or not detect_descriptor_usage(vnfd, "vnfds", self.db):
+ return
+
+ # Get the old and new descriptor contents in order to compare them.
with self.fs.file_open(
(old_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
) as old_descriptor_file:
+
with self.fs.file_open(
- (new_descriptor_directory, descriptor_file_name), "r"
+ (new_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
) as new_descriptor_file:
- old_content = yaml.load(
- old_descriptor_file.read(), Loader=yaml.SafeLoader
- )
- new_content = yaml.load(
- new_descriptor_file.read(), Loader=yaml.SafeLoader
- )
+
+ old_content = yaml.safe_load(old_descriptor_file.read())
+ new_content = yaml.safe_load(new_descriptor_file.read())
+
+ # If software version has changed, we do not need to validate
+ # the differences anymore.
if old_content and new_content:
if self.find_software_version(
old_content
) != self.find_software_version(new_content):
return
+
disallowed_change = DeepDiff(
self.remove_modifiable_items(old_content),
self.remove_modifiable_items(new_content),
)
+
if disallowed_change:
changed_nodes = functools.reduce(
lambda a, b: a + " , " + b,
).keys()
],
)
+
raise EngineException(
f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
"there are disallowed changes in the vnf descriptor.",
def _validate_descriptor_changes(
self,
+ descriptor_id: str,
descriptor_file_name: str,
old_descriptor_directory: str,
new_descriptor_directory: str,
Args:
old_descriptor_directory: Directory of descriptor which is in-use
- new_descriptor_directory: Directory of directory which is proposed to update (new revision)
+ new_descriptor_directory: Directory of descriptor which is proposed to update (new revision)
Returns:
None
"""
try:
+ # If NSD does not exist in DB, or it is not in use by any NS,
+ # validation is not required.
+ nsd = self.db.get_one("nsds", {"_id": descriptor_id}, fail_on_empty=False)
+ if not nsd or not detect_descriptor_usage(nsd, "nsds", self.db):
+ return
+
+ # Get the old and new descriptor contents in order to compare them.
with self.fs.file_open(
- (old_descriptor_directory, descriptor_file_name), "r"
+ (old_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
) as old_descriptor_file:
+
with self.fs.file_open(
(new_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
) as new_descriptor_file:
- old_content = yaml.load(
- old_descriptor_file.read(), Loader=yaml.SafeLoader
- )
- new_content = yaml.load(
- new_descriptor_file.read(), Loader=yaml.SafeLoader
- )
+
+ old_content = yaml.safe_load(old_descriptor_file.read())
+ new_content = yaml.safe_load(new_descriptor_file.read())
+
if old_content and new_content:
disallowed_change = DeepDiff(
self.remove_modifiable_items(old_content),
self.remove_modifiable_items(new_content),
)
+
if disallowed_change:
changed_nodes = functools.reduce(
lambda a, b: a + ", " + b,
).keys()
],
)
+
raise EngineException(
f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
"there are disallowed changes in the ns descriptor. ",