Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
with app.app_context():
created_community = Community.create_community(**community_metadata)
block_schema = BlockSchema.create_block_schema(created_community.id, 'abc')
block_schema_id = block_schema.id
db.session.commit()
retrieved = BlockSchema.get_block_schema(block_schema_id)
retrieved.patch([{'op': 'replace', 'path': '/name', 'value': 'patched'}])
db.session.commit()
with app.app_context():
patched = BlockSchema.get_block_schema(block_schema_id)
assert block_schema_id == patched.id
assert getattr(patched, 'name') == 'patched'
with pytest.raises(JsonPatchConflict):
patched.patch([{'op': 'replace', 'path': '/non_exist_name', 'value': None}])
assert getattr(patched, 'name') == 'patched'
db.session.commit()
with app.app_context():
retrieved = Community.get(id=community_id)
retrieved.patch(community_patch)
db.session.commit()
with app.app_context():
patched = Community.get(id=community_id)
assert community_id == patched.id
assert patched.updated
for field, value in patched_community_metadata.items():
assert getattr(patched, field) == value
# test invalid or conflicting patchs
with pytest.raises(JsonPatchConflict):
patched.patch([{
'op': 'replace',
'path': '/name',
'value': 'this should not be applied'
}, {
'op': 'replace',
'path': '/non-existing-field',
'value': 'random value'
}])
with pytest.raises(InvalidJsonPatch):
patched.patch({'whatever': 'key'})
with pytest.raises(InvalidCommunityError):
patched.patch([{
'op': 'replace',
'path': '/name',
'value': None
def test_replace_missing(self):
src = {"foo": 1}
patch_obj = [ { "op": "replace", "path": "/bar", "value": 10} ]
self.assertRaises(jsonpatch.JsonPatchConflict, jsonpatch.apply_patch, src, patch_obj)
def clean(self, deposit_id, version_id, *args, **kwargs):
"""Undo metadata extraction."""
# 1. Revert patch on record
recid = str(PersistentIdentifier.get(
'depid', deposit_id).object_uuid)
patch = [{
'op': 'remove',
'path': '/_cds/extracted_metadata',
}]
validator = 'cds.modules.records.validators.PartialDraft4Validator'
try:
patch_record(recid=recid, patch=patch, validator=validator)
except jsonpatch.JsonPatchConflict as c:
logger.warning(
'Failed to apply JSON Patch to deposit {0}: {1}'.format(
recid, c))
# Delete tmp file if any
obj = as_object_version(version_id)
temp_location = obj.get_tags().get('temp_location', None)
if temp_location:
shutil.rmtree(temp_location)
ObjectVersionTag.delete(obj, 'temp_location')
db.session.commit()
raise InvalidJsonPatch(
"The operation does not contain a 'value' member")
subobj, part = self.pointer.to_last(obj)
if part is None:
return value
if isinstance(subobj, MutableSequence):
if part > len(subobj) or part < 0:
raise JsonPatchConflict("can't replace outside of list")
elif isinstance(subobj, MutableMapping):
if part not in subobj:
msg = "can't replace non-existent object '{0}'".format(part)
raise JsonPatchConflict(msg)
else:
raise TypeError("invalid document type {0}".format(type(subobj)))
subobj[part] = value
return obj
def apply(self, obj):
try:
value = self.operation["value"]
except KeyError as ex:
raise InvalidJsonPatch(
"The operation does not contain a 'value' member")
subobj, part = self.pointer.to_last(obj)
if part is None:
return value
if isinstance(subobj, MutableSequence):
if part > len(subobj) or part < 0:
raise JsonPatchConflict("can't replace outside of list")
elif isinstance(subobj, MutableMapping):
if part not in subobj:
msg = "can't replace non-existent object '{0}'".format(part)
raise JsonPatchConflict(msg)
else:
raise TypeError("invalid document type {0}".format(type(subobj)))
subobj[part] = value
return obj
def apply(self, obj):
try:
from_ptr = JsonPointer(self.operation['from'])
except KeyError as ex:
raise InvalidJsonPatch(
"The operation does not contain a 'from' member")
subobj, part = from_ptr.to_last(obj)
try:
value = copy.deepcopy(subobj[part])
except (KeyError, IndexError) as ex:
raise JsonPatchConflict(str(ex))
obj = AddOperation({
'op': 'add',
'path': self.location,
'value': value
}).apply(obj)
return obj
# check to make sure the request has the right Content-Type
if (pecan.request.content_type is None or
pecan.request.content_type != 'application/json-patch+json'):
raise exception.UnsupportedMediaType(
name=pecan.request.content_type,
method='PATCH')
try:
patch = jsonpatch.JsonPatch.from_string(pecan.request.body)
patched_obj = patch.apply(plan_obj.refined_content())
db_obj = handler.update(uuid, patched_obj)
except KeyError:
# a key error indicates one of the patch operations is missing a
# component
raise exception.BadRequest(reason=MAL_PATCH_ERR)
except jsonpatch.JsonPatchConflict:
raise exception.Unprocessable
except jsonpatch.JsonPatchException as jpe:
raise JsonPatchProcessingException(reason=str(jpe))
return fluff_plan(db_obj.refined_content(), db_obj.uuid)
def apply(self, obj):
try:
value = self.operation["value"]
except KeyError as ex:
raise InvalidJsonPatch(
"The operation does not contain a 'value' member")
subobj, part = self.pointer.to_last(obj)
if isinstance(subobj, MutableSequence):
if part == '-':
subobj.append(value) # pylint: disable=E1103
elif part > len(subobj) or part < 0:
raise JsonPatchConflict("can't insert outside of list")
else:
subobj.insert(part, value) # pylint: disable=E1103
elif isinstance(subobj, MutableMapping):
if part is None:
obj = value # we're replacing the root
else:
subobj[part] = value
else:
raise TypeError("invalid document type {0}".format(type(subobj)))
return obj
from ironic import api
from ironic.api.controllers.v1 import versions
from ironic.api import types as atypes
from ironic.common import exception
from ironic.common import faults
from ironic.common.i18n import _
from ironic.common import policy
from ironic.common import states
from ironic.common import utils
from ironic import objects
CONF = cfg.CONF
_JSONPATCH_EXCEPTIONS = (jsonpatch.JsonPatchConflict,
jsonpatch.JsonPatchException,
jsonpatch.JsonPointerException,
KeyError,
IndexError)
# Minimum API version to use for certain verbs
MIN_VERB_VERSIONS = {
# v1.4 added the MANAGEABLE state and two verbs to move nodes into
# and out of that state. Reject requests to do this in older versions
states.VERBS['manage']: versions.MINOR_4_MANAGEABLE_STATE,
states.VERBS['provide']: versions.MINOR_4_MANAGEABLE_STATE,
states.VERBS['inspect']: versions.MINOR_6_INSPECT_STATE,
states.VERBS['abort']: versions.MINOR_13_ABORT_VERB,
states.VERBS['clean']: versions.MINOR_15_MANUAL_CLEAN,