peter
f8a8e905da
The scan followed many shapes and called the rest unread, but each shape it learned opened a silent branch: a parameter rebound in the body, a second visit of a followed helper, an `import app.x`, a name bound twice or under an `if`, a spread or a store that replaces the code, a detail rebuilt from its variable, a None reason. It now reads only what the app writes and reports everything else: a branch the app never takes is gone (an `or`, a default, a base class's attribute, string and union annotations, a bare module constant), a spread is clean only from an untouched `**kwargs` no call puts a `code` in, and the detail's variable may only grow by `|=` or `["reason"]`. The real tree reads exactly as before; every exit and every kept branch has a self-test that a mutant of it fails. The detail test also fails a dict without a `code`, a JSONResponse sentence and a detail-less HTTPException; the two that are no refusal (the setup gate's 503, an unmatched api path's 404) are listed with why. The contract's assertion that compared the reason map with itself is gone.
770 lines
34 KiB
Python
770 lines
34 KiB
Python
"""Every refusal the backend can answer, read from its syntax tree: each `{"code": …, "reason": …}` detail,
|
|
with every reason its code can be sent with.
|
|
|
|
The frontend's contract (`i18n/backendIds.contract.test.ts`) runs this and requires a sentence for each
|
|
code and reason in both languages. A regex over the source read only the shapes it was written for, and a
|
|
scan that tries to follow every shape opens a silent gap with each one it learns. So this one reads a
|
|
CLOSED list, the shapes the app writes, and reports everything else, never guesses:
|
|
|
|
- a reason: a string literal (`None` sends none: the site counts as reasonless); both arms of
|
|
`a if c else b`; a local variable bound only by plain `x = …`; a module constant reached as `mod.NAME`;
|
|
a call's `Literal` return annotation;
|
|
- a parameter typed with a `Literal` (basedpyright holds every value passed in), or a `str` one of a
|
|
module-level helper (`refused`, `not_found`), followed to every call, each of which must pass it by
|
|
position or by name; a parameter its body binds again is unread;
|
|
- an attribute: `e.x` of the exception an `except C as e` catches, or of a parameter typed `C`, when a
|
|
method of `C` declares `self.x: <Literal>`;
|
|
- a map lookup (`_SETTING_REASONS[exc.reason]`): the map's values, and its keys must be exactly what the
|
|
index can be;
|
|
- the code: one literal, never changed after. A `**` spread in the detail is the `**kwargs` of a helper or
|
|
of an exception's `__init__` that takes `reason` by name, used for nothing but being spread, and no call
|
|
of it passes `code=` or a `**` that is not such a spread itself. The detail's variable is changed only by
|
|
`|= {…}` or `["reason"] = …`, and goes nowhere but into an `HTTPException` / `JSONResponse`.
|
|
|
|
A name bound twice at module level, inside a module-level `if`/`try`/loop, or imported from two places is
|
|
not looked up. A reference named like a followed helper that the scan cannot resolve (`import app.x`, a
|
|
relative import) is unread. Anything else is UNREAD, or an unread code, and the contract fails on it unless
|
|
no screen shows that code. Stdlib only: it runs on the host, beside the frontend's tests
|
|
(`python -I backend/tests/refusal_sites.py` → JSON).
|
|
"""
|
|
|
|
import ast
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
from collections.abc import Iterator
|
|
from dataclasses import dataclass, field
|
|
|
|
APP = pathlib.Path(__file__).resolve().parents[1] / "app"
|
|
|
|
# Files whose `"code"` dicts are data, not refusals: a subtitle track's status payload, TMDB country codes.
|
|
NOT_REFUSALS = {
|
|
"plex/subtitles.py": "a subtitle track's status payload (its own `code`)",
|
|
"plex/tmdb.py": "a production country's ISO code",
|
|
}
|
|
|
|
# What a detail held in a variable may be handed to: the response it is sent in, nothing that could change it.
|
|
SENDERS = {"HTTPException", "JSONResponse"}
|
|
READS = {"get", "items", "keys", "values", "copy"}
|
|
|
|
Node = ast.AST
|
|
FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef)
|
|
SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)
|
|
COMPOUND = (
|
|
ast.If,
|
|
ast.Try,
|
|
ast.TryStar,
|
|
ast.For,
|
|
ast.AsyncFor,
|
|
ast.While,
|
|
ast.With,
|
|
ast.AsyncWith,
|
|
ast.Match,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Res:
|
|
values: set[str] = field(default_factory=set)
|
|
unread: list[str] = field(default_factory=list)
|
|
none: bool = False # an arm sends no reason
|
|
|
|
def __or__(self, other: "Res") -> "Res":
|
|
return Res(self.values | other.values, self.unread + other.unread, self.none or other.none)
|
|
|
|
|
|
@dataclass
|
|
class Module:
|
|
path: str # relative to app/, posix
|
|
name: str # dotted, `app.…`
|
|
tree: ast.Module
|
|
parents: dict[Node, Node]
|
|
# local name → (module, symbol or None for the module)
|
|
imports: dict[str, tuple[str, str | None]]
|
|
top: dict[str, Node] # module-level bindings: def, class, or the assignment
|
|
ambiguous: set[str] # bound more than once, conditionally, or imported from two places
|
|
|
|
|
|
def _own_scope(node: Node) -> Iterator[Node]:
|
|
"""`node` and what it holds in its own scope: a nested def, class or lambda is itself, not its body."""
|
|
stack = [node]
|
|
while stack:
|
|
n = stack.pop()
|
|
yield n
|
|
if n is not node and isinstance(n, SCOPES):
|
|
continue
|
|
stack.extend(ast.iter_child_nodes(n))
|
|
|
|
|
|
def _stored(target: Node) -> set[str]:
|
|
"""The names an assignment target binds, not the ones it reads (`LABELS[k] = v` binds none)."""
|
|
return {
|
|
n.id for n in ast.walk(target) if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store)
|
|
}
|
|
|
|
|
|
def _binds(node: Node) -> set[str]:
|
|
"""The names one node binds in the scope it sits in."""
|
|
if isinstance(node, ast.Assign):
|
|
return set().union(*(_stored(t) for t in node.targets))
|
|
if isinstance(node, (ast.AnnAssign, ast.AugAssign, ast.For, ast.AsyncFor, ast.NamedExpr)):
|
|
return _stored(node.target)
|
|
if isinstance(node, (ast.With, ast.AsyncWith)):
|
|
return set().union(*(_stored(i.optional_vars) for i in node.items if i.optional_vars))
|
|
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
|
return {(a.asname or a.name).split(".")[0] for a in node.names}
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
|
return {node.name}
|
|
if isinstance(node, (ast.MatchAs, ast.MatchStar, ast.ExceptHandler)) and node.name:
|
|
return {node.name}
|
|
return set()
|
|
|
|
|
|
def _module_name(path: str) -> str:
|
|
parts = path.removesuffix(".py").split("/")
|
|
if parts[-1] == "__init__":
|
|
parts = parts[:-1]
|
|
return ".".join(["app", *parts])
|
|
|
|
|
|
def _load(sources: dict[str, str]) -> dict[str, Module]:
|
|
mods: dict[str, Module] = {}
|
|
for path, src in sources.items():
|
|
tree = ast.parse(src)
|
|
parents = {child: node for node in ast.walk(tree) for child in ast.iter_child_nodes(node)}
|
|
origins: dict[str, set[tuple[str, str | None]]] = {}
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
|
|
for a in node.names:
|
|
origins.setdefault(a.asname or a.name, set()).add((node.module, a.name))
|
|
elif isinstance(node, ast.Import):
|
|
for a in node.names:
|
|
if a.asname:
|
|
origins.setdefault(a.asname, set()).add((a.name, None))
|
|
top: dict[str, Node] = {}
|
|
count: dict[str, int] = {}
|
|
ambiguous = {name for name, o in origins.items() if len(o) > 1}
|
|
for stmt in tree.body:
|
|
if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
|
top[stmt.name] = stmt
|
|
elif isinstance(stmt, ast.Assign):
|
|
for t in stmt.targets:
|
|
if isinstance(t, ast.Name):
|
|
top[t.id] = stmt
|
|
elif isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name):
|
|
top[stmt.target.id] = stmt
|
|
for n in [stmt] if isinstance(stmt, SCOPES) else _own_scope(stmt):
|
|
for name in _binds(n):
|
|
count[name] = count.get(name, 0) + 1
|
|
if isinstance(stmt, COMPOUND):
|
|
ambiguous.add(name)
|
|
ambiguous |= {name for name, c in count.items() if c > 1}
|
|
imports = {name: next(iter(o)) for name, o in origins.items() if name not in ambiguous}
|
|
name = _module_name(path)
|
|
mods[name] = Module(path, name, tree, parents, imports, top, ambiguous)
|
|
return mods
|
|
|
|
|
|
class Scan:
|
|
def __init__(self, sources: dict[str, str]):
|
|
self.mods = _load(sources)
|
|
self.errors: list[str] = []
|
|
self._following: set[tuple[str, str, str]] = set()
|
|
self._followed: dict[tuple[str, str, str], Res] = {}
|
|
self._clean: dict[int, bool] = {}
|
|
|
|
# --- names ---------------------------------------------------------------------------------------
|
|
|
|
def lookup(
|
|
self, m: Module, name: str, seen: frozenset[tuple[str, str]] = frozenset()
|
|
) -> tuple[Module, Node] | tuple[str, None] | None:
|
|
"""What a module-level name is: (its module, its binding), or (a module's name, None)."""
|
|
if name in m.ambiguous:
|
|
return None
|
|
if name in m.top:
|
|
return m, m.top[name]
|
|
if name in m.imports:
|
|
mod, sym = m.imports[name]
|
|
if sym is None:
|
|
return (mod, None) if mod in self.mods else None
|
|
if f"{mod}.{sym}" in self.mods:
|
|
return f"{mod}.{sym}", None
|
|
if mod in self.mods and (mod, sym) not in seen:
|
|
return self.lookup(self.mods[mod], sym, seen | {(mod, sym)})
|
|
return None
|
|
|
|
def lookup_expr(self, m: Module, e: Node) -> tuple[Module, Node] | tuple[str, None] | None:
|
|
if isinstance(e, ast.Name):
|
|
return None if self.local(m, e) else self.lookup(m, e.id)
|
|
if isinstance(e, ast.Attribute):
|
|
base = self.lookup_expr(m, e.value)
|
|
if base and base[1] is None and isinstance(base[0], str):
|
|
return self.lookup(self.mods[base[0]], e.attr)
|
|
return None
|
|
|
|
def local(self, m: Module, e: ast.Name) -> bool:
|
|
"""`e` names a parameter or a variable of a function it sits in."""
|
|
fn = self.enclosing(m, e, FUNCS)
|
|
while fn is not None:
|
|
a = fn.args
|
|
if e.id in {
|
|
p.arg for p in [*a.posonlyargs, *a.args, *a.kwonlyargs, a.vararg, a.kwarg] if p
|
|
}:
|
|
return True
|
|
if self.assignments(fn, e.id):
|
|
return True
|
|
fn = self.enclosing(m, fn, FUNCS)
|
|
return False
|
|
|
|
def at(self, m: Module, e: Node) -> str:
|
|
return f"{m.path}:{getattr(e, 'lineno', '?')}: {ast.unparse(e)}"
|
|
|
|
def enclosing(self, m: Module, e: Node, kind: type | tuple[type, ...]) -> Node | None:
|
|
node = m.parents.get(e)
|
|
while node is not None and not isinstance(node, kind):
|
|
node = m.parents.get(node)
|
|
return node
|
|
|
|
def assignments(self, fn: Node, name: str) -> list[Node]:
|
|
"""What binds `name` in this function's own body (not in a nested one)."""
|
|
return [n for n in _own_scope(fn) if n is not fn and name in _binds(n)]
|
|
|
|
def caught_as(self, m: Module, e: Node, name: str) -> ast.ExceptHandler | None:
|
|
node = m.parents.get(e)
|
|
while node is not None and not isinstance(node, FUNCS):
|
|
if isinstance(node, ast.ExceptHandler) and node.name == name:
|
|
return node
|
|
node = m.parents.get(node)
|
|
return None
|
|
|
|
def references(
|
|
self, m: Module, target: Node
|
|
) -> Iterator[tuple[Module, Node, ast.Call | None, bool]]:
|
|
"""Every reference to a module-level def or class: the call it makes (None: not a call), and
|
|
whether it resolved. A reference named like it that the scan cannot resolve is yielded too."""
|
|
name = getattr(target, "name", None)
|
|
for cm in self.mods.values():
|
|
for node in ast.walk(cm.tree):
|
|
if not isinstance(node, (ast.Name, ast.Attribute)) or not isinstance(
|
|
node.ctx, ast.Load
|
|
):
|
|
continue
|
|
if (node.id if isinstance(node, ast.Name) else node.attr) != name:
|
|
continue
|
|
parent = cm.parents.get(node)
|
|
if isinstance(parent, ast.Attribute) and parent.value is node:
|
|
continue
|
|
found = self.lookup_expr(cm, node)
|
|
if found:
|
|
if not (found[0] is m and found[1] is target):
|
|
continue
|
|
elif not self.unresolved(cm, node):
|
|
continue
|
|
call = parent if isinstance(parent, ast.Call) and parent.func is node else None
|
|
yield cm, node, call, found is not None
|
|
|
|
def unresolved(self, m: Module, node: Node) -> bool:
|
|
"""A name, or a dotted path, the app could mean that the scan cannot resolve: not a local, nor an
|
|
outside module's (an attribute of an object the function holds is not one)."""
|
|
root = node
|
|
while isinstance(root, ast.Attribute):
|
|
root = root.value
|
|
if not isinstance(root, ast.Name) or self.local(m, root):
|
|
return False
|
|
if root.id in m.ambiguous or (root.id not in m.top and root.id not in m.imports):
|
|
return True
|
|
if root is node or root.id in m.top:
|
|
return False
|
|
found = self.lookup(m, root.id)
|
|
return (
|
|
found is not None and found[1] is None
|
|
) # an app module whose dotted path went nowhere
|
|
|
|
# --- types ---------------------------------------------------------------------------------------
|
|
|
|
def ann_values(self, m: Module, ann: Node | None, depth: int = 0) -> set[str] | None:
|
|
"""The values a `Literal[…]` annotation, or an alias of one, allows; None when it is not one."""
|
|
if isinstance(ann, ast.Subscript) and getattr(ann.value, "id", None) == "Literal":
|
|
items = ann.slice.elts if isinstance(ann.slice, ast.Tuple) else [ann.slice]
|
|
if all(isinstance(i, ast.Constant) and isinstance(i.value, str) for i in items):
|
|
return {i.value for i in items}
|
|
return None
|
|
if isinstance(ann, (ast.Name, ast.Attribute)) and depth < 8:
|
|
found = self.lookup_expr(m, ann)
|
|
if found and isinstance(found[0], Module) and isinstance(found[1], ast.Assign):
|
|
return self.ann_values(found[0], found[1].value, depth + 1)
|
|
return None
|
|
|
|
def classes(self, m: Module, e: Node | None) -> list[tuple[Module, ast.ClassDef]] | None:
|
|
found = self.lookup_expr(m, e) if e is not None else None
|
|
if found and isinstance(found[0], Module) and isinstance(found[1], ast.ClassDef):
|
|
return [(found[0], found[1])]
|
|
return None
|
|
|
|
def typed(self, m: Module, at: Node, name: str) -> list[tuple[Module, ast.ClassDef]] | None:
|
|
"""The class `name` holds where `at` sits: the one an `except C as name` catches, or a parameter's
|
|
annotation (a parameter its body binds again holds no known class)."""
|
|
handler = self.caught_as(m, at, name)
|
|
if handler is not None:
|
|
return self.classes(m, handler.type)
|
|
fn = self.enclosing(m, at, FUNCS)
|
|
if fn is None:
|
|
return None
|
|
params = {a.arg: a for a in [*fn.args.posonlyargs, *fn.args.args, *fn.args.kwonlyargs]}
|
|
if name in params and not self.assignments(fn, name):
|
|
return self.classes(m, params[name].annotation)
|
|
return None
|
|
|
|
def attr_values(self, cm: Module, cls: ast.ClassDef, attr: str) -> set[str] | None:
|
|
"""An attribute's `Literal` values, from `self.attr: T` in a method of the class."""
|
|
for node in ast.walk(cls):
|
|
if (
|
|
isinstance(node, ast.AnnAssign)
|
|
and isinstance(node.target, ast.Attribute)
|
|
and node.target.attr == attr
|
|
and getattr(node.target.value, "id", None) == "self"
|
|
):
|
|
return self.ann_values(cm, node.annotation)
|
|
return None
|
|
|
|
# --- reasons -------------------------------------------------------------------------------------
|
|
|
|
def resolve(self, m: Module, e: Node) -> Res:
|
|
if isinstance(e, ast.Constant):
|
|
if e.value is None:
|
|
return Res(none=True)
|
|
if isinstance(e.value, str):
|
|
return Res({e.value})
|
|
if isinstance(e, ast.IfExp):
|
|
return self.resolve(m, e.body) | self.resolve(m, e.orelse)
|
|
if isinstance(e, ast.Name):
|
|
return self.resolve_name(m, e)
|
|
if isinstance(e, ast.Attribute):
|
|
return self.resolve_attr(m, e)
|
|
if isinstance(e, ast.Call):
|
|
found = self.lookup_expr(m, e.func)
|
|
if found and isinstance(found[0], Module) and isinstance(found[1], FUNCS):
|
|
v = self.ann_values(found[0], found[1].returns)
|
|
if v is not None:
|
|
return Res(v)
|
|
if isinstance(e, ast.Subscript):
|
|
return self.resolve_lookup(m, e)
|
|
return Res(unread=[self.at(m, e)])
|
|
|
|
def resolve_name(self, m: Module, e: ast.Name) -> Res:
|
|
fn = self.enclosing(m, e, FUNCS)
|
|
while fn is not None:
|
|
assert isinstance(fn, FUNCS)
|
|
if self.caught_as(m, e, e.id) is not None:
|
|
return Res(unread=[self.at(m, e)]) # the exception itself, not a reason
|
|
a = fn.args
|
|
params = {p.arg: p for p in [*a.posonlyargs, *a.args, *a.kwonlyargs]}
|
|
assigned = self.assignments(fn, e.id)
|
|
if e.id in params:
|
|
if assigned:
|
|
return Res(unread=[f"{self.at(m, e)} (a parameter its body binds again)"])
|
|
return self.resolve_param(m, fn, params[e.id], e)
|
|
if assigned:
|
|
out = Res()
|
|
for s in assigned:
|
|
if isinstance(s, ast.Assign) and all(
|
|
isinstance(t, ast.Name) for t in s.targets
|
|
):
|
|
out = out | self.resolve(m, s.value)
|
|
else:
|
|
out = out | Res(unread=[self.at(m, e)])
|
|
return out
|
|
fn = self.enclosing(m, fn, FUNCS)
|
|
return Res(unread=[self.at(m, e)])
|
|
|
|
def resolve_param(self, m: Module, fn: Node, param: ast.arg, e: Node) -> Res:
|
|
v = self.ann_values(m, param.annotation)
|
|
if v is not None:
|
|
return Res(v)
|
|
if param.annotation is not None and getattr(param.annotation, "id", None) != "str":
|
|
return Res(unread=[self.at(m, e)])
|
|
return self.follow(m, fn, param)
|
|
|
|
def follow(self, m: Module, fn: Node, param: ast.arg) -> Res:
|
|
"""A `str` parameter of a helper: what every call of the helper passes it."""
|
|
assert isinstance(fn, FUNCS)
|
|
key = (m.name, fn.name, param.arg)
|
|
if key in self._followed:
|
|
return self._followed[key]
|
|
where = f"{m.path}:{fn.lineno}: {fn.name}({param.arg})"
|
|
if key in self._following:
|
|
return Res(unread=[f"{where} reached again while it is followed"])
|
|
if m.top.get(fn.name) is not fn or fn.name in m.ambiguous:
|
|
return Res(unread=[f"{where} is not a module-level helper"])
|
|
self._following.add(key)
|
|
positional = [a.arg for a in [*fn.args.posonlyargs, *fn.args.args]]
|
|
by_name = param not in fn.args.posonlyargs
|
|
out = Res()
|
|
for cm, node, call, resolved in self.references(m, fn):
|
|
if not resolved:
|
|
out = out | Res(
|
|
unread=[f"{self.at(cm, node)} (a reference the scan cannot resolve)"]
|
|
)
|
|
continue
|
|
if call is None:
|
|
out = out | Res(unread=[f"{self.at(cm, node)} (the helper used as a value)"])
|
|
continue
|
|
kw = {k.arg: k.value for k in call.keywords}
|
|
given = param.arg in positional and positional.index(param.arg) < len(call.args)
|
|
if by_name and param.arg in kw:
|
|
out = out | self.resolve(cm, kw[param.arg])
|
|
elif any(isinstance(a, ast.Starred) for a in call.args):
|
|
out = out | Res(unread=[self.at(cm, call)])
|
|
elif given:
|
|
out = out | self.resolve(cm, call.args[positional.index(param.arg)])
|
|
else:
|
|
out = out | Res(unread=[f"{self.at(cm, call)} (does not pass {param.arg})"])
|
|
self._followed[key] = out # once followed, its key answers from here: never reached again
|
|
return out
|
|
|
|
def resolve_attr(self, m: Module, e: ast.Attribute) -> Res:
|
|
base = e.value
|
|
if isinstance(base, ast.Name):
|
|
typed = self.typed(m, e, base.id)
|
|
if typed is not None:
|
|
out = Res()
|
|
for cm, cls in typed:
|
|
v = self.attr_values(cm, cls, e.attr)
|
|
out = out | (
|
|
Res(v)
|
|
if v is not None
|
|
else Res(
|
|
unread=[f"{self.at(m, e)} ({cls.name}.{e.attr} has no Literal type)"]
|
|
)
|
|
)
|
|
return out
|
|
found = self.lookup_expr(m, e)
|
|
if (
|
|
found
|
|
and isinstance(found[0], Module)
|
|
and isinstance(found[1], (ast.Assign, ast.AnnAssign))
|
|
):
|
|
stmt = found[1]
|
|
if isinstance(stmt.value, ast.Constant):
|
|
return self.resolve(found[0], stmt.value)
|
|
return Res(unread=[self.at(m, e)])
|
|
|
|
def resolve_lookup(self, m: Module, e: ast.Subscript) -> Res:
|
|
found = self.lookup_expr(m, e.value)
|
|
if not (
|
|
found
|
|
and isinstance(found[0], Module)
|
|
and isinstance(found[1], (ast.Assign, ast.AnnAssign))
|
|
):
|
|
return Res(unread=[self.at(m, e)])
|
|
table = found[1].value
|
|
if not isinstance(table, ast.Dict) or not all(
|
|
isinstance(k, ast.Constant) and isinstance(k.value, str) for k in table.keys
|
|
):
|
|
return Res(unread=[self.at(m, e)])
|
|
out = Res()
|
|
for v in table.values:
|
|
out = out | self.resolve(found[0], v)
|
|
index = self.resolve(m, e.slice)
|
|
keys = {k.value for k in table.keys if isinstance(k, ast.Constant)}
|
|
if index.unread or index.none or index.values != keys:
|
|
self.errors.append(
|
|
f"{self.at(m, e)}: the map's keys {sorted(keys)} are not what its index can be "
|
|
f"{sorted(index.values)}{' + unread ' + str(index.unread) if index.unread else ''}"
|
|
)
|
|
return out
|
|
|
|
# --- spreads -------------------------------------------------------------------------------------
|
|
|
|
def clean_spread(self, m: Module, at: Node, src: Node) -> bool:
|
|
"""`**src` cannot carry a `reason` or a `code`: it is a function's `**kwargs` (or a caught
|
|
exception's `e.params` kept from its `__init__`'s), or `{k: … for k, v in that.items()}`: a
|
|
value may change, a key may not."""
|
|
if isinstance(src, ast.DictComp):
|
|
g = src.generators
|
|
if len(g) != 1 or not isinstance(g[0].target, ast.Tuple) or len(g[0].target.elts) != 2:
|
|
return False
|
|
k, v = g[0].target.elts
|
|
it = g[0].iter
|
|
return (
|
|
isinstance(k, ast.Name)
|
|
and isinstance(v, ast.Name)
|
|
and getattr(src.key, "id", None) == k.id
|
|
and isinstance(it, ast.Call)
|
|
and isinstance(it.func, ast.Attribute)
|
|
and it.func.attr == "items"
|
|
and not it.args
|
|
and self.clean_spread(m, at, it.func.value)
|
|
)
|
|
if isinstance(src, ast.Attribute) and isinstance(src.value, ast.Name):
|
|
typed = self.typed(m, at, src.value.id)
|
|
return bool(typed) and all(self.kept(cm, cls, src.attr) for cm, cls in typed or [])
|
|
if isinstance(src, ast.Name):
|
|
fn = self.enclosing(m, at, FUNCS)
|
|
return (
|
|
fn is not None
|
|
and fn.args.kwarg is not None
|
|
and src.id == fn.args.kwarg.arg
|
|
and self.kwargs_clean(m, fn, fn)
|
|
)
|
|
return False
|
|
|
|
def kept(self, cm: Module, cls: ast.ClassDef, attr: str) -> bool:
|
|
"""`self.attr` is only ever the `**kwargs` of the class's `__init__`, and that one is clean."""
|
|
init = next(
|
|
(s for s in cls.body if isinstance(s, ast.FunctionDef) and s.name == "__init__"), None
|
|
)
|
|
if init is None or init.args.kwarg is None:
|
|
return False
|
|
stores = [
|
|
n
|
|
for n in ast.walk(cls)
|
|
if isinstance(n, ast.Attribute)
|
|
and n.attr == attr
|
|
and getattr(n.value, "id", None) == "self"
|
|
and not isinstance(n.ctx, ast.Load)
|
|
]
|
|
kept = [
|
|
s
|
|
for s in stores
|
|
if isinstance(p := cm.parents.get(s), ast.Assign)
|
|
and getattr(p.value, "id", None) == init.args.kwarg.arg
|
|
and self.enclosing(cm, p, FUNCS) is init
|
|
]
|
|
return len(stores) == 1 and len(kept) == 1 and self.kwargs_clean(cm, init, cls)
|
|
|
|
def kwargs_clean(self, m: Module, fn: Node, target: Node) -> bool:
|
|
"""`fn`'s `**kwargs` holds no `reason` (`fn` takes it by name) and no `code`: it is used for
|
|
nothing but `**kwargs`, `kwargs.items()`, `self.x = kwargs` or an f-string, and every call of
|
|
`target` (the helper, or the class `fn` is the `__init__` of) passes no `code=` and no unclean
|
|
`**`."""
|
|
assert isinstance(fn, FUNCS) and fn.args.kwarg is not None
|
|
if id(fn) in self._clean:
|
|
return self._clean[id(fn)]
|
|
self._clean[id(fn)] = False # a call that reaches it again while it is checked is not clean
|
|
kwarg = fn.args.kwarg.arg
|
|
clean = "reason" in {a.arg for a in [*fn.args.args, *fn.args.kwonlyargs]} and not (
|
|
self.assignments(fn, kwarg)
|
|
)
|
|
for n in _own_scope(fn):
|
|
if not clean:
|
|
break
|
|
if isinstance(n, ast.Name) and n.id == kwarg:
|
|
p = m.parents.get(n)
|
|
clean = (
|
|
isinstance(p, ast.FormattedValue) # printed in a message
|
|
or isinstance(p, ast.keyword)
|
|
and p.arg is None
|
|
or isinstance(p, ast.Dict)
|
|
and any(k is None and v is n for k, v in zip(p.keys, p.values, strict=True))
|
|
or isinstance(p, ast.Attribute)
|
|
and p.attr == "items"
|
|
and isinstance(m.parents.get(p), ast.Call)
|
|
or isinstance(p, ast.Assign)
|
|
and p.value is n
|
|
and all(
|
|
isinstance(t, ast.Attribute) and getattr(t.value, "id", None) == "self"
|
|
for t in p.targets
|
|
)
|
|
)
|
|
refs = list(self.references(m, target)) if clean else []
|
|
for cm, node, call, resolved in refs:
|
|
if call is not None:
|
|
clean = not any(
|
|
k.arg == "code" or k.arg is None and not self.clean_spread(cm, call, k.value)
|
|
for k in call.keywords
|
|
)
|
|
elif not resolved or not isinstance(target, ast.ClassDef):
|
|
clean = False # a helper passed around, or a name the scan cannot resolve
|
|
else:
|
|
# `except C`, an annotation: not a construction; a subclass is built some other way
|
|
cls = self.enclosing(cm, node, ast.ClassDef)
|
|
clean = not (cls is not None and any(node in ast.walk(b) for b in cls.bases))
|
|
if not clean:
|
|
break
|
|
self._clean[id(fn)] = clean
|
|
return clean
|
|
|
|
# --- sites ---------------------------------------------------------------------------------------
|
|
|
|
def entries(self, d: Node) -> list[tuple[str | None, Node]]:
|
|
"""A dict's (key, value) pairs: None for a `**` spread, "?" for a key that is not a literal."""
|
|
if isinstance(d, ast.Dict):
|
|
return [
|
|
(
|
|
k.value
|
|
if isinstance(k, ast.Constant) and isinstance(k.value, str)
|
|
else None
|
|
if k is None
|
|
else "?",
|
|
v,
|
|
)
|
|
for k, v in zip(d.keys, d.values, strict=True)
|
|
]
|
|
assert isinstance(d, ast.Call)
|
|
return [(None, a) for a in d.args] + [(k.arg, k.value) for k in d.keywords]
|
|
|
|
def site(self, m: Module, d: Node, out: dict) -> None:
|
|
items = self.entries(d)
|
|
code = Res()
|
|
for v in (v for k, v in items if k == "code"):
|
|
code = code | self.resolve(m, v)
|
|
problems = [
|
|
f"{self.at(m, v)} (a key the scan cannot read)" for k, v in items if k == "?"
|
|
] + [
|
|
f"{self.at(m, v)} (a spread that may carry the code)"
|
|
for k, v in items
|
|
if k is None and not self.clean_spread(m, d, v)
|
|
]
|
|
res, has_reason, changed = self.mutations(m, d)
|
|
problems += changed
|
|
if code.unread or code.none or len(code.values) != 1 or problems:
|
|
out["unreadCodes"] += problems or [self.at(m, d)]
|
|
return
|
|
entry = out["codes"].setdefault(
|
|
next(iter(code.values)), {"reasons": set(), "unread": [], "sites": 0, "reasonless": []}
|
|
)
|
|
entry["sites"] += 1
|
|
for k, v in items:
|
|
if k == "reason":
|
|
res, has_reason = res | self.resolve(m, v), True
|
|
if not has_reason or res.none:
|
|
entry["reasonless"].append(f"{m.path}:{d.lineno}")
|
|
entry["reasons"] |= res.values
|
|
entry["unread"] += res.unread
|
|
|
|
def mutations(self, m: Module, d: Node) -> tuple[Res, bool, list[str]]:
|
|
"""A detail put in a variable: what `|= {…}` and `["reason"] = …` add, and every other use that
|
|
could change it or send it elsewhere."""
|
|
holder, stmt = d, m.parents.get(d)
|
|
while isinstance(stmt, ast.IfExp) and holder is not stmt.test:
|
|
holder, stmt = stmt, m.parents.get(stmt)
|
|
if not (isinstance(stmt, (ast.Assign, ast.AnnAssign)) and stmt.value is holder):
|
|
return Res(), False, []
|
|
targets = stmt.targets if isinstance(stmt, ast.Assign) else [stmt.target]
|
|
fn = self.enclosing(m, d, FUNCS)
|
|
if fn is None or len(targets) != 1 or not isinstance(targets[0], ast.Name):
|
|
return (
|
|
Res(),
|
|
False,
|
|
[f"{self.at(m, stmt)} (a detail put where the scan cannot follow it)"],
|
|
)
|
|
name = targets[0].id
|
|
res, found, problems = Res(), False, []
|
|
for n in _own_scope(fn):
|
|
if n is stmt:
|
|
continue
|
|
if name in _binds(n):
|
|
if not (isinstance(n, ast.AugAssign) and isinstance(n.op, ast.BitOr)):
|
|
problems.append(f"{self.at(m, n)} (binds the detail's variable again)")
|
|
elif isinstance(n.value, ast.Dict):
|
|
for k, v in self.entries(n.value):
|
|
if k == "reason":
|
|
res, found = res | self.resolve(m, v), True
|
|
elif k in ("code", "?") or k is None and not self.clean_spread(m, n, v):
|
|
problems.append(f"{self.at(m, v)} (may change the code)")
|
|
elif not self.clean_spread(m, n, n.value):
|
|
problems.append(f"{self.at(m, n)} (may change the code)")
|
|
continue
|
|
if not (isinstance(n, ast.Name) and n.id == name and isinstance(n.ctx, ast.Load)):
|
|
continue
|
|
p = m.parents.get(n)
|
|
if isinstance(p, ast.AugAssign) and p.target is n:
|
|
continue
|
|
if isinstance(p, ast.Subscript) and p.value is n:
|
|
key = p.slice.value if isinstance(p.slice, ast.Constant) else None
|
|
assign = m.parents.get(p)
|
|
if isinstance(p.ctx, ast.Load):
|
|
continue
|
|
if (
|
|
key == "reason"
|
|
and isinstance(p.ctx, ast.Store)
|
|
and isinstance(assign, ast.Assign)
|
|
and assign.targets == [p]
|
|
):
|
|
res, found = res | self.resolve(m, assign.value), True
|
|
elif not (
|
|
isinstance(key, str)
|
|
and key not in ("code", "reason")
|
|
and isinstance(p.ctx, ast.Store)
|
|
):
|
|
problems.append(f"{self.at(m, p)} (may change the code or drop the reason)")
|
|
continue
|
|
if isinstance(p, ast.Attribute) and p.attr in READS:
|
|
continue
|
|
sender = (
|
|
p
|
|
if isinstance(p, ast.Call)
|
|
else m.parents.get(p)
|
|
if isinstance(p, ast.keyword)
|
|
else None
|
|
)
|
|
func = getattr(sender, "func", None)
|
|
if (
|
|
isinstance(sender, ast.Call)
|
|
and (getattr(func, "id", None) or getattr(func, "attr", None)) in SENDERS
|
|
and (n in sender.args or isinstance(p, ast.keyword) and p.arg == "detail")
|
|
):
|
|
continue
|
|
problems.append(f"{self.at(m, n)} (the detail goes where the scan cannot follow it)")
|
|
return res, found, problems
|
|
|
|
def run(self) -> dict:
|
|
out: dict = {
|
|
"codes": {},
|
|
"unreadCodes": [],
|
|
"errors": self.errors,
|
|
"excluded": {},
|
|
"aliases": {},
|
|
"notRefusals": sorted(NOT_REFUSALS),
|
|
}
|
|
for m in self.mods.values():
|
|
for name, stmt in m.top.items():
|
|
if (
|
|
isinstance(stmt, ast.Assign)
|
|
and isinstance(stmt.value, ast.Subscript)
|
|
and (v := self.ann_values(m, stmt.value))
|
|
):
|
|
out["aliases"].setdefault(m.path, {})[name] = sorted(v)
|
|
for node in ast.walk(m.tree):
|
|
is_dict = isinstance(node, ast.Dict) and any(
|
|
isinstance(k, ast.Constant) and k.value == "code" for k in node.keys
|
|
)
|
|
is_call = (
|
|
isinstance(node, ast.Call)
|
|
and getattr(node.func, "id", None) == "dict"
|
|
and any(k.arg == "code" for k in node.keywords)
|
|
)
|
|
if not (is_dict or is_call):
|
|
continue
|
|
if m.path in NOT_REFUSALS:
|
|
out["excluded"][m.path] = out["excluded"].get(m.path, 0) + 1
|
|
continue
|
|
self.site(m, node, out)
|
|
for entry in out["codes"].values():
|
|
entry["reasons"] = sorted(entry["reasons"])
|
|
return out
|
|
|
|
|
|
def scan(sources: dict[str, str]) -> dict:
|
|
return Scan(sources).run()
|
|
|
|
|
|
def scan_app(app: pathlib.Path = APP) -> dict:
|
|
return scan(
|
|
{
|
|
p.relative_to(app).as_posix(): p.read_text(encoding="utf-8")
|
|
for p in sorted(app.rglob("*.py"))
|
|
if "__pycache__" not in p.parts
|
|
}
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
json.dump(
|
|
scan_app(pathlib.Path(sys.argv[1]) if len(sys.argv) > 1 else APP),
|
|
sys.stdout,
|
|
indent=1,
|
|
sort_keys=True,
|
|
)
|