"""Runnable check for TEST-26 - a missing value must be a NAMED red. `python3 test_named_absence.py`

Every test file here is one script. A check or print that RAISES on a missing
value (an empty list indexed, `next()` exhausted, `:.0f` on None) kills every
check after it: the mutation it should name shows up as one unnamed traceback,
and the checks below it are never graded (MUTATION.md, FCST-06).

1. `named_absence.py`, the shared non-raising accessors.
2. A scan of every `test_*.py` for the shapes that raise, which must find none.
3. What the scan cannot see, counted on every run.
"""
import ast
import sys
from collections import Counter
from pathlib import Path

from named_absence import field, first, given, nth, present

HERE = Path(__file__).resolve().parent

fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


# ==============================================================================
print("1. the accessors never raise, and an absence is never a pass")
# ==============================================================================
_a = nth([], 0)
_b = first(x for x in (1, 2) if x > 5)
_c = field({"a": 1}, "pv")
_d = given(None)
print(f"    {_a!r} {_b!r} {_c!r} {_d!r}")
check("each accessor hands back the value when it is there",
      nth([7, 8], -1) == 8 and first(iter([3])) == 3
      and field({"pv": 0.0}, "pv") == 0.0 and given(0.0) == 0.0
      and present(0.0, None, "", []))
check("...and a stand-in that says what was missing when it is not",
      repr(_a) == "<ABSENT [0] of 0 item(s)>" and repr(_c) == "<ABSENT key 'pv'>"
      and not present(_a) and not present(1, _b) and not present(_c, _d))
check("formatting, indexing, attributes and arithmetic do not raise",
      f"{_a:.2f}|{_a[2]['x']:+.0f}|{_d.attr.get('k'):,}|{_d / 2 - 1:.1f}"
      == "|".join(["<ABSENT [0] of 0 item(s)>"] * 2 + ["<ABSENT None>"] * 2))
check("an absence is never equal, unequal, ordered or truthy - so `==`, `!=` "
      "and `<` grade it FAIL",
      not (_a == 0) and not (_a != 0) and not (_a < 1) and not (_a >= 1)
      and not bool(_a) and not (_a == _a))


# ==============================================================================
print()
print("2. no test file reads a possibly-missing value in a way that raises")
# ==============================================================================
# The scan flags, anywhere in a file:
#   index   `name[k]` (constant k) where `name` holds a FILTERED list - a
#           comprehension with an `if`, `filter()`, or an accumulator started
#           as `[]` - directly or as returned by a function in the same file;
#   next    `next(it)` without a default;
#   format  a numeric format spec on a name that may be None - assigned None,
#           an `... else None`, a one-argument `.get()`, or tested `is None`.
# A site is guarded when an earlier operand of the same `and`/`or`, or the
# test of an enclosing `if`/`while`/conditional expression, names the value,
# or it sits in a `try`. A check's NAME is evaluated before its condition, so
# a guard in the condition does not guard the name.
_NUMFMT = set("fdeEgGn%xXbo")
_FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)
_COMPS = (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)


def _filtered(rhs, returns):
    if isinstance(rhs, (ast.ListComp, ast.GeneratorExp)):
        return any(g.ifs for g in rhs.generators)
    if isinstance(rhs, ast.List):
        return not rhs.elts
    if isinstance(rhs, ast.Call) and isinstance(rhs.func, ast.Name):
        f = rhs.func.id
        if f == "filter":
            return True
        if f in ("list", "sorted", "tuple") and rhs.args:
            return _filtered(rhs.args[0], returns)
        return "listy" in returns.get((f, None), ())
    return False


def _maybe_none(rhs, returns):
    if isinstance(rhs, ast.Constant):
        return rhs.value is None
    if isinstance(rhs, ast.IfExp):
        return _maybe_none(rhs.body, returns) or _maybe_none(rhs.orelse, returns)
    if isinstance(rhs, ast.Call):
        if isinstance(rhs.func, ast.Attribute) and rhs.func.attr == "get":
            return len(rhs.args) == 1 or (len(rhs.args) == 2 and _maybe_none(rhs.args[1], returns))
        if isinstance(rhs.func, ast.Name):
            return "noney" in returns.get((rhs.func.id, None), ())
    return False


class _Scope:
    def __init__(self, node, parent):
        self.node, self.parent = node, parent
        self.bound, self.listy, self.noney = set(), set(), set()


def _dump(n):
    return ast.dump(n)


def _mentions(expr, base):
    """`base` appears in `expr` other than as the thing being indexed, so
    `x[0] == 1 and x[1]` does not count as guarding `x[1]`."""
    b = _dump(base)
    inner = {id(n.value) for n in ast.walk(expr) if isinstance(n, ast.Subscript)}
    return any(_dump(x) == b and id(x) not in inner for x in ast.walk(expr))


def _exits_on(st, base):
    """`if not x: return` (or continue/break/raise) - x is guarded below it."""
    return (isinstance(st, ast.If) and _mentions(st.test, base) and st.body
            and isinstance(st.body[-1], (ast.Return, ast.Continue, ast.Break, ast.Raise)))


def scan(src, census=None):
    """[(line, kind, text)] for every unguarded raising read in `src`."""
    tree = ast.parse(src)
    parent = {c: p for p in ast.walk(tree) for c in ast.iter_child_nodes(p)}

    def scope_node(n):
        p = parent.get(n)
        while p is not None and not isinstance(p, _FUNCS + _COMPS + (ast.Module,)):
            p = parent.get(p)
        return p

    scopes = {tree: _Scope(tree, None)}
    for n in ast.walk(tree):
        if isinstance(n, _FUNCS + _COMPS):
            scopes[n] = None
    for n in list(scopes):
        if n is not tree:
            scopes[n] = _Scope(n, None)
    for n, s in scopes.items():
        if n is not tree:
            s.parent = scopes[scope_node(n)]

    def targets(t):
        if isinstance(t, ast.Name):
            yield t
        elif isinstance(t, (ast.Tuple, ast.List)):
            for e in t.elts:
                yield from targets(e)
        elif isinstance(t, ast.Starred):
            yield from targets(t.value)

    for n in ast.walk(tree):
        if isinstance(n, _FUNCS):
            a = n.args
            for arg in a.posonlyargs + a.args + a.kwonlyargs + [a.vararg, a.kwarg]:
                if arg is not None:
                    scopes[n].bound.add(arg.arg)
        elif isinstance(n, ast.comprehension):
            for t in targets(n.target):
                scopes[scope_node(n)].bound.add(t.id)
        elif isinstance(n, (ast.For, ast.AsyncFor, ast.With, ast.AugAssign, ast.NamedExpr)):
            ts = ([n.target] if not isinstance(n, ast.With)
                  else [i.optional_vars for i in n.items if i.optional_vars])
            for t0 in ts:
                for t in targets(t0):
                    scopes[scope_node(n)].bound.add(t.id)

    # `x = <default> if x is None else x` makes x not-None: a scope that
    # normalises x this way does not get x marked by its `is None` tests.
    normalised = set()
    for n in ast.walk(tree):
        if (isinstance(n, ast.Assign) and isinstance(n.value, ast.IfExp)
                and isinstance(n.value.test, ast.Compare)):
            for t in n.targets:
                if isinstance(t, ast.Name) and any(isinstance(x, ast.Name) and x.id == t.id
                                                  for x in ast.walk(n.value.test)):
                    normalised.add((scope_node(n), t.id))

    # Two passes, so a function's return shape is known before its callers.
    returns = {}
    for _ in range(2):
        for n in ast.walk(tree):
            if isinstance(n, (ast.Assign, ast.AnnAssign)) and n.value is not None:
                s = scopes[scope_node(n)]
                for t0 in (n.targets if isinstance(n, ast.Assign) else [n.target]):
                    if isinstance(t0, ast.Name):
                        s.bound.add(t0.id)
                        if _filtered(n.value, returns):
                            s.listy.add(t0.id)
                        if _maybe_none(n.value, returns):
                            s.noney.add(t0.id)
                    elif (isinstance(t0, ast.Tuple) and isinstance(n.value, ast.Tuple)
                          and len(t0.elts) == len(n.value.elts)):
                        for e, v in zip(t0.elts, n.value.elts):
                            for t in targets(e):
                                s.bound.add(t.id)
                                if _filtered(v, returns):
                                    s.listy.add(t.id)
                                if _maybe_none(v, returns):
                                    s.noney.add(t.id)
                    elif (isinstance(t0, ast.Tuple) and isinstance(n.value, ast.Call)
                          and isinstance(n.value.func, ast.Name)):
                        for i, e in enumerate(t0.elts):
                            for t in targets(e):
                                s.bound.add(t.id)
                                got = returns.get((n.value.func.id, i), ())
                                if "listy" in got:
                                    s.listy.add(t.id)
                                if "noney" in got:
                                    s.noney.add(t.id)
                    else:
                        for t in targets(t0):
                            s.bound.add(t.id)
            elif (isinstance(n, ast.Compare)
                  and any(isinstance(o, (ast.Is, ast.IsNot)) for o in n.ops)
                  and any(isinstance(c, ast.Constant) and c.value is None
                          for c in [n.left] + n.comparators)):
                for x in [n.left] + n.comparators:
                    if isinstance(x, ast.Name) and (scope_node(n), x.id) not in normalised:
                        s = scopes[scope_node(n)]
                        while s.parent is not None and x.id not in s.bound:
                            s = s.parent
                        s.noney.add(x.id)
        for fn in tree.body:
            if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            s = scopes[fn]
            for r in ast.walk(fn):
                if not isinstance(r, ast.Return) or r.value is None or scope_node(r) is not fn:
                    continue
                vals = r.value.elts if isinstance(r.value, ast.Tuple) else [r.value]
                for i, v in enumerate(vals):
                    key = (fn.name, i if isinstance(r.value, ast.Tuple) else None)
                    tags = returns.setdefault(key, set())
                    if isinstance(v, ast.Name) and v.id in s.listy or _filtered(v, returns):
                        tags.add("listy")
                    if isinstance(v, ast.Name) and v.id in s.noney or _maybe_none(v, returns):
                        tags.add("noney")

    def resolve(name_node, attr):
        s = scopes[scope_node(name_node)]
        while s is not None:
            if name_node.id in s.bound or name_node.id in getattr(s, attr):
                return name_node.id in getattr(s, attr)
            s = s.parent
        return False

    def guarded(node, base):
        ch, p = node, parent.get(node)
        while p is not None:
            if isinstance(p, ast.BoolOp) and ch in p.values:
                if any(_mentions(v, base) for v in p.values[:p.values.index(ch)]):
                    return True
            elif isinstance(p, (ast.IfExp, ast.If, ast.While)):
                if ch is not p.test and _mentions(p.test, base):
                    return True
            elif isinstance(p, (ast.ListComp, ast.GeneratorExp, ast.SetComp)):
                if ch is p.elt and any(_mentions(i, base) for g in p.generators for i in g.ifs):
                    return True
            elif isinstance(p, ast.Try) and ch in p.body:
                return True
            for field_ in ("body", "orelse"):
                block = getattr(p, field_, None)
                if (isinstance(block, list) and ch in block
                        and any(_exits_on(st, base) for st in block[:block.index(ch)])):
                    return True
            if isinstance(p, _FUNCS):
                return False
            ch, p = p, parent.get(p)
        return False

    out = []
    for n in ast.walk(tree):
        if (isinstance(n, ast.Subscript) and isinstance(n.ctx, ast.Load)
                and isinstance(n.value, ast.Name)):
            k = n.slice
            if isinstance(k, ast.UnaryOp) and isinstance(k.op, ast.USub):
                k = k.operand
            if (isinstance(k, ast.Constant) and type(k.value) is int
                    and resolve(n.value, "listy") and not guarded(n, n.value)):
                out.append((n.lineno, "index", ast.unparse(n)))
        elif (isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
              and n.func.id == "next" and len(n.args) == 1 and not n.keywords):
            if not guarded(n, n.args[0]):
                out.append((n.lineno, "next", ast.unparse(n)[:60]))
        elif (isinstance(n, ast.FormattedValue) and n.format_spec is not None
              and isinstance(n.value, ast.Name)):
            spec = "".join(v.value for v in n.format_spec.values
                           if isinstance(v, ast.Constant))
            if (spec and (spec[-1] in _NUMFMT or "," in spec)
                    and resolve(n.value, "noney") and not guarded(n, n.value)):
                out.append((n.lineno, "format", f"{{{n.value.id}:{spec}}}"))
    if census is not None:
        _count_blind(tree, resolve, census)
    return sorted(set(out))


def _count_blind(tree, resolve, census):
    """Count, inside check() and print() arguments, the forms part 2 skips."""
    for call in ast.walk(tree):
        if not (isinstance(call, ast.Call) and isinstance(call.func, ast.Name)
                and call.func.id in ("check", "print")):
            continue
        for n in (x for a in call.args for x in ast.walk(a)):
            if isinstance(n, ast.Subscript) and isinstance(n.ctx, ast.Load):
                k = n.slice.operand if isinstance(n.slice, ast.UnaryOp) else n.slice
                if not isinstance(k, ast.Constant):
                    continue
                if isinstance(k.value, str):
                    census["key"] += 1
                elif isinstance(n.value, ast.Name):
                    if not resolve(n.value, "listy"):
                        census["name-list"] += 1
                elif isinstance(n.value, ast.Attribute):
                    census["attr-list"] += 1
                else:
                    census["other-index"] += 1
            elif isinstance(n, ast.FormattedValue) and n.format_spec is not None:
                if not isinstance(n.value, ast.Name):
                    census["format-expr"] += 1
                elif not resolve(n.value, "noney"):
                    census["format-name"] += 1
            elif isinstance(n, ast.BinOp) and isinstance(n.op, (ast.Div, ast.FloorDiv)):
                census["division"] += 1


# The scan must bite before its silence means anything: one planted site of
# each kind, and the guarded spelling of each, which must NOT be flagged.
_PLANTED = '''
def run():
    seen = []
    t_gate = None
    return seen, t_gate
rows = [r for r in range(3) if r > 9]
seen, t_gate = run()
waits, n = [], 0
check(f"first {rows[0]}", bool(rows))
print(waits[0])
check("pair", rows[0] == 1 and rows[1] == 2)
print(seen[-1])
print(f"{t_gate:.2f}")
print(next(r for r in rows))
'''
_GUARDED = '''
def run():
    seen = []
    t_gate = None
    return seen, t_gate
rows = [r for r in range(3) if r > 9]
seen, t_gate = run()
check("first", bool(rows) and rows[0] == 1)
print(seen[-1] if seen else "-")
check("pair", len(rows) == 2 and rows[0] < rows[1])
print(f"{t_gate:.2f}" if t_gate is not None else "-")
print(next((r for r in rows), None))
def show(rows):
    print(rows[0])
def head():
    if not rows:
        return None
    return rows[0]
'''
_hit = scan(_PLANTED)
print(f"    planted: {_hit}")
check("the scan flags a planted site of each kind - a filtered list indexed in "
      "a check NAME whose condition guards it, an accumulator bound in a tuple "
      "and one returned from a function, a maybe-None formatted, and next() "
      "without a default",
      [(k, t) for _l, k, t in _hit] == [
          ("index", "rows[0]"), ("index", "waits[0]"), ("index", "rows[0]"),
          ("index", "rows[1]"), ("index", "seen[-1]"),
          ("format", "{t_gate:.2f}"),
          ("next", "next((r for r in rows))")])
check("...and flags none of the guarded spellings - `and`, a conditional, "
      "an early return, a default - nor a parameter that shadows a filtered "
      "name", scan(_GUARDED) == [])

_SUITES = sorted(HERE.glob("test_*.py"))
_found = {p.name: scan(p.read_text(encoding="utf-8")) for p in _SUITES}
_all = [(f, *h) for f, hs in _found.items() for h in hs]
for f, line, kind, text in _all:
    print(f"    {f}:{line}  {kind:6}  {text}")
check(f"no unguarded raising read in any of the {len(_SUITES)} test files "
      f"({len(_all)} found) - read it through named_absence.py and conjoin "
      f"present() in the check", _all == [] and len(_SUITES) >= 30)


# ==============================================================================
print()
print("3. what part 2 cannot see, counted - these can still raise")
# ==============================================================================
# The scan knows a list is filtered, or a name may be None, only from its own
# file's syntax. Each form below can still raise and is NOT graded; the counts
# are of every such site in check() and print() arguments, largest first.
_BLIND = {
    "key": "`x['k']` - a dict key the code under test may omit",
    "name-list": "`x[k]` on a list the scan cannot see is filtered - returned "
                 "by another module or a method, or of a fixed length",
    "format-name": "a numeric format on a name the scan has no reason to think "
                   "is None - e.g. set from another module's function",
    "format-expr": "a numeric format on an attribute, subscript or call",
    "attr-list": "`obj.items[k]` - an attribute list",
    "other-index": "`x[k]` on a call, a slice or a nested subscript",
    "division": "`/` or `//` by a value that can be 0",
}
_census = Counter()
for p in _SUITES:
    scan(p.read_text(encoding="utf-8"), _census)
for k, why in sorted(_BLIND.items(), key=lambda kv: -_census[kv[0]]):
    print(f"    {_census[k]:5d}  {why}")
print("    never counted: these forms outside check()/print() arguments; "
      "attribute access, iteration, int() or range() on a None or an ABSENT; "
      "a raise inside a fixture function or inside app/.")
check("every counted form is in the printed table", set(_census) <= set(_BLIND))


print()
if fails:
    print(f"{len(fails)} CHECK(S) FAILED")
    sys.exit(1)
print("all checks passed")