Files
troupe-standup/wizard-work/null/jsonpatch/test_jsonptr_patch.py
T

567 lines
22 KiB
Python

#!/usr/bin/env python3
"""test_jsonptr_patch.py -- verification battery for jsonptr_patch.py.
Stages (all must pass; the process exit code is the verdict):
1. vendored-suite integrity -- sha256 of the upstream suite files matches the
pins recorded when they were fetched, so the oracle cannot be edited
silently.
2. official suite -- json-patch/json-patch-tests (tests.json +
spec_tests.json) run against the engine. `disabled` cases are skipped.
Expected results are compared with stdlib json.dumps(sort_keys=True), not
with the engine's own equality function.
3. engine selftest -- RFC 6902 Appendix A + RFC 6901 pointer quirks +
test-op equality semantics.
4. atomicity -- a patch whose later op fails must leave the
caller's document byte-identical (canonical JSON comparison).
5. fuzz -- seeded random documents and patches; checks
(a) only JsonPatchError/JsonPointerError ever escape, (b) the input is
untouched on failure, (c) results are deterministic, (d) constructed
add/copy/move round-trips are identities, (e) constructed bad patches
raise, (f) `test` accepts the live value and rejects a mutated one.
6. teeth -- N source mutants of the engine are generated
and the whole battery is re-run against each; every mutant must be
killed (nonzero exit) or the stage fails.
Usage:
python3 test_jsonptr_patch.py # full battery
python3 test_jsonptr_patch.py --no-teeth # skip stage 6
python3 test_jsonptr_patch.py --engine FILE # run against a mutant copy
python3 test_jsonptr_patch.py --quiet # suppress per-case output
"""
from __future__ import annotations
import copy
import hashlib
import importlib.util
import json
import os
import random
import subprocess
import sys
import tempfile
HERE = os.path.dirname(os.path.abspath(__file__))
ENGINE = os.path.join(HERE, "jsonptr_patch.py")
# Vendored upstream suite (fetched from raw.githubusercontent.com,
# json-patch/json-patch-tests @ master, 2026-10-21). Pinned so a silent edit of
# the oracle is a test failure, not a pass.
VENDORED = {
"_vendor_json-patch-tests_tests.json": (
"de3dce3d0d5029fed83007e50b54607750dd3d1478d3c59ca35fdc18fb1a04ae", 95),
"_vendor_json-patch-tests_spec_tests.json": (
"a26b050292207033e5cccc5d6102b7bd6f8add7db0d0680e5d46a7ecf40a8c7b", 17),
}
MUTANTS = [
("add index==len allowed", 'idx <= len(container)', 'idx < len(container)'),
("missing parent tolerated", '_require(tok in container,',
'_require(True or tok in container,'),
("bool is a number", 'if isinstance(a, bool) or isinstance(b, bool):',
'if isinstance(a, (int, float)) and isinstance(b, (int, float)):'),
("unescape ~0 wrong", 'out.append("~")', 'out.append("0")'),
("move into descendant allowed",
'not (len(tokens) > len(from_tokens)', '(len(tokens) > len(from_tokens)'),
("pointer slash not enforced", 'if not ref.startswith("/"):', 'if False:'),
("leading zeros allowed", 'return tok == "0" or tok[0] != "0"', 'return True'),
("remove root tolerated", '_require(bool(tokens), "operation target must not be the whole document")',
'pass'),
# Regression guard for the Unicode-digit array-index bug (2026-10-21): a
# bare str.isdigit() accepts "²", and int("²") raises a plain ValueError
# that leaks past the JsonPointerError contract. This mutant re-introduces
# exactly that bug, so the suite fails if the isascii() guard is ever
# dropped again. `conformance.py` detects it via the error-contract stage.
("unicode digit index allowed", 'if not tok or not tok.isascii() or not tok.isdigit():',
'if not tok or not tok.isdigit():'),
]
def load_engine(path: str = ENGINE):
spec = importlib.util.spec_from_file_location("jsonptr_patch_under_test", path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def canon(value) -> str:
"""Independent comparison key: stdlib canonical JSON."""
return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
class Report:
def __init__(self, quiet: bool):
self.quiet = quiet
self.checks = 0
self.failures = []
def check(self, ok: bool, label: str, detail: str = "") -> bool:
self.checks += 1
if not ok:
self.failures.append(f"{label}{(' :: ' + detail) if detail else ''}")
if not self.quiet:
print(f" FAIL {label}{(' :: ' + detail) if detail else ''}")
elif not self.quiet:
print(f" pass {label}")
return ok
def summary(self, stage: str) -> None:
print(f"[{stage}] {self.checks} checks, {len(self.failures)} failed")
# ---------------------------------------------------------------------------
# Stage 1 + 2: vendored official suite
# ---------------------------------------------------------------------------
def stage_vendored(rep: Report) -> int:
for fname, (digest, n_cases) in VENDORED.items():
path = os.path.join(HERE, fname)
raw = open(path, "rb").read()
got = hashlib.sha256(raw).hexdigest()
if not rep.check(got == digest, f"sha256 {fname}", f"{got} != {digest}"):
return 1
cases = json.loads(raw.decode("utf-8"))
if not rep.check(len(cases) == n_cases, f"case count {fname}",
f"{len(cases)} != {n_cases}"):
return 1
return 0
def stage_official(eng, rep: Report) -> int:
passed = failed = skipped = 0
for fname in VENDORED:
for case in json.loads(open(os.path.join(HERE, fname), encoding="utf-8").read()):
label = case.get("comment") or f"{fname}#{passed + failed + skipped}"
if case.get("disabled"):
skipped += 1
continue
doc = case["doc"]
snapshot = canon(doc)
try:
got = eng.apply_patch(doc, case["patch"])
except (eng.JsonPatchError, eng.JsonPointerError) as exc:
if "error" in case:
passed += 1
if canon(doc) != snapshot:
failed += 1
rep.check(False, f"{label} atomicity", canon(doc))
continue
failed += 1
rep.check(False, f"{label} raised", str(exc)[:120])
continue
except Exception as exc: # noqa: BLE001
failed += 1
rep.check(False, f"{label} wrong exception type",
f"{type(exc).__name__}: {exc}"[:120])
continue
if "error" in case:
failed += 1
rep.check(False, f"{label} expected error", canon(got)[:120])
continue
if canon(got) == canon(case["expected"]):
passed += 1
else:
failed += 1
rep.check(False, f"{label} result",
f"got {canon(got)[:120]} want {canon(case['expected'])[:120]}")
print(f" official suite: {passed} passed, {failed} failed, {skipped} skipped (disabled)")
rep.checks += passed
rep.check(failed == 0, "official suite", f"{failed} failed cases")
return 0 if failed == 0 else 1
# ---------------------------------------------------------------------------
# Stage 4: atomicity
# ---------------------------------------------------------------------------
def stage_atomicity(eng, rep: Report) -> int:
doc = {"a": {"b": [1, 2, 3]}, "c": "keep"}
snapshot = canon(doc)
# op 2 succeeds, op 3 fails -> nothing may have been applied
patch = [{"op": "add", "path": "/d", "value": 9},
{"op": "remove", "path": "/a/b/9"}]
try:
eng.apply_patch(doc, patch)
rep.check(False, "atomicity: expected failure", "patch returned a document")
except eng.JsonPatchError:
rep.check(canon(doc) == snapshot, "atomicity: doc unchanged after failure",
canon(doc))
# test-op failure after a successful mutation
doc2 = {"x": 1}
patch2 = [{"op": "add", "path": "/y", "value": 2},
{"op": "test", "path": "/x", "value": 99}]
try:
eng.apply_patch(doc2, patch2)
rep.check(False, "atomicity: expected test failure")
except eng.JsonPatchError:
rep.check(canon(doc2) == canon({"x": 1}),
"atomicity: doc unchanged after test failure", canon(doc2))
# success path must not mutate the caller's document
doc3 = {"k": [1]}
out = eng.apply_patch(doc3, [{"op": "add", "path": "/k/-", "value": 2}])
rep.check(canon(doc3) == canon({"k": [1]}), "purity: input untouched on success",
canon(doc3))
rep.check(canon(out) == canon({"k": [1, 2]}), "purity: returned doc patched",
canon(out))
return 0
# ---------------------------------------------------------------------------
# Stage 5: fuzz
# ---------------------------------------------------------------------------
LEAF_POOL = [None, True, False, 0, 1, -7, 2.5, "", "x", "héllo", "0", "~", "/"]
def rand_value(rng, depth=0):
r = rng.random()
if depth < 3 and r < 0.25:
return [rand_value(rng, depth + 1) for _ in range(rng.randint(0, 3))]
if depth < 3 and r < 0.5:
return {rng.choice(["a", "b", "k", "n", "0", "~1"]): rand_value(rng, depth + 1)
for _ in range(rng.randint(0, 3))}
return rng.choice(LEAF_POOL)
def rand_doc(rng):
if rng.random() < 0.25:
return rand_value(rng)
return {rng.choice(["a", "b", "arr", "obj", "n"]): rand_value(rng)
for _ in range(rng.randint(1, 4))}
def all_paths(node, prefix=()):
yield prefix
if isinstance(node, dict):
for k, v in node.items():
yield from all_paths(v, prefix + (str(k),))
elif isinstance(node, list):
for i, v in enumerate(node):
yield from all_paths(v, prefix + (str(i),))
def ptr(tokens) -> str:
return "".join("/" + str(t).replace("~", "~0").replace("/", "~1") for t in tokens)
def node_at(node, tokens):
for t in tokens:
node = node[int(t)] if isinstance(node, list) else node[t]
return node
def stage_fuzz(eng, rep: Report) -> int:
rng = random.Random(20261021)
stats = {"patches": 0, "ok": 0, "raised": 0,
"wrong_exc": 0, "mutated_input": 0, "nondeterministic": 0,
"identity_fail": 0, "expected_fail_ok": 0, "expected_ok_fail": 0}
for _ in range(2000):
doc = rand_doc(rng)
paths = [p for p in all_paths(doc) if p]
ops = []
for _ in range(rng.randint(1, 4)):
op = rng.choice(["add", "add", "remove", "replace", "move", "copy", "test"])
p = rng.choice(paths) if paths else ()
if op == "add":
if not p or rng.random() < 0.5:
ops.append({"op": "add", "path": "", "value": rand_value(rng)})
else:
parent, tok = p[:-1], p[-1]
try:
container = node_at(doc, parent) if parent else doc
except (KeyError, IndexError, TypeError):
ops.append({"op": "add", "path": ptr(p), "value": rand_value(rng)})
continue
if isinstance(container, dict):
newtok = tok if rng.random() < 0.5 else "new" + str(rng.randint(0, 9))
ops.append({"op": "add", "path": ptr(parent + (newtok,)),
"value": rand_value(rng)})
elif isinstance(container, list):
idx = "-" if rng.random() < 0.3 else str(rng.randint(0, len(container)))
ops.append({"op": "add", "path": ptr(parent + (idx,)),
"value": rand_value(rng)})
else:
ops.append({"op": "add", "path": ptr(p), "value": rand_value(rng)})
elif op in ("remove", "replace", "test"):
ops.append({"op": op, "path": ptr(p)})
if op != "remove":
ops[-1]["value"] = rand_value(rng)
else:
frm = rng.choice(paths) if paths else ()
ops.append({"op": op, "from": ptr(frm), "path": ptr(p)})
stats["patches"] += 1
snapshot = canon(doc)
try:
out1 = eng.apply_patch(doc, ops)
out2 = eng.apply_patch(copy.deepcopy(doc), ops)
except (eng.JsonPatchError, eng.JsonPointerError):
stats["raised"] += 1
if canon(doc) != snapshot:
stats["mutated_input"] += 1
continue
except Exception as exc: # noqa: BLE001
stats["wrong_exc"] += 1
if not rep.quiet:
print(f" FAIL fuzz wrong exception {type(exc).__name__}: {exc} "
f"ops={ops}")
continue
stats["ok"] += 1
if canon(out1) != canon(out2):
stats["nondeterministic"] += 1
if canon(doc) != snapshot:
stats["mutated_input"] += 1
rep.check(stats["wrong_exc"] == 0, "fuzz: only JsonPatchError escapes",
str(stats["wrong_exc"]))
rep.check(stats["mutated_input"] == 0, "fuzz: input never mutated",
str(stats["mutated_input"]))
rep.check(stats["nondeterministic"] == 0, "fuzz: deterministic",
str(stats["nondeterministic"]))
print(f" fuzz: {stats['patches']} patches, {stats['ok']} ok, "
f"{stats['raised']} rejected")
# ---- constructed identities ----
# copy X -> fresh slot, remove the slot => original document
# add fresh slot, remove the slot => original document
def identity_ops(doc, p, rng):
root = doc
if isinstance(root, dict):
slot = "/__copy__" if rng.random() < 0.5 else "/__add__"
elif isinstance(root, list):
slot = "/-" # append
else:
return None
try:
out = eng.apply_patch(doc, [{"op": "copy", "from": ptr(p), "path": slot}])
except eng.JsonPatchError:
return None
if isinstance(root, list):
idx = len(out) - 1 # the appended element
else:
idx = slot
try:
out = eng.apply_patch(out, [{"op": "remove", "path": idx}])
except eng.JsonPatchError:
return None
return out
ident = skipped = 0
for _ in range(500):
doc = rand_doc(rng)
paths = [p for p in all_paths(doc) if p]
if not paths:
skipped += 1
continue
p = rng.choice(paths)
out = identity_ops(doc, p, rng)
if out is None:
skipped += 1
continue
ident += 1
if canon(out) != canon(doc):
stats["identity_fail"] += 1
rep.check(stats["identity_fail"] == 0, "fuzz: copy/add-then-remove is identity",
f"{stats['identity_fail']} failures over {ident} trials "
f"({skipped} skipped)")
print(f" fuzz: identity {ident} ok / {stats['identity_fail']} failed / "
f"{skipped} skipped")
# move out and back
def move_roundtrip(doc, p):
root = doc
if isinstance(root, dict):
tmp, back = "/__tmp__", ptr(p)
elif isinstance(root, list):
tmp = "/-"
back = ptr(p)
else:
return None
try:
out = eng.apply_patch(doc, [{"op": "move", "from": ptr(p), "path": tmp}])
except eng.JsonPatchError:
return None
if isinstance(root, list):
tmp = f"/{len(out) - 1}"
try:
out = eng.apply_patch(out, [{"op": "move", "from": tmp, "path": back}])
except eng.JsonPatchError:
return None
return out
mv_ok = mv_fail = mv_skip = 0
for _ in range(500):
doc = rand_doc(rng)
paths = [p for p in all_paths(doc) if p]
if not paths:
mv_skip += 1
continue
out = move_roundtrip(doc, rng.choice(paths))
if out is None:
mv_skip += 1
continue
mv_ok += 1
if canon(out) != canon(doc):
mv_fail += 1
rep.check(mv_fail == 0, "fuzz: move out and back is identity",
f"{mv_fail} failures over {mv_ok + mv_fail} trials "
f"({mv_skip} skipped)")
# ---- constructed failures must raise ----
bad = 0
for _ in range(500):
doc = rand_doc(rng)
paths = [p for p in all_paths(doc) if p]
p = rng.choice(paths) if paths else ("a",)
cases = [
[{"op": "remove", "path": ptr(p + ("__nope__",))}],
[{"op": "replace", "path": ptr(p + ("__nope__",)), "value": 1}],
[{"op": "add", "path": ptr(p + ("__nope__", "x")), "value": 1}],
[{"op": "test", "path": ptr(p + ("__nope__",)), "value": 1}],
[{"op": "move", "from": ptr(p + ("__nope__",)), "path": "/x"}],
[{"op": "copy", "from": ptr(p + ("__nope__",)), "path": "/x"}],
]
for patch in cases:
try:
eng.apply_patch(doc, patch)
stats["expected_fail_ok"] += 1
except eng.JsonPatchError:
bad += 1
rep.check(stats["expected_fail_ok"] == 0,
"fuzz: bad paths are rejected", str(stats["expected_fail_ok"]))
print(f" fuzz: {bad} bad patches rejected as expected")
# ---- test op: live value accepted, mutated value rejected ----
t_ok = t_rej = t_wrong = 0
SENTINEL = "__SENTINEL_9f3a__"
for _ in range(1000):
doc = rand_doc(rng)
if SENTINEL in canon(doc):
continue
paths = [p for p in all_paths(doc) if p]
if not paths:
continue
p = rng.choice(paths)
live = node_at(doc, p)
if rng.random() < 0.5:
try:
eng.apply_patch(doc, [{"op": "test", "path": ptr(p), "value": live}])
t_ok += 1
except eng.JsonPatchError:
t_wrong += 1
else:
try:
eng.apply_patch(doc, [{"op": "test", "path": ptr(p), "value": SENTINEL}])
t_wrong += 1
except eng.JsonPatchError:
t_rej += 1
rep.check(t_wrong == 0, "fuzz: test accepts live value, rejects sentinel",
f"{t_wrong} wrong outcomes")
print(f" fuzz: test op {t_ok} accepted / {t_rej} rejected / {t_wrong} wrong")
return 0
# ---------------------------------------------------------------------------
# Stage 6: teeth (mutants must be killed)
# ---------------------------------------------------------------------------
def stage_teeth(rep: Report) -> int:
"""Every source mutant must be killed by the battery OR the conformance stage.
The conformance stage is run once per mutant because some regressions (the
Unicode-digit array-index bug, for instance) are invisible to the official
suite and to the engine's own selftest, and are only caught by the
independent error-contract sweep.
"""
src = open(ENGINE, encoding="utf-8").read()
killed = survived = 0
conformance = os.path.join(HERE, "conformance.py")
with tempfile.TemporaryDirectory(prefix="jsonptr_teeth_") as tmp:
for i, (name, old, new) in enumerate(MUTANTS):
if src.count(old) != 1:
rep.check(False, f"mutant anchor {name}",
f"anchor occurs {src.count(old)} times")
survived += 1
continue
mutant = src.replace(old, new)
mpath = os.path.join(tmp, f"mutant_{i}.py")
open(mpath, "w", encoding="utf-8").write(mutant)
proc = subprocess.run(
[sys.executable, os.path.abspath(__file__), "--engine", mpath,
"--no-teeth", "--quiet"],
capture_output=True, text=True, timeout=120)
conf = None
if proc.returncode == 0 and os.path.exists(conformance):
conf = subprocess.run(
[sys.executable, conformance, "--engine", mpath, "--quiet"],
capture_output=True, text=True, timeout=900)
if proc.returncode != 0 or (conf is not None and conf.returncode != 0):
killed += 1
if not rep.quiet:
print(f" killed {name}")
else:
survived += 1
rep.check(False, f"mutant survived: {name}",
proc.stdout.strip()[-200:])
if not rep.quiet:
print(f" SURVIVED {name}")
print(f" teeth: {killed} killed, {survived} survived")
rep.check(survived == 0, "teeth: every mutant killed", f"{survived} survived")
return 0 if survived == 0 else 1
# ---------------------------------------------------------------------------
def main(argv=None) -> int:
argv = list(sys.argv[1:] if argv is None else argv)
quiet = "--quiet" in argv
teeth = "--no-teeth" not in argv
engine_path = ENGINE
if "--engine" in argv:
engine_path = os.path.abspath(argv[argv.index("--engine") + 1])
rep = Report(quiet)
eng = load_engine(engine_path)
print(f"battery: engine={engine_path}")
rc = 0
rc |= stage_vendored(rep)
rc |= stage_official(eng, rep)
rep.summary("vendored+official")
ok_selftest = eng.run_selftest(verbose=False) == 0
rep.check(ok_selftest, "engine selftest (Appendix A + 6901 + equality)")
rep.summary("selftest")
stage_atomicity(eng, rep)
rep.summary("atomicity")
stage_fuzz(eng, rep)
rep.summary("fuzz")
# Independent conformance stage (broad pointer space + CLI contract).
if os.path.exists(os.path.join(HERE, "conformance.py")):
proc = subprocess.run(
[sys.executable, os.path.join(HERE, "conformance.py"),
"--engine", engine_path, "--quiet"],
capture_output=True, text=True, timeout=900)
last = proc.stdout.strip().splitlines()[-2:] if proc.stdout.strip() else []
rep.check(proc.returncode == 0, "conformance stage",
" | ".join(last) if last else proc.stderr.strip()[:200])
rep.summary("conformance")
if teeth:
stage_teeth(rep)
rep.summary("teeth")
print(f"\nTOTAL: {rep.checks} checks, {len(rep.failures)} failed")
for f in rep.failures[:20]:
print(f" FAILED: {f}")
print("VERDICT: PASS" if (not rep.failures and rc == 0) else "VERDICT: FAIL")
return 0 if (not rep.failures and rc == 0) else 1
if __name__ == "__main__":
sys.exit(main())