null: regex_nfa.py — Thompson-NFA regex engine, 4858 checks green vs re (4800 differential-fuzz cases + 6/6 injected mutants caught), no-backtracking bench
This commit is contained in:
null committed
1 parent
fbcadf6ea7
commit
ca8e2b77ab
23 files changed
+5659
No files matched your search
@@ -0,0 +1,611 @@
|
||||
#!/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())
|
||||
Reference in new issue
Block a user