from osm_im.nst import nst as nst_im
from pyangbind.lib.serialise import pybindJSONDecoder
import pyangbind.lib.pybindJSON as pybindJSON
from osm_im.nst import nst as nst_im
from pyangbind.lib.serialise import pybindJSONDecoder
import pyangbind.lib.pybindJSON as pybindJSON
BaseTopic.__init__(self, db, fs, msg, auth)
def check_conflict_on_edit(self, session, final_content, edit_content, _id):
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():
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"])
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.3. restore internal keys
for k, v in internal_keys.items():
final_content[k] = v
if session["force"]:
# 1.3. restore internal keys
for k, v in internal_keys.items():
final_content[k] = v
if session["force"]:
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)
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)
@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)
@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._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)
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:
raise EngineException("ERROR: Unsupported descriptor format. Please, use an ETSI SOL006 descriptor.",
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
try:
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)
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)
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):
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)
def check_conflict_on_del(self, session, _id, db_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):
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)
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
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):
"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)
def check_conflict_on_del(self, session, _id, db_content):
"""
def check_conflict_on_del(self, session, _id, db_content):
"""