Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _test(self, test):
if not 'doc' in test or not 'patch' in test:
# incomplete
return
if test.get('disabled', False):
# test is disabled
return
if 'error' in test:
self.assertRaises(
(jsonpatch.JsonPatchException, jsonpatch.JsonPointerException),
jsonpatch.apply_patch, test['doc'], test['patch']
)
else:
try:
res = jsonpatch.apply_patch(test['doc'], test['patch'])
except jsonpatch.JsonPatchException as jpe:
raise Exception(test.get('comment', '')) from jpe
# if there is no 'expected' we only verify that applying the patch
# does not raies an exception
if 'expected' in test:
self.assertEquals(res, test['expected'], test.get('comment', ''))
# Permissions should be mapped as a dict, since jsonpatch doesn't accept
# sets and lists are mapped as JSON arrays (not indexed by value)
permissions = {k: {i: i for i in v} for k, v in permissions.items()}
resource = {"data": data, "permissions": permissions}
# Allow patch permissions without value since key and value are equal on sets
for op in ops:
# 'path' is here since it was validated.
if op["path"].startswith(("/permissions/read/", "/permissions/write/")):
op["value"] = op["path"].split("/")[-1]
try:
result = jsonpatch.apply_patch(resource, ops)
except (jsonpatch.JsonPatchException, jsonpatch.JsonPointerException) as e:
raise ValueError(e)
return result
#. The record is patched.
#. The HTTP response is built with the help of the link factory.
:param pid: Persistent identifier for record.
:param record: Record object.
:returns: The modified record.
"""
data = self.loaders[request.mimetype]()
if data is None:
raise InvalidDataRESTError()
self.check_etag(str(record.revision_id))
try:
record = record.patch(data)
except (JsonPatchException, JsonPointerException):
raise PatchJSONFailureRESTError()
record.commit()
db.session.commit()
return self.make_response(
pid, record, links_factory=self.links_factory)
from oslo_config import cfg
from oslo_utils import reflection
from oslo_utils import uuidutils
import pecan
import wsme
from watcher._i18n import _
from watcher.api.controllers.v1 import versions
from watcher.common import utils
from watcher import objects
CONF = cfg.CONF
JSONPATCH_EXCEPTIONS = (jsonpatch.JsonPatchException,
jsonpatch.JsonPointerException,
KeyError)
def validate_limit(limit):
if limit is None:
return CONF.api.max_limit
if limit <= 0:
# Case where we don't a valid limit value
raise wsme.exc.ClientSideError(_("Limit must be positive"))
if limit and not CONF.api.max_limit:
# Case where we don't have an upper limit
return limit
return min(CONF.api.max_limit, limit)
import jsonpatch
from oslo_utils import uuidutils
import pecan
import wsme
from magnum.common import exception
from magnum.common import utils
import magnum.conf
from magnum.i18n import _
from magnum import objects
CONF = magnum.conf.CONF
JSONPATCH_EXCEPTIONS = (jsonpatch.JsonPatchException,
jsonpatch.JsonPointerException,
KeyError)
DOCKER_MINIMUM_MEMORY = 4 * 1024 * 1024
def validate_limit(limit):
if limit is not None and limit <= 0:
raise wsme.exc.ClientSideError(_("Limit must be positive"))
if limit is not None:
return min(CONF.api.max_limit, limit)
else:
return CONF.api.max_limit
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,
states.VERBS['adopt']: versions.MINOR_17_ADOPT_VERB,
states.VERBS['rescue']: versions.MINOR_38_RESCUE_INTERFACE,
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import jsonpatch
import wsme
from cyborg.common.i18n import _
JSONPATCH_EXCEPTIONS = (jsonpatch.JsonPatchException,
jsonpatch.JsonPointerException,
KeyError)
def apply_jsonpatch(doc, patch):
for p in patch:
if p['op'] == 'add' and p['path'].count('/') == 1:
if p['path'].lstrip('/') not in doc:
msg = _('Adding a new attribute (%s) to the root of '
' the resource is not allowed')
raise wsme.exc.ClientSideError(msg % p['path'])
return jsonpatch.apply_patch(doc, jsonpatch.JsonPatch(patch))