import yaml
import json
import importlib
+import copy
# import logging
from hashlib import md5
from osm_common.dbbase import DbException, deep_update_rfc7396
from osm_im.nst import nst as nst_im
from pyangbind.lib.serialise import pybindJSONDecoder
import pyangbind.lib.pybindJSON as pybindJSON
+from osm_nbi.utils import deep_update_dict
__author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
BaseTopic.__init__(self, db, fs, msg, auth)
def check_conflict_on_edit(self, session, final_content, edit_content, _id):
- super().check_conflict_on_edit(session, final_content, edit_content, _id)
+ final_content = super().check_conflict_on_edit(session, final_content, edit_content, _id)
def _check_unique_id_name(descriptor, position=""):
for desc_key, desc_item in descriptor.items():
internal_keys[k] = final_content.pop(k)
storage_params = internal_keys["_admin"].get("storage")
serialized = self._validate_input_new(final_content, storage_params, session["force"])
+
# 1.2. modify final_content with a serialized version
- final_content.clear()
- final_content.update(serialized)
+ final_content = copy.deepcopy(serialized)
# 1.3. restore internal keys
for k, v in internal_keys.items():
final_content[k] = v
if session["force"]:
- return
+ return final_content
+
# 2. check that this id is not present
if "id" in edit_content:
_filter = self._get_project_filter(session)
+
_filter["id"] = final_content["id"]
_filter["_id.neq"] = _id
+
if self.db.get_one(self.topic, _filter, fail_on_empty=False):
raise EngineException("{} with id '{}' already exists for this project".format(self.topic[:-1],
final_content["id"]),
HTTPStatus.CONFLICT)
+ return final_content
+
@staticmethod
def format_on_new(content, project_id=None, make_public=False):
BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
self._update_input_with_kwargs(indata, kwargs)
deep_update_rfc7396(current_desc, indata)
- self.check_conflict_on_edit(session, current_desc, indata, _id=_id)
+ current_desc = self.check_conflict_on_edit(session, current_desc, indata, _id=_id)
current_desc["_admin"]["modified"] = time()
self.db.replace(self.topic, _id, current_desc)
self.fs.dir_rename(temp_folder, _id)
raise EngineException("ERROR: Unsupported descriptor format. Please, use an ETSI SOL006 descriptor.",
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
try:
- virtual_compute_descriptors = data.get('virtual-compute-desc')
- virtual_storage_descriptors = data.get('virtual-storage-desc')
myvnfd = etsi_nfv_vnfd.etsi_nfv_vnfd()
pybindJSONDecoder.load_ietf_json({'etsi-nfv-vnfd:vnfd': data}, None, None, obj=myvnfd,
path_helper=True, skip_unknown=force)
out = pybindJSON.dumps(myvnfd, mode="ietf")
desc_out = self._remove_envelop(yaml.safe_load(out))
desc_out = self._remove_yang_prefixes_from_descriptor(desc_out)
- if virtual_compute_descriptors:
- desc_out['virtual-compute-desc'] = virtual_compute_descriptors
- if virtual_storage_descriptors:
- desc_out['virtual-storage-desc'] = virtual_storage_descriptors
- return desc_out
+ return deep_update_dict(data, desc_out)
except Exception as e:
raise EngineException("Error in pyangbind validation: {}".format(str(e)),
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
return clean_indata
def check_conflict_on_edit(self, session, final_content, edit_content, _id):
- super().check_conflict_on_edit(session, final_content, edit_content, _id)
+ final_content = super().check_conflict_on_edit(session, final_content, edit_content, _id)
# set type of vnfd
contains_pdu = False
elif contains_vdu:
final_content["_admin"]["type"] = "vnfd"
# if neither vud nor pdu do not fill type
+ return final_content
def check_conflict_on_del(self, session, _id, db_content):
"""
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
def check_conflict_on_edit(self, session, final_content, edit_content, _id):
- super().check_conflict_on_edit(session, final_content, edit_content, _id)
+ final_content = super().check_conflict_on_edit(session, final_content, edit_content, _id)
self._check_descriptor_dependencies(session, final_content)
+ return final_content
+
def check_conflict_on_del(self, session, _id, db_content):
"""
Check that there is not any NSR that uses this NSD. Only NSRs belonging to this project are considered. Note
"existing nsd".format(nsd_id), http_code=HTTPStatus.CONFLICT)
def check_conflict_on_edit(self, session, final_content, edit_content, _id):
- super().check_conflict_on_edit(session, final_content, edit_content, _id)
+ final_content = super().check_conflict_on_edit(session, final_content, edit_content, _id)
self._check_descriptor_dependencies(session, final_content)
+ return final_content
def check_conflict_on_del(self, session, _id, db_content):
"""