612 lines
24 KiB
Python
612 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
"""jsonptr_patch.py -- RFC 6901 JSON Pointer + RFC 6902 JSON Patch.
|
||
|
||
Pure stdlib (json, copy, sys). No third-party imports.
|
||
|
||
Public API
|
||
----------
|
||
JSON Pointer (RFC 6901 s.3/s.4):
|
||
decode_pointer(ref) -> tuple[str, ...] validate + unescape
|
||
unescape_token(tok) -> str '~1' -> '/', '~0' -> '~'
|
||
escape_token(tok) -> str inverse
|
||
pointer_to_string(toks) -> str inverse of decode_pointer
|
||
resolve(doc, tokens) -> value strict read, raises on miss
|
||
|
||
JSON Patch (RFC 6902 s.4):
|
||
json_equal(a, b) -> bool JSON value equality
|
||
apply_patch(doc, patch) -> new document input is NEVER mutated
|
||
|
||
Errors: JsonPointerError and JsonPatchError (both ValueError subclasses).
|
||
|
||
Semantics, measured against the vendored official suite
|
||
json-patch/json-patch-tests (tests.json + spec_tests.json, sha256 pinned in
|
||
test_jsonptr_patch.py):
|
||
|
||
* `add` never creates missing intermediate objects -- add /a/b against {} is
|
||
an error (spec_tests.json case "A.12. Adding to a Non-existent Target").
|
||
* array `add`: index == len(array) is legal, index > len(array) is an error,
|
||
"-" appends.
|
||
* `remove` / `replace` / `copy` / `test` require the target to exist;
|
||
`move` requires `from` to exist and forbids moving into a strict
|
||
descendant of `from`.
|
||
* `test` uses JSON value equality: 1 == 1.0, true != 1, null == null,
|
||
object member order is irrelevant, array element order is significant.
|
||
* the patch is atomic: every op is applied to a deep copy and the copy is
|
||
returned only if all ops succeed; the caller's document is untouched either
|
||
way.
|
||
|
||
CLI:
|
||
python3 jsonptr_patch.py --selftest [--verbose]
|
||
python3 jsonptr_patch.py --selftest-json
|
||
python3 jsonptr_patch.py --doc doc.json --patch patch.json
|
||
|
||
Exit codes: 0 = success, 1 = RFC 6901/6902 error, 2 = usage error.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import sys
|
||
|
||
__all__ = [
|
||
"JsonPointerError", "JsonPatchError",
|
||
"decode_pointer", "unescape_token", "escape_token", "pointer_to_string",
|
||
"resolve", "json_equal", "apply_patch", "run_selftest",
|
||
]
|
||
|
||
PATCH_ERR = 1
|
||
USAGE_ERR = 2
|
||
|
||
|
||
class JsonPointerError(ValueError):
|
||
"""Malformed JSON Pointer string (RFC 6901 s.3/s.4)."""
|
||
|
||
|
||
class JsonPatchError(ValueError):
|
||
"""Malformed patch, or a failing operation (RFC 6902 s.4)."""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# RFC 6901 -- JSON Pointer
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def escape_token(tok: str) -> str:
|
||
"""'~' -> '~0' first, then '/' -> '~1' (order matters: '~/' -> '~0~1')."""
|
||
return str(tok).replace("~", "~0").replace("/", "~1")
|
||
|
||
|
||
def unescape_token(tok: str) -> str:
|
||
"""Inverse of escape_token. '~0' -> '~', '~1' -> '/'; anything else is illegal."""
|
||
out = []
|
||
i, n = 0, len(tok)
|
||
while i < n:
|
||
ch = tok[i]
|
||
if ch != "~":
|
||
out.append(ch)
|
||
i += 1
|
||
continue
|
||
if i + 1 >= n:
|
||
raise JsonPointerError(f"dangling '~' at end of reference token {tok!r}")
|
||
nxt = tok[i + 1]
|
||
if nxt == "0":
|
||
out.append("~")
|
||
elif nxt == "1":
|
||
out.append("/")
|
||
else:
|
||
raise JsonPointerError(
|
||
f"invalid escape '~{nxt}' in reference token {tok!r} "
|
||
"(only ~0 and ~1 are legal)"
|
||
)
|
||
i += 2
|
||
return "".join(out)
|
||
|
||
|
||
def decode_pointer(ref: str) -> tuple[str, ...]:
|
||
"""Split a JSON Pointer string into unescaped reference tokens.
|
||
|
||
Grammar (RFC 6901 s.3): json-pointer = *( "/" reference-token )
|
||
"" -> () the whole document
|
||
"/a/b" -> ("a", "b")
|
||
"a" -> rejected: must be empty or start with "/"
|
||
"/a~2b" -> rejected: '~2' is not a legal escape
|
||
"""
|
||
if not isinstance(ref, str):
|
||
raise JsonPointerError(f"pointer must be a str, got {type(ref).__name__}")
|
||
if ref == "":
|
||
return ()
|
||
if not ref.startswith("/"):
|
||
raise JsonPointerError(f"pointer {ref!r} must be empty or start with '/'")
|
||
return tuple(unescape_token(t) for t in ref[1:].split("/"))
|
||
|
||
|
||
def pointer_to_string(tokens) -> str:
|
||
"""Inverse of decode_pointer: () -> "", ("a", "b") -> "/a/b"."""
|
||
return "".join("/" + escape_token(t) for t in tokens)
|
||
|
||
|
||
def is_array_index(tok: str) -> bool:
|
||
"""RFC 6901 s.4: digits only, and a leading zero is only allowed for '0'.
|
||
|
||
``str.isdigit()`` is NOT sufficient: it is True for Unicode digit glyphs
|
||
such as ``"²"`` (SUPERSCRIPT TWO) and ``"١"`` (ARABIC-INDIC ONE), whose
|
||
``int()`` raises a plain ``ValueError`` -- which would leak past the
|
||
JsonPointerError contract. ``str.isascii()`` pins the check to the ASCII
|
||
digits that ``int()`` actually accepts.
|
||
"""
|
||
if not tok or not tok.isascii() or not tok.isdigit():
|
||
return False
|
||
return tok == "0" or tok[0] != "0"
|
||
|
||
|
||
def _descend(node, tok: str):
|
||
"""Walk one token. Arrays accept an RFC 6901 array-index token."""
|
||
if isinstance(node, dict):
|
||
if tok not in node:
|
||
raise JsonPointerError(f"key {tok!r} not found in object")
|
||
return node[tok]
|
||
if isinstance(node, list):
|
||
if not is_array_index(tok):
|
||
raise JsonPointerError(f"invalid array index {tok!r}")
|
||
idx = int(tok)
|
||
if idx >= len(node):
|
||
raise JsonPointerError(
|
||
f"array index {tok!r} out of range (len={len(node)})"
|
||
)
|
||
return node[idx]
|
||
raise JsonPointerError(
|
||
f"cannot index {type(node).__name__} with {tok!r}: not an object or array"
|
||
)
|
||
|
||
|
||
def resolve(doc, tokens) -> object:
|
||
"""Evaluate a token tuple against `doc`; raise JsonPointerError on any miss."""
|
||
node = doc
|
||
for tok in tokens:
|
||
node = _descend(node, tok)
|
||
return node
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# JSON value equality (used by the `test` operation)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def json_equal(a, b) -> bool:
|
||
"""RFC 6902 `test` semantics.
|
||
|
||
Numbers compare numerically (1 == 1.0); booleans are a distinct type from
|
||
numbers (True != 1); null equals only null; object member order is
|
||
irrelevant; array order is significant.
|
||
"""
|
||
if isinstance(a, bool) or isinstance(b, bool):
|
||
return isinstance(a, bool) and isinstance(b, bool) and a is b
|
||
if a is None or b is None:
|
||
return a is None and b is None
|
||
if isinstance(a, (int, float)) and isinstance(b, (int, float)):
|
||
return a == b
|
||
if isinstance(a, str) or isinstance(b, str):
|
||
return isinstance(a, str) and isinstance(b, str) and a == b
|
||
if isinstance(a, dict):
|
||
if not isinstance(b, dict) or set(a) != set(b):
|
||
return False
|
||
return all(json_equal(a[k], b[k]) for k in a)
|
||
if isinstance(b, dict):
|
||
return False
|
||
if isinstance(a, list):
|
||
if not isinstance(b, list) or len(a) != len(b):
|
||
return False
|
||
return all(json_equal(x, y) for x, y in zip(a, b))
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# RFC 6902 -- JSON Patch
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_OPS = ("add", "remove", "replace", "move", "copy", "test")
|
||
|
||
|
||
def _canonical(value) -> str:
|
||
"""Canonical JSON, used only for error messages."""
|
||
return json.dumps(value, sort_keys=True, ensure_ascii=False,
|
||
separators=(",", ":"))
|
||
|
||
|
||
def _require(cond: bool, msg: str) -> None:
|
||
if not cond:
|
||
raise JsonPatchError(msg)
|
||
|
||
|
||
def _parent_and_token(doc, tokens):
|
||
"""Return (container, final_token) for a non-root pointer, validated."""
|
||
_require(bool(tokens), "operation target must not be the whole document")
|
||
container = resolve(doc, tokens[:-1])
|
||
tok = tokens[-1]
|
||
if isinstance(container, dict):
|
||
_require(tok in container,
|
||
f"path {pointer_to_string(tokens)!r} does not exist")
|
||
elif isinstance(container, list):
|
||
_require(tok != "-", "'-' is not a valid target for this operation")
|
||
_require(is_array_index(tok), f"invalid array index {tok!r}")
|
||
_require(int(tok) < len(container),
|
||
f"array index {tok!r} out of range (len={len(container)})")
|
||
else:
|
||
raise JsonPatchError(
|
||
f"path {pointer_to_string(tokens)!r} targets a scalar"
|
||
)
|
||
return container, tok
|
||
|
||
|
||
def _op_add(doc, tokens, value):
|
||
if not tokens: # root replacement
|
||
return copy.deepcopy(value)
|
||
container = resolve(doc, tokens[:-1]) # strict: parents must exist
|
||
tok = tokens[-1]
|
||
if isinstance(container, list):
|
||
if tok == "-":
|
||
container.append(copy.deepcopy(value))
|
||
return doc
|
||
_require(is_array_index(tok), f"invalid array index {tok!r}")
|
||
idx = int(tok)
|
||
_require(idx <= len(container),
|
||
f"array index {tok!r} is greater than number of items "
|
||
f"in array (len={len(container)})")
|
||
container.insert(idx, copy.deepcopy(value))
|
||
return doc
|
||
if isinstance(container, dict):
|
||
container[tok] = copy.deepcopy(value) # create-or-replace
|
||
return doc
|
||
raise JsonPatchError(
|
||
f"cannot add into {type(container).__name__} at "
|
||
f"{pointer_to_string(tokens)!r}"
|
||
)
|
||
|
||
|
||
def _op_remove(doc, tokens):
|
||
container, tok = _parent_and_token(doc, tokens)
|
||
if isinstance(container, dict):
|
||
del container[tok]
|
||
else:
|
||
del container[int(tok)]
|
||
return doc
|
||
|
||
|
||
def _op_replace(doc, tokens, value):
|
||
if not tokens:
|
||
return copy.deepcopy(value)
|
||
container, tok = _parent_and_token(doc, tokens)
|
||
if isinstance(container, dict):
|
||
container[tok] = copy.deepcopy(value)
|
||
else:
|
||
container[int(tok)] = copy.deepcopy(value)
|
||
return doc
|
||
|
||
|
||
def _op_move(doc, tokens, from_tokens):
|
||
_require(bool(from_tokens), "'move' 'from' must not be the whole document")
|
||
if pointer_to_string(from_tokens) == pointer_to_string(tokens):
|
||
return doc # /a -> /a is a no-op
|
||
# Moving an object into its own descendant would destroy the target path.
|
||
_require(
|
||
not (len(tokens) > len(from_tokens)
|
||
and tuple(tokens[:len(from_tokens)]) == from_tokens),
|
||
f"cannot move {pointer_to_string(from_tokens)!r} into its own "
|
||
f"descendant {pointer_to_string(tokens)!r}",
|
||
)
|
||
moved = copy.deepcopy(resolve(doc, from_tokens))
|
||
_op_remove(doc, from_tokens)
|
||
_op_add(doc, tokens, moved) # `to` resolves on the post-removal doc
|
||
return doc
|
||
|
||
|
||
def _op_copy(doc, tokens, from_tokens):
|
||
_require(bool(from_tokens), "'copy' 'from' must not be the whole document")
|
||
src = resolve(doc, from_tokens)
|
||
_op_add(doc, tokens, copy.deepcopy(src))
|
||
return doc
|
||
|
||
|
||
def _op_test(doc, tokens, value):
|
||
actual = resolve(doc, tokens)
|
||
_require(
|
||
json_equal(actual, value),
|
||
f"test failed at {pointer_to_string(tokens)!r}: "
|
||
f"expected {_canonical(value)!r}, got {_canonical(actual)!r}",
|
||
)
|
||
return doc
|
||
|
||
|
||
def _apply(work, op):
|
||
if not isinstance(op, dict):
|
||
raise JsonPatchError(
|
||
f"each patch entry must be an object, got {type(op).__name__}"
|
||
)
|
||
name = op.get("op")
|
||
_require(isinstance(name, str) and name in _OPS,
|
||
f"unrecognized op {name!r}")
|
||
_require("path" in op, f"missing 'path' parameter to {name!r}")
|
||
if name in ("add", "replace", "test"):
|
||
_require("value" in op, f"missing 'value' parameter to {name!r}")
|
||
if name in ("move", "copy"):
|
||
_require("from" in op, f"missing 'from' parameter to {name!r}")
|
||
|
||
try:
|
||
tokens = decode_pointer(op["path"])
|
||
if name in ("move", "copy"):
|
||
from_tokens = decode_pointer(op["from"])
|
||
else:
|
||
from_tokens = ()
|
||
except JsonPointerError as exc:
|
||
raise JsonPatchError(str(exc)) from exc
|
||
|
||
if name == "add":
|
||
return _op_add(work, tokens, op["value"])
|
||
if name == "remove":
|
||
return _op_remove(work, tokens)
|
||
if name == "replace":
|
||
return _op_replace(work, tokens, op["value"])
|
||
if name == "move":
|
||
return _op_move(work, tokens, from_tokens)
|
||
if name == "copy":
|
||
return _op_copy(work, tokens, from_tokens)
|
||
return _op_test(work, tokens, op["value"])
|
||
|
||
|
||
def apply_patch(doc, patch):
|
||
"""Apply an RFC 6902 patch and return the patched document.
|
||
|
||
`doc` is never mutated: the operations run on a deep copy which is returned
|
||
only when every op succeeds, so a failing patch is atomic (nothing applied).
|
||
"""
|
||
if not isinstance(patch, list):
|
||
raise JsonPatchError("a JSON Patch must be an array of operations")
|
||
work = copy.deepcopy(doc)
|
||
for i, op in enumerate(patch):
|
||
try:
|
||
work = _apply(work, op)
|
||
except (JsonPatchError, JsonPointerError) as exc:
|
||
raise JsonPatchError(
|
||
f"patch[{i}] ({op.get('op') if isinstance(op, dict) else op!r}): {exc}"
|
||
) from exc
|
||
return work
|
||
|
||
|
||
def apply_patch_text(doc_str: str, patch_str: str) -> str:
|
||
"""Parse JSON text, apply the patch, return the result as JSON text."""
|
||
result = apply_patch(json.loads(doc_str), json.loads(patch_str))
|
||
return json.dumps(result, ensure_ascii=False)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Self-test
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _selftest_cases():
|
||
"""(name, doc, patch, expected_or_None, must_raise)."""
|
||
return [
|
||
# ---- RFC 6902 Appendix A, verbatim ----
|
||
("A.1 add object member", {"foo": "bar"},
|
||
[{"op": "add", "path": "/baz", "value": "qux"}],
|
||
{"foo": "bar", "baz": "qux"}, False),
|
||
("A.2 add array element", {"foo": ["bar", "baz"]},
|
||
[{"op": "add", "path": "/foo/1", "value": "qux"}],
|
||
{"foo": ["bar", "qux", "baz"]}, False),
|
||
("A.3 remove object member", {"baz": "qux", "foo": "bar"},
|
||
[{"op": "remove", "path": "/baz"}], {"foo": "bar"}, False),
|
||
("A.4 remove array element", {"foo": ["bar", "qux", "baz"]},
|
||
[{"op": "remove", "path": "/foo/1"}], {"foo": ["bar", "baz"]}, False),
|
||
("A.5 replace value", {"baz": "qux", "foo": "bar"},
|
||
[{"op": "replace", "path": "/baz", "value": "boo"}],
|
||
{"baz": "boo", "foo": "bar"}, False),
|
||
("A.6 replace missing -> error", {"foo": "bar"},
|
||
[{"op": "replace", "path": "/baz", "value": "qux"}], None, True),
|
||
("A.7 move value",
|
||
{"foo": {"bar": "baz", "waldo": "fred"}, "qux": {"corge": "grault"}},
|
||
[{"op": "move", "from": "/foo/waldo", "path": "/qux/thud"}],
|
||
{"foo": {"bar": "baz"}, "qux": {"corge": "grault", "thud": "fred"}}, False),
|
||
("A.8 move array element", {"foo": ["all", "grass", "cows", "eat"]},
|
||
[{"op": "move", "from": "/foo/1", "path": "/foo/3"}],
|
||
{"foo": ["all", "cows", "eat", "grass"]}, False),
|
||
("A.9 test success", {"baz": "qux", "foo": ["a", 2, "c"]},
|
||
[{"op": "test", "path": "/baz", "value": "qux"}],
|
||
{"baz": "qux", "foo": ["a", 2, "c"]}, False),
|
||
("A.10 test failure -> error", {"baz": "qux"},
|
||
[{"op": "test", "path": "/baz", "value": "bar"}], None, True),
|
||
("A.11 add to root", {"foo": "bar"},
|
||
[{"op": "add", "path": "", "value": {"baz": "qux"}}],
|
||
{"baz": "qux"}, False),
|
||
("A.12 escaped ~1", {"": "/", "/": " slash "},
|
||
[{"op": "test", "path": "/~1", "value": " slash "}],
|
||
{"": "/", "/": " slash "}, False),
|
||
("A.13 escaped ~0", {"~": "tilde", "/": "slash"},
|
||
[{"op": "test", "path": "/~0", "value": "tilde"}],
|
||
{"~": "tilde", "/": "slash"}, False),
|
||
("A.14 add out of range -> error", {"foo": ["bar", "qux", "baz"]},
|
||
[{"op": "add", "path": "/foo/5", "value": "qux"}], None, True),
|
||
("A.15 append with '-'", {"foo": ["bar", "baz"]},
|
||
[{"op": "add", "path": "/foo/-", "value": "qux"}],
|
||
{"foo": ["bar", "baz", "qux"]}, False),
|
||
("A.16 add into existing nested", {"foo": {"bar": "baz"}},
|
||
[{"op": "add", "path": "/foo/bar", "value": "qux"}],
|
||
{"foo": {"bar": "qux"}}, False),
|
||
("A.17 test array element", {"foo": ["bar", "baz"]},
|
||
[{"op": "test", "path": "/foo/1", "value": "baz"}],
|
||
{"foo": ["bar", "baz"]}, False),
|
||
("A.18 test object", {"foo": ["bar", "baz"]},
|
||
[{"op": "test", "path": "/foo", "value": ["bar", "baz"]}],
|
||
{"foo": ["bar", "baz"]}, False),
|
||
("A.19 test nested value", {"foo": {"bar": {"baz": "qux"}}},
|
||
[{"op": "test", "path": "/foo/bar/baz", "value": "qux"}],
|
||
{"foo": {"bar": {"baz": "qux"}}}, False),
|
||
("A.20 add into nested array", {"foo": [{"bar": "baz"}]},
|
||
[{"op": "add", "path": "/foo/0/bar", "value": "qux"}],
|
||
{"foo": [{"bar": "qux"}]}, False),
|
||
("A.21 test failure -> error", {"foo": ["bar", "baz"]},
|
||
[{"op": "test", "path": "/foo/1", "value": "qux"}], None, True),
|
||
|
||
# ---- spec_tests.json extras ----
|
||
("spec 4.1 add with missing object -> error", {"q": {"bar": 2}},
|
||
[{"op": "add", "path": "/a/b", "value": 1}], None, True),
|
||
("spec A.12 add to non-existent target -> error", {"foo": "bar"},
|
||
[{"op": "add", "path": "/baz/bat", "value": "qux"}], None, True),
|
||
("spec A.14 ~ escape ordering", {"/": 9, "~1": 10},
|
||
[{"op": "test", "path": "/~01", "value": 10}],
|
||
{"/": 9, "~1": 10}, False),
|
||
("spec A.15 number vs string -> error", {"/": 9, "~1": 10},
|
||
[{"op": "test", "path": "/~01", "value": "10"}], None, True),
|
||
("spec A.16 append array value", {"foo": ["bar"]},
|
||
[{"op": "add", "path": "/foo/-", "value": ["abc", "def"]}],
|
||
{"foo": ["bar", ["abc", "def"]]}, False),
|
||
|
||
# ---- RFC 6901 pointer quirks ----
|
||
("6901 whole document pointer is ''", {"a": 1},
|
||
[{"op": "test", "path": "", "value": {"a": 1}}], {"a": 1}, False),
|
||
("6901 leading-zero index rejected", {"foo": ["a", "b"]},
|
||
[{"op": "remove", "path": "/foo/01"}], None, True),
|
||
("6901 '-' is not a read index", {"foo": ["a"]},
|
||
[{"op": "remove", "path": "/foo/-"}], None, True),
|
||
("6901 '~2' escape rejected", {"foo": 1},
|
||
[{"op": "test", "path": "/fo~2o", "value": 1}], None, True),
|
||
("6901 dangling '~' rejected", {"foo": 1},
|
||
[{"op": "test", "path": "/foo~", "value": 1}], None, True),
|
||
("6901 pointer must start with '/'", {"foo": 1},
|
||
[{"op": "test", "path": "foo", "value": 1}], None, True),
|
||
("6901 path must be a string", {"foo": 1},
|
||
[{"op": "test", "path": None, "value": 1}], None, True),
|
||
|
||
# ---- operation semantics ----
|
||
("move /a -> /a is a no-op", {"a": 1, "b": 2},
|
||
[{"op": "move", "from": "/a", "path": "/a"}], {"a": 1, "b": 2}, False),
|
||
("move into own descendant -> error", {"a": {"b": 1}},
|
||
[{"op": "move", "from": "/a", "path": "/a/b"}], None, True),
|
||
("copy array element into object", {"foo": ["a", "b"]},
|
||
[{"op": "copy", "from": "/foo/0", "path": "/bar"}],
|
||
{"foo": ["a", "b"], "bar": "a"}, False),
|
||
("copy into array middle", {"foo": [1, 2, 3]},
|
||
[{"op": "copy", "from": "/foo/0", "path": "/foo/1"}],
|
||
{"foo": [1, 1, 2, 3]}, False),
|
||
("add at index == len is legal", {"foo": [1, 2]},
|
||
[{"op": "add", "path": "/foo/2", "value": 3}], {"foo": [1, 2, 3]}, False),
|
||
("replace whole document", {"a": 1},
|
||
[{"op": "replace", "path": "", "value": [1, 2]}], [1, 2], False),
|
||
("remove whole document -> error", {"a": 1},
|
||
[{"op": "remove", "path": ""}], None, True),
|
||
("empty patch is a no-op", {"a": 1}, [], {"a": 1}, False),
|
||
("chained ops on the live doc", {"a": 1},
|
||
[{"op": "add", "path": "/b", "value": 2},
|
||
{"op": "remove", "path": "/a"},
|
||
{"op": "replace", "path": "/b", "value": 3}], {"b": 3}, False),
|
||
("unknown op rejected", {"a": 1},
|
||
[{"op": "spam", "path": "/a", "value": 1}], None, True),
|
||
("missing value rejected", {"a": 1},
|
||
[{"op": "add", "path": "/b"}], None, True),
|
||
("missing path rejected", {"a": 1},
|
||
[{"op": "remove"}], None, True),
|
||
("patch must be an array", {"a": 1},
|
||
{"op": "add", "path": "/b", "value": 2}, None, True),
|
||
("non-object patch entry rejected", {"a": 1}, ["add"], None, True),
|
||
|
||
# ---- test-op equality ----
|
||
("test: key order irrelevant", {"b": 1, "a": 2},
|
||
[{"op": "test", "path": "", "value": {"a": 2, "b": 1}}],
|
||
{"b": 1, "a": 2}, False),
|
||
("test: 1 == 1.0", {"a": 1},
|
||
[{"op": "test", "path": "/a", "value": 1.0}], {"a": 1}, False),
|
||
("test: true != 1", {"a": True},
|
||
[{"op": "test", "path": "/a", "value": 1}], None, True),
|
||
("test: false != 0", {"a": False},
|
||
[{"op": "test", "path": "/a", "value": 0}], None, True),
|
||
("test: null == null", {"a": None},
|
||
[{"op": "test", "path": "/a", "value": None}], {"a": None}, False),
|
||
("test: array order significant", {"a": [1, 2]},
|
||
[{"op": "test", "path": "/a", "value": [2, 1]}], None, True),
|
||
("test: deep-copied value is not shared", {"a": {"x": 1}},
|
||
[{"op": "copy", "from": "/a", "path": "/b"}],
|
||
{"a": {"x": 1}, "b": {"x": 1}}, False),
|
||
]
|
||
|
||
|
||
def run_selftest(verbose: bool = False) -> int:
|
||
"""RFC 6902 Appendix A + RFC 6901 + equality semantics. 0 = all pass."""
|
||
passed = failed = 0
|
||
for name, doc, patch, expected, must_raise in _selftest_cases():
|
||
try:
|
||
got = apply_patch(doc, patch)
|
||
except JsonPatchError as exc:
|
||
if must_raise:
|
||
passed += 1
|
||
if verbose:
|
||
print(f" PASS {name} (raised: {exc})")
|
||
else:
|
||
failed += 1
|
||
print(f" FAIL {name}: unexpected JsonPatchError: {exc}")
|
||
continue
|
||
except Exception as exc: # noqa: BLE001 - surface, never mask
|
||
failed += 1
|
||
print(f" FAIL {name}: unexpected {type(exc).__name__}: {exc}")
|
||
continue
|
||
if must_raise:
|
||
failed += 1
|
||
print(f" FAIL {name}: expected JsonPatchError, got {got!r}")
|
||
elif json_equal(got, expected):
|
||
passed += 1
|
||
if verbose:
|
||
print(f" PASS {name}")
|
||
else:
|
||
failed += 1
|
||
print(f" FAIL {name}: expected {_canonical(expected)!r}, "
|
||
f"got {_canonical(got)!r}")
|
||
print(f"\n== {passed} passed, {failed} failed ==")
|
||
return 0 if failed == 0 else 1
|
||
|
||
|
||
def _selftest_json() -> int:
|
||
doc_s = '{"foo": ["bar", "baz"], "n": 1}'
|
||
patch_s = ('[{"op":"add","path":"/foo/1","value":"qux"},'
|
||
'{"op":"test","path":"/n","value":1}]')
|
||
out_s = apply_patch_text(doc_s, patch_s)
|
||
want = '{"foo": ["bar", "qux", "baz"], "n": 1}'
|
||
if out_s == want:
|
||
print(f"json-selftest PASS -> {out_s}")
|
||
return 0
|
||
print(f"json-selftest FAIL: {out_s!r} != {want!r}")
|
||
return 1
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
argv = list(sys.argv[1:] if argv is None else argv)
|
||
if "--selftest" in argv:
|
||
return run_selftest(verbose=("-v" in argv or "--verbose" in argv))
|
||
if "--selftest-json" in argv:
|
||
return _selftest_json()
|
||
|
||
doc_s = patch_s = None
|
||
i = 0
|
||
while i < len(argv):
|
||
if argv[i] == "--doc" and i + 1 < len(argv):
|
||
doc_s = argv[i + 1]
|
||
i += 2
|
||
elif argv[i] == "--patch" and i + 1 < len(argv):
|
||
patch_s = argv[i + 1]
|
||
i += 2
|
||
else:
|
||
print(f"usage: {sys.argv[0]} [--selftest [--verbose]] "
|
||
"[--selftest-json] | --doc FILE --patch FILE", file=sys.stderr)
|
||
return USAGE_ERR
|
||
if doc_s is None or patch_s is None:
|
||
print("usage: --doc FILE --patch FILE (or --selftest)", file=sys.stderr)
|
||
return USAGE_ERR
|
||
|
||
def _read(spec: str) -> str:
|
||
return sys.stdin.read() if spec == "-" else open(spec, encoding="utf-8").read()
|
||
|
||
try:
|
||
print(apply_patch_text(_read(doc_s), _read(patch_s)))
|
||
except (JsonPatchError, json.JSONDecodeError, OSError) as exc:
|
||
print(f"error: {exc}", file=sys.stderr)
|
||
return PATCH_ERR
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|