461 lines
18 KiB
Python
461 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""conformance.py -- independent RFC 6901/6902 conformance stage for jsonptr_patch.
|
|
|
|
WHY THIS EXISTS
|
|
---------------
|
|
The vendored upstream suite (json-patch/json-patch-tests) is the primary oracle,
|
|
but it pins only ~50 distinct JSON Pointer strings and never exercises the CLI.
|
|
Two things therefore go unguarded:
|
|
|
|
* the *error contract* -- the engine documents that only JsonPointerError and
|
|
JsonPatchError escape -- over a broad adversarial path space, and
|
|
* the *CLI contract* -- exit codes 0/1/2, one-line errors, never a traceback.
|
|
|
|
HOW IT VERIFIES (independent oracle, not the engine's own predicates)
|
|
--------------------------------------------------------------------
|
|
RFC 6901 section 4 says an array index token is evaluated by `int(tok)`, so
|
|
`0 <= int(tok) < len` and the leading-zero rule are checked *directly against
|
|
the stdlib*, not through `is_array_index`. Every `remove`/`replace`/`test`
|
|
outcome is checked against an independently-computed expectation. Every result
|
|
is canonicalised with `json.dumps(sort_keys=True)` so the engine's own
|
|
`json_equal` cannot grade itself.
|
|
|
|
Exit code is the verdict: 0 = every check passed.
|
|
|
|
Usage:
|
|
python3 conformance.py # full stage
|
|
python3 conformance.py --quiet # no per-check output
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
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")
|
|
if "--engine" in sys.argv:
|
|
ENGINE = os.path.abspath(sys.argv[sys.argv.index("--engine") + 1])
|
|
sys.path.insert(0, HERE)
|
|
import jsonptr_patch as E # noqa: E402
|
|
|
|
CONTRACT = (E.JsonPatchError, E.JsonPointerError)
|
|
|
|
|
|
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, label, detail=""):
|
|
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")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Independent RFC 6901 s.4 predicate -- deliberately NOT E.is_array_index.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def std_index(tok: str):
|
|
"""Return the array index `tok` denotes, or None if it is not a valid one.
|
|
|
|
Mirrors the RFC text (evaluate as int) while rejecting non-ASCII numerals,
|
|
which `str.isdigit()` wrongly accepts.
|
|
"""
|
|
if not isinstance(tok, str) or not tok or not tok.isascii():
|
|
return None
|
|
if not tok.isdigit():
|
|
return None
|
|
if tok != "0" and tok[0] == "0": # leading zeros are not canonical
|
|
return None
|
|
return int(tok)
|
|
|
|
|
|
def std_resolve(doc, ref: str):
|
|
"""Independent RFC 6901 evaluation.
|
|
|
|
Returns ("ok", value) or ("err", reason). Never calls into the engine.
|
|
"""
|
|
if not isinstance(ref, str):
|
|
return ("err", "pointer must be a str")
|
|
if ref == "":
|
|
return ("ok", doc)
|
|
if not ref.startswith("/"):
|
|
return ("err", "must be empty or start with '/'")
|
|
node = doc
|
|
for raw in ref[1:].split("/"):
|
|
# RFC 6901 s.4 unescape: ~1 -> /, ~0 -> ~; anything else is illegal.
|
|
out, i, n = [], 0, len(raw)
|
|
while i < n:
|
|
ch = raw[i]
|
|
if ch != "~":
|
|
out.append(ch)
|
|
i += 1
|
|
continue
|
|
if i + 1 >= n:
|
|
return ("err", f"dangling '~' in {raw!r}")
|
|
nxt = raw[i + 1]
|
|
if nxt == "0":
|
|
out.append("~")
|
|
elif nxt == "1":
|
|
out.append("/")
|
|
else:
|
|
return ("err", f"invalid escape '~{nxt}' in {raw!r}")
|
|
i += 2
|
|
tok = "".join(out)
|
|
if isinstance(node, dict):
|
|
if tok not in node:
|
|
return ("err", f"key {tok!r} not found")
|
|
node = node[tok]
|
|
elif isinstance(node, list):
|
|
idx = std_index(tok)
|
|
if idx is None:
|
|
return ("err", f"invalid array index {tok!r}")
|
|
if idx >= len(node):
|
|
return ("err", f"index {tok!r} out of range")
|
|
node = node[idx]
|
|
else:
|
|
return ("err", f"cannot index {type(node).__name__} with {tok!r}")
|
|
return ("ok", node)
|
|
|
|
|
|
def grammar_error(ref: str):
|
|
"""True if `ref` is grammatically invalid (RFC 6901 s.3) regardless of doc.
|
|
|
|
Only the grammar is checked here: a leading '/', and '~' followed by 0 or 1
|
|
or nothing. Everything else (missing keys, bad array indices) is
|
|
*semantic* and depends on the document, so it is not a grammar error.
|
|
"""
|
|
if not isinstance(ref, str):
|
|
return True
|
|
if ref == "":
|
|
return False
|
|
if not ref.startswith("/"):
|
|
return True
|
|
for raw in ref[1:].split("/"):
|
|
i, n = 0, len(raw)
|
|
while i < n:
|
|
if raw[i] == "~":
|
|
if i + 1 >= n:
|
|
return True
|
|
if raw[i + 1] not in "01":
|
|
return True
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adversarial path space
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Tokens that stress the RFC 6901 grammar and the array-index rule.
|
|
TOKENS = [
|
|
"0", "1", "2", "3", "9", "10", "00", "01", "007", "-1", "-", "+1", "1e0",
|
|
"0x1", " 1", "1 ", "1_0", "0b1", "true", "false", "null", "a", "A", "FOO",
|
|
"bar", "baz", "foo", "c%d", "e^f", "a~b", "a/b", "~", "~0", "~1", "~2",
|
|
"~01", "~10", "~00", "~11", "~0~1", "", " ", "\t", "\n", "\u0000", "%20",
|
|
"\u200b", "\u00b2", "\u00b3", "\u00b9", "\u2070", "\u0661", "\u0662",
|
|
"\u0660", "\u06f1", "\uff11", "\U0001d7d9", "\U0001d7da", "\u00bd", "\u2153",
|
|
"\u0663", "\u06f2",
|
|
]
|
|
|
|
# Documents that give the pointer grammar something to bite on.
|
|
DOCS = [
|
|
{},
|
|
[],
|
|
[1, 2, 3],
|
|
{"a": 1},
|
|
{"a": {"b": 2}},
|
|
{"a": [10, 20, 30]},
|
|
{"foo": ["bar", "qux", "baz"], "obj": {"k": "~x/y", "n": 1}},
|
|
{"\u00b2": "superscript-key", "a/b": "slash-key", "a~b": "tilde-key"},
|
|
{"a": [{"b": [1, 2]}, {"c": 3}]},
|
|
{"": "empty-key", "-": "dash-key", "0": "zero-key"},
|
|
]
|
|
|
|
|
|
def build_refs():
|
|
"""A deterministic, broad set of pointer strings."""
|
|
refs = ["", "/"]
|
|
for t in TOKENS:
|
|
refs.append("/" + t)
|
|
# two- and three-token paths over a reduced token set, to bound the count.
|
|
small = ["0", "1", "2", "a", "b", "foo", "obj", "~0", "~1", "~2", "-",
|
|
"00", "01", "\u00b2", "", " ", "\u0661"]
|
|
for t1 in small:
|
|
for t2 in small:
|
|
refs.append("/" + t1 + "/" + t2)
|
|
for t1 in ["a", "foo", "obj", "0", "~2", "\u00b2"]:
|
|
for t2 in ["0", "1", "b", "n", "1e0"]:
|
|
for t3 in ["0", "1", "x"]:
|
|
refs.append("/" + t1 + "/" + t2 + "/" + t3)
|
|
return refs
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage A: pointer grammar -- decode/escape/round-trip + std_resolve oracle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def stage_pointer_grammar(rep: Report) -> None:
|
|
for ref in build_refs():
|
|
# decode_pointer must accept exactly the grammatically valid pointers and
|
|
# reject exactly the grammatically invalid ones.
|
|
try:
|
|
toks = E.decode_pointer(ref)
|
|
dec_err = None
|
|
except E.JsonPointerError as exc:
|
|
toks, dec_err = None, str(exc)
|
|
|
|
if grammar_error(ref):
|
|
rep.check(dec_err is not None, f"decode_pointer rejects {ref!r}",
|
|
f"accepted, but the grammar is invalid")
|
|
continue
|
|
rep.check(dec_err is None, f"decode_pointer accepts {ref!r}",
|
|
f"rejected with {dec_err}")
|
|
if dec_err is not None:
|
|
continue
|
|
# round-trip: pointer_to_string(decode_pointer(x)) must be a pointer whose
|
|
# independent evaluation agrees, and escape/unescape must be inverses.
|
|
rt = E.pointer_to_string(toks)
|
|
rep.check(rt.startswith("/") or rt == "",
|
|
f"pointer_to_string({ref!r}) is a pointer", repr(rt))
|
|
joined = "/".join(E.escape_token(t) for t in toks)
|
|
rep.check(E.pointer_to_string(toks) == ("" if not toks else "/" + joined),
|
|
f"escape round-trip {ref!r}",
|
|
f"{E.pointer_to_string(toks)!r} vs {'/' + joined!r}")
|
|
# escape_token must escape '~' before '/', so '~/' -> '~0~1'.
|
|
rep.check(E.escape_token("~/") == "~0~1", "escape_token order '~/ -> ~0~1'",
|
|
repr(E.escape_token("~/")))
|
|
rep.check(E.unescape_token("~0~1") == "~/", "unescape '~0~1' -> '~/'",
|
|
repr(E.unescape_token("~0~1")))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage B: resolve() against the independent oracle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def stage_resolve(rep: Report) -> None:
|
|
refs = build_refs()
|
|
for doc in DOCS:
|
|
for ref in refs:
|
|
want_kind, want_val = std_resolve(doc, ref)
|
|
try:
|
|
got = E.resolve(doc, E.decode_pointer(ref))
|
|
got_kind, got_val = "ok", got
|
|
except CONTRACT:
|
|
got_kind, got_val = "err", None
|
|
except Exception as exc: # noqa: BLE001
|
|
rep.check(False, f"resolve {ref!r} leaks", type(exc).__name__)
|
|
continue
|
|
if want_kind == "ok":
|
|
rep.check(got_kind == "ok" and canon(got_val) == canon(want_val),
|
|
f"resolve {ref!r}",
|
|
f"want {want_val!r}, got {got_val!r}")
|
|
else:
|
|
rep.check(got_kind == "err", f"resolve {ref!r} must fail",
|
|
f"want error ({want_val}), got {got_val!r}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage C: the error contract over a broad (doc, patch) space
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def stage_error_contract(rep: Report) -> None:
|
|
ops = ["add", "remove", "replace", "move", "copy", "test"]
|
|
refs = build_refs()
|
|
trials = 0
|
|
for doc in DOCS:
|
|
for ref in refs:
|
|
for op in ops:
|
|
patch = [{"op": op, "path": ref}]
|
|
if op in ("add", "replace", "test"):
|
|
patch[0]["value"] = 1
|
|
if op in ("move", "copy"):
|
|
patch[0]["from"] = ref
|
|
trials += 1
|
|
try:
|
|
E.apply_patch(doc, patch)
|
|
except CONTRACT:
|
|
pass
|
|
except Exception as exc: # noqa: BLE001 - exactly what we hunt
|
|
rep.check(False, f"contract: {op} {ref!r}",
|
|
f"{type(exc).__name__}: {exc}")
|
|
rep.check(True, f"error contract over {trials} (doc,patch) pairs")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage D: remove/replace/test outcomes vs the independent oracle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def stage_op_outcomes(rep: Report) -> None:
|
|
refs = build_refs()
|
|
for doc in DOCS:
|
|
for ref in refs:
|
|
kind, val = std_resolve(doc, ref)
|
|
for op in ("remove", "replace", "test"):
|
|
# RFC 6902 s.4: `remove` and `test` require the target to exist
|
|
# and may not target the whole document (empty pointer).
|
|
# `replace` (s.4.3) *does* allow the empty pointer: replacing the
|
|
# value at the root of the document is legal.
|
|
if op == "replace":
|
|
want_err = kind == "err"
|
|
else:
|
|
want_err = (kind == "err") or ref == ""
|
|
patch = [{"op": op, "path": ref, "value": 1}]
|
|
out = None
|
|
try:
|
|
out = E.apply_patch(doc, patch)
|
|
got_err = False
|
|
except CONTRACT:
|
|
got_err = True
|
|
except Exception as exc: # noqa: BLE001
|
|
rep.check(False, f"{op} {ref!r} leaks", type(exc).__name__)
|
|
continue
|
|
if want_err:
|
|
rep.check(got_err, f"{op} {ref!r} must fail",
|
|
"expected an error")
|
|
elif op == "test":
|
|
# value 1 vs the live value: compare with the independent oracle
|
|
rep.check(got_err == (canon(val) != canon(1)),
|
|
f"test {ref!r}",
|
|
f"live={val!r} want_err={canon(val) != canon(1)} "
|
|
f"got_err={got_err}")
|
|
else:
|
|
rep.check(not got_err, f"{op} {ref!r} must succeed",
|
|
"unexpected error")
|
|
if op == "remove":
|
|
want_doc = _std_remove(doc, ref)
|
|
rep.check(canon(out) == canon(want_doc),
|
|
f"remove {ref!r} result",
|
|
f"want {want_doc!r} got {out!r}")
|
|
|
|
|
|
def _std_remove(doc, ref: str):
|
|
"""Independent removal: mutate a deep copy of `doc` at `ref`."""
|
|
import copy as _copy
|
|
node = _copy.deepcopy(doc)
|
|
if ref == "":
|
|
return None
|
|
toks = ref[1:].split("/")
|
|
parent = node
|
|
for raw in toks[:-1]:
|
|
tok = _std_unescape(raw)
|
|
parent = parent[tok] if isinstance(parent, dict) else parent[int(tok)]
|
|
last = _std_unescape(toks[-1])
|
|
if isinstance(parent, dict):
|
|
del parent[last]
|
|
else:
|
|
del parent[int(last)]
|
|
return node
|
|
|
|
|
|
def _std_unescape(raw: str) -> str:
|
|
return raw.replace("~1", "/").replace("~0", "~")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage E: the CLI contract (exit codes, no tracebacks)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CLI_CASES = [
|
|
# (argv-suffix, stdin, expected rc, must-not-contain)
|
|
(["--doc", "D", "--patch", "P_OK"], None, 0, "Traceback"),
|
|
(["--doc", "D", "--patch", "P_BADTEST"], None, 1, "Traceback"),
|
|
(["--doc", "D", "--patch", "P_BADJSON"], None, 1, "Traceback"),
|
|
(["--doc", "D", "--patch", "P_NOFILE"], None, 1, "Traceback"),
|
|
(["--doc", "D"], None, 2, "Traceback"),
|
|
(["--nope"], None, 2, "Traceback"),
|
|
(["--doc", "D", "--patch", "P_UNICODE"], None, 1, "Traceback"),
|
|
(["--selftest"], None, 0, "Traceback"),
|
|
(["--selftest-json"], None, 0, "Traceback"),
|
|
]
|
|
|
|
|
|
def stage_cli(rep: Report) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="conformance_cli_") as tmp:
|
|
paths = {}
|
|
paths["D"] = os.path.join(tmp, "doc.json")
|
|
with open(paths["D"], "w", encoding="utf-8") as fh:
|
|
fh.write('{"foo": ["bar", "qux", "baz"], "obj": {"n": 1}, "n": 1}')
|
|
files = {
|
|
"P_OK": '[{"op":"add","path":"/foo/1","value":"zzz"},'
|
|
'{"op":"test","path":"/n","value":1}]',
|
|
"P_BADTEST": '[{"op":"test","path":"/n","value":999}]',
|
|
"P_BADJSON": 'not json',
|
|
"P_NOFILE": None,
|
|
"P_USAGE": '[{"op":"add","path":"/x","value":1}]',
|
|
"P_UNICODE": '[{"op":"remove","path":"/\\u00b2"}]',
|
|
}
|
|
for name, content in files.items():
|
|
if content is None:
|
|
paths[name] = os.path.join(tmp, "does_not_exist.json")
|
|
continue
|
|
p = os.path.join(tmp, name + ".json")
|
|
with open(p, "w", encoding="utf-8") as fh:
|
|
fh.write(content)
|
|
paths[name] = p
|
|
|
|
for suffix, _stdin, want_rc, bad in CLI_CASES:
|
|
argv = [sys.executable, ENGINE] + [
|
|
paths.get(a, a) if a in paths else a for a in suffix]
|
|
proc = subprocess.run(argv, capture_output=True, text=True,
|
|
timeout=60)
|
|
label = " ".join(suffix)
|
|
rep.check(proc.returncode == want_rc, f"cli rc {label}",
|
|
f"want {want_rc}, got {proc.returncode}")
|
|
rep.check(bad not in proc.stderr, f"cli clean stderr {label}",
|
|
" ".join(proc.stderr.split())[:100])
|
|
rep.check(proc.stderr.count("\n") <= 2 or want_rc == 0,
|
|
f"cli one-line error {label}",
|
|
" ".join(proc.stderr.split())[:100])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main(argv=None) -> int:
|
|
argv = list(sys.argv[1:] if argv is None else argv)
|
|
quiet = "--quiet" in argv
|
|
rep = Report(quiet)
|
|
|
|
stage_pointer_grammar(rep)
|
|
rep.summary("pointer grammar")
|
|
stage_resolve(rep)
|
|
rep.summary("resolve oracle")
|
|
stage_error_contract(rep)
|
|
rep.summary("error contract")
|
|
stage_op_outcomes(rep)
|
|
rep.summary("op outcomes")
|
|
stage_cli(rep)
|
|
rep.summary("cli")
|
|
|
|
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 else "VERDICT: FAIL")
|
|
return 0 if not rep.failures else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|