#!/usr/bin/env python3
"""q153_verify.py -- grade the Q153 repair: a beam test that graded the host.

The claim this gate exists to keep true:

  wizard-work/beam/test_next_greater.py resolves `next_greater` from its own
  directory, grades the COMMITTED bytes, and the lane's CHECK now reds when a
  beam test binds a host copy -- the defect that landed at `096ba9b`.

Six legs, each an independent reading, and the last three are behavioural
rather than textual:

  tool      tools/import_resolve_check.py over the work tree reads rc 0 and
            names no OUTSIDE_REPO row (the repo authority agrees with the fix).
  binding   the test file's source carries a `__file__` binding and an
            assertion over the imported module's path, and carries NO absolute
            constant in a `sys.path` call (parsed with `ast`, so a comment or
            docstring that quotes the old line is not read as a binding).
  sibling   the test and its module copied together into a directory outside
            the tree run there at rc 0 -- the file is self-contained.
  poison    the same copy with `next_greater_right` poisoned reads rc 1: the
            cell grades the bytes beside it, which is the whole repair.
  mutant    the pre-repair binding (the line committed at `096ba9b`, put back
            into a copy of the repaired file) reads rc 1 and names the host
            path -- the gate can see the defect it is meant to catch.
  lane      wizard-work/beam/check_bindings.py --selftest reads 0 failures,
            and its live reading of this lane is rc 0.

Exit: 0 every leg passed, 1 a leg faulted, 2 could not grade (a file or the
interpreter is missing). Nothing here writes to the tracked tree; every copy
lives under `--scratch`.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile

DEFAULT_REPO = os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.abspath(__file__))))
TEST_REL = os.path.join("wizard-work", "beam", "test_next_greater.py")
MODULE_REL = os.path.join("wizard-work", "beam", "next_greater.py")
TOOL_REL = os.path.join("tools", "import_resolve_check.py")
GUARD_REL = os.path.join("wizard-work", "beam", "check_bindings.py")

# The defect this gate must still be able to build: the binding line as it
# landed at 096ba9b. The mutant leg puts it back into a copy of the repaired
# file, so the gate proves it can see the class rather than asserting it can.
HEAD_BINDING = 'sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")\n'


class CannotGrade(Exception):
    pass


def _read(path):
    try:
        with open(path, "rb") as fh:
            return fh.read()
    except OSError as exc:
        raise CannotGrade("cannot read %s: %s" % (path, exc))


def _sha(raw):
    return hashlib.sha256(raw).hexdigest()


def _run(args, cwd, timeout=180):
    env = {"PATH": "/usr/bin:/bin", "HOME": os.environ.get("HOME", "/root")}
    try:
        return subprocess.run(args, cwd=cwd, capture_output=True, text=True,
                              timeout=timeout, env=env)
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise CannotGrade("could not run %s: %s" % (args[0], exc))


def _last(proc):
    lines = (proc.stdout + proc.stderr).strip().splitlines()
    return lines[-1][:200] if lines else ""


# ------------------------------------------------------------------------ legs

def leg_tool(repo, scratch):
    proc = _run([sys.executable, TOOL_REL, "--repo", repo, "--rev", "WORKTREE"],
                repo)
    if proc.returncode == 2:
        raise CannotGrade("import_resolve_check could not grade the tree")
    rows = [l for l in proc.stdout.splitlines() if l.startswith("FAULT")]
    if proc.returncode != 0 or rows:
        return False, "import_resolve_check rc %d: %s" % (
            proc.returncode, rows[0] if rows else _last(proc))
    return True, _last(proc)


def leg_binding(repo, scratch):
    raw = _read(os.path.join(repo, TEST_REL))
    text = raw.decode("utf-8")
    try:
        tree = ast.parse(text, filename=TEST_REL)
    except SyntaxError as exc:
        return False, "does not parse: line %s: %s" % (exc.lineno, exc.msg)
    absolute = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        f = node.func
        if not (isinstance(f, ast.Attribute) and f.attr in
                ("insert", "append", "extend")):
            continue
        inner = f.value
        if not (isinstance(inner, ast.Attribute) and inner.attr == "path"):
            continue
        for arg in node.args:
            if isinstance(arg, ast.Constant) and isinstance(arg.value, str) \
                    and os.path.isabs(arg.value):
                absolute.append((node.lineno, arg.value))
    if absolute:
        return False, "absolute sys.path constant at line %d: %s" % absolute[0]
    uses_file = any(isinstance(n, ast.Name) and n.id == "__file__"
                    for n in ast.walk(tree))
    asserts_path = False
    for node in tree.body:                     # module level only
        if isinstance(node, ast.Assert):
            dumped = ast.dump(node.test)
            if "next_greater.py" in dumped and "Div" in dumped:
                asserts_path = True
    if not uses_file:
        return False, "no __file__-based resolution in the file"
    if not asserts_path:
        return False, ("no module-level assert that the imported module is "
                       "HERE / next_greater.py")
    return True, "sha256 %s, __file__ binding asserted at import" % _sha(raw)


def _copy_pair(repo, dest):
    os.makedirs(dest, exist_ok=True)
    shutil.copy2(os.path.join(repo, TEST_REL), os.path.join(dest, "test_next_greater.py"))
    shutil.copy2(os.path.join(repo, MODULE_REL), os.path.join(dest, "next_greater.py"))
    return dest


def leg_sibling(repo, scratch):
    dest = _copy_pair(repo, os.path.join(scratch, "sibling"))
    proc = _run([sys.executable, "test_next_greater.py"], dest)
    if proc.returncode != 0:
        return False, "rc %d outside the tree: %s" % (proc.returncode, _last(proc))
    return True, "runs in a directory outside the repo, rc 0 (%s)" % _last(proc)


def leg_poison(repo, scratch):
    dest = _copy_pair(repo, os.path.join(scratch, "poison"))
    mod = os.path.join(dest, "next_greater.py")
    src = _read(mod).decode("utf-8")
    marker = 'def next_greater_right(nums: List[int]) -> List[int]:\n'
    if marker not in src:
        raise CannotGrade("next_greater.py no longer defines "
                          "next_greater_right with that signature")
    poisoned = src.replace(
        marker + '    """For each i, index of first j>i with nums[j] > nums[i], '
                 'else -1."""',
        marker + "    return [-2] * len(nums)  # POISON", 1)
    if poisoned == src:
        raise CannotGrade("the poison target moved: the docstring line is gone")
    with open(mod, "w") as fh:
        fh.write(poisoned)
    proc = _run([sys.executable, "test_next_greater.py"], dest)
    if proc.returncode != 1:
        return False, ("a poisoned sibling module still reads rc %d -- the cell "
                       "is not grading the bytes beside it" % proc.returncode)
    return True, "a poisoned sibling module flips the cell to rc 1"


def leg_mutant(repo, scratch):
    dest = os.path.join(scratch, "mutant")
    os.makedirs(dest, exist_ok=True)
    src = _read(os.path.join(repo, TEST_REL)).decode("utf-8")
    lines = src.splitlines(True)
    # Put the pre-repair binding back: replace the __file__ insert with the
    # host path, keeping the assert (which is what makes it fail loudly here).
    out, swapped = [], False
    for line in lines:
        if not swapped and line.startswith("sys.path.insert(0, str(HERE))"):
            out.append(HEAD_BINDING)
            swapped = True
        else:
            out.append(line)
    if not swapped:
        raise CannotGrade("the repaired insert line is absent -- cannot build "
                          "the pre-repair mutant")
    with open(os.path.join(dest, "test_next_greater.py"), "w") as fh:
        fh.write("".join(out))
    shutil.copy2(os.path.join(repo, MODULE_REL), os.path.join(dest, "next_greater.py"))
    proc = _run([sys.executable, "test_next_greater.py"], dest)
    text = proc.stdout + proc.stderr
    if proc.returncode == 0:
        return False, "the pre-repair binding still reads rc 0"
    if "wizard-outputs" not in text and "MODULE_PATH" not in text \
            and "AssertionError" not in text:
        return False, "rc %d but nothing names the host binding: %s" % (
            proc.returncode, _last(proc))
    note = _last(proc)
    if "AssertionError" in note:
        note = "assert fired: %s" % note[:160]
    return True, "pre-repair binding reads rc %d (%s)" % (proc.returncode, note)


def leg_lane(repo, scratch):
    guard = os.path.join(repo, GUARD_REL)
    if not os.path.isfile(guard):
        raise CannotGrade("%s is missing" % GUARD_REL)
    st = _run([sys.executable, guard, "--selftest"], os.path.dirname(guard))
    if st.returncode != 0:
        return False, "guard selftest rc %d: %s" % (st.returncode, _last(st))
    live = _run([sys.executable, guard], os.path.dirname(guard))
    if live.returncode != 0:
        return False, "the lane's live binding reading is rc %d" % live.returncode
    cells = [l for l in st.stdout.splitlines() if l.startswith("selftest:")]
    return True, "%s; %s" % (cells[0] if cells else "selftest rc 0", _last(live))


LEGS = [
    ("tool", leg_tool),
    ("binding", leg_binding),
    ("sibling", leg_sibling),
    ("poison", leg_poison),
    ("mutant", leg_mutant),
    ("lane", leg_lane),
]


def main(argv=None):
    ap = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
    ap.add_argument("--repo", default=DEFAULT_REPO)
    ap.add_argument("--scratch", default=None,
                    help="where the throwaway copies go (default: a temp dir)")
    ap.add_argument("--json", metavar="PATH")
    a = ap.parse_args(argv)
    repo = os.path.abspath(a.repo)
    if not os.path.isdir(os.path.join(repo, "tools")):
        print("cannot grade: %s does not look like the repo root" % repo,
              file=sys.stderr)
        return 2
    rows, faults, ungraded, rc = [], 0, 0, 0
    scratch = a.scratch or tempfile.mkdtemp(prefix="q153-")
    if os.path.isdir(scratch):
        shutil.rmtree(scratch)
    os.makedirs(scratch)
    try:
        for name, leg in LEGS:
            try:
                ok, detail = leg(repo, scratch)
            except CannotGrade as exc:
                print("UNGRADABLE %-8s %s" % (name, exc))
                rows.append({"leg": name, "ok": None, "detail": str(exc)})
                ungraded += 1
                rc = 2
                continue
            print("%s %-8s %s" % ("PASS" if ok else "FAIL", name, detail))
            rows.append({"leg": name, "ok": ok, "detail": detail})
            if not ok:
                faults += 1
    finally:
        shutil.rmtree(scratch, ignore_errors=True)
    if rc == 0:
        rc = 1 if faults else 0
    print("q153: %d leg(s), %d passed, %d fault(s), %d ungradable"
          % (len(LEGS), len(LEGS) - faults - ungraded, faults, ungraded))
    if a.json:
        with open(a.json, "w") as fh:
            json.dump({"repo": repo, "rows": rows, "rc": rc}, fh, indent=1)
    return rc


if __name__ == "__main__":
    sys.exit(main())