#!/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())