peter
31a75f4985
A refusal's detail now stands where it is sent (HTTPException's detail, a JSONResponse's "detail") or is returned by a factory that only ever is, with any spread before its code and reason, so no spread can replace them. The scan follows no variable and no kwargs any more; what it cannot read in a reason (a lambda's or comprehension's name, a nonlocal or global, a local that reaches itself, a constant or map changed elsewhere, an app import it cannot resolve) is reported. The detail guard reads the same slots and fails on anything but a code dict or a factory's call.
739 lines
33 KiB
Python
739 lines
33 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 scan that tries to follow every shape opens a silent gap with each one
|
|
it learns, and following how a value flows through Python (variables, scopes, aliases, a dict changed in
|
|
place) never closes. So the CODE keeps to a narrow form, and this reads a CLOSED list of it, reporting
|
|
everything else, never guessing:
|
|
|
|
- where a detail stands: where it is sent — `HTTPException`'s second argument or `detail=`, the `"detail"`
|
|
value of a `JSONResponse`'s content dict — through the arms of `a if c else b`, or as what a module-level
|
|
factory returns when every return of it is a detail and every reference to it is a call sent as one. A
|
|
detail held in a variable is not followed;
|
|
- the code: one literal. A `**` spread, or a key that is not a literal, stands BEFORE a literal `code` and
|
|
`reason`, both of which the detail then carries: the later key wins, so no spread can change them;
|
|
- a reason: a string literal (`None` sends none: the site counts as reasonless); both arms of
|
|
`a if c else b`; a function's local 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.
|
|
|
|
A name bound twice at module level, inside a module-level `if`/`try`/loop, named by a `global`, or imported
|
|
from two places is not looked up; nor is a local a lambda or a comprehension binds, or one a `nonlocal` or
|
|
`global` names. A module constant or map that anything stores into, deletes from or (a map) uses but by
|
|
`map[k]` or a read method is unread. A reference named like a followed helper that the scan cannot resolve
|
|
(`import app.x`, a relative import, an app module that does not bind it) 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 map may be used by besides `map[k]`: a method that only reads it.
|
|
READS = {"get", "items", "keys", "values"}
|
|
|
|
Node = ast.AST
|
|
FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef)
|
|
SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)
|
|
COMPREHENSIONS = (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)
|
|
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, by a `global`, or imported from two places
|
|
slots: set[int] # the ids of the nodes sent as a detail
|
|
|
|
|
|
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 _sender(call: ast.Call) -> str | None:
|
|
return getattr(call.func, "id", None) or getattr(call.func, "attr", None)
|
|
|
|
|
|
def detail_slots(tree: Node) -> Iterator[tuple[ast.Call, Node | None]]:
|
|
"""Every place a detail is sent from, with what it holds: `HTTPException`'s second argument or
|
|
`detail=` (None: sent without one), and the `"detail"` value of a `JSONResponse`'s content dict (a
|
|
content without one is an answer, not a refusal)."""
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
kw = {k.arg: k.value for k in node.keywords}
|
|
if _sender(node) == "HTTPException":
|
|
yield node, kw.get("detail", node.args[1] if len(node.args) > 1 else None)
|
|
elif _sender(node) == "JSONResponse":
|
|
content = kw.get("content", node.args[0] if node.args else None)
|
|
if isinstance(content, ast.Dict):
|
|
for k, v in zip(content.keys, content.values, strict=True):
|
|
if isinstance(k, ast.Constant) and k.value == "detail":
|
|
yield node, v
|
|
|
|
|
|
def _code_dict(node: Node) -> bool:
|
|
"""A `{"code": …}` dict, or a `dict(code=…)` call."""
|
|
if isinstance(node, ast.Dict):
|
|
return any(isinstance(k, ast.Constant) and k.value == "code" for k in node.keys)
|
|
return (
|
|
isinstance(node, ast.Call)
|
|
and getattr(node.func, "id", None) == "dict"
|
|
and any(k.arg == "code" for k in node.keywords)
|
|
)
|
|
|
|
|
|
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]]] = {}
|
|
ambiguous: set[str] = set()
|
|
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))
|
|
elif isinstance(node, ast.Global):
|
|
ambiguous |= set(node.names)
|
|
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}
|
|
slots = {id(d) for _, d in detail_slots(tree) if d is not None}
|
|
name = _module_name(path)
|
|
mods[name] = Module(path, name, tree, parents, imports, top, ambiguous, slots)
|
|
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._resolving: set[int] = set()
|
|
self._factory: dict[int, bool] = {}
|
|
self._changed: 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.binder(m, e) is not None 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 binder(self, m: Module, e: ast.Name) -> Node | None:
|
|
"""The innermost scope `e` is bound in: a function (a parameter, or a binding in its own body), a
|
|
lambda (a parameter), a comprehension (its target); None: a module-level name."""
|
|
node = m.parents.get(e)
|
|
while node is not None:
|
|
if isinstance(node, (*FUNCS, ast.Lambda)):
|
|
a = node.args
|
|
params = [*a.posonlyargs, *a.args, *a.kwonlyargs, a.vararg, a.kwarg]
|
|
if e.id in {p.arg for p in params if p}:
|
|
return node
|
|
if isinstance(node, FUNCS) and self.assignments(node, e.id):
|
|
return node
|
|
elif isinstance(node, COMPREHENSIONS) and any(
|
|
e.id in _stored(g.target) for g in node.generators
|
|
):
|
|
return node
|
|
node = m.parents.get(node)
|
|
return None
|
|
|
|
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.binder(m, root) is not None:
|
|
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.id in m.top:
|
|
return False
|
|
if root is node: # imported from an app module that does not bind it where the scan can see
|
|
mod = m.imports[root.id][0]
|
|
return (mod == "app" or mod.startswith("app.")) and self.lookup(m, root.id) is None
|
|
found = self.lookup(m, root.id)
|
|
return (
|
|
found is not None and found[1] is None
|
|
) # an app module whose dotted path went nowhere
|
|
|
|
def changed(self, m: Module, stmt: Node, name: str, *, only_read: bool) -> bool:
|
|
"""Something in the app stores into, deletes or rebinds the module-level `name` that `stmt` binds,
|
|
or (`only_read`, a map) uses it but by `map[k]` or a method that only reads it."""
|
|
key = id(stmt) * 2 + only_read
|
|
if key in self._changed:
|
|
return self._changed[key]
|
|
own = stmt.targets if isinstance(stmt, ast.Assign) else [getattr(stmt, "target", None)]
|
|
out = False
|
|
for cm in self.mods.values():
|
|
for node in ast.walk(cm.tree):
|
|
if out:
|
|
break
|
|
if not (
|
|
isinstance(node, ast.Name)
|
|
and node.id == name
|
|
or isinstance(node, ast.Attribute)
|
|
and node.attr == name
|
|
) or (cm is m and node in own):
|
|
continue
|
|
found = self.lookup_expr(cm, node)
|
|
if not (found and found[0] is m and found[1] is stmt):
|
|
continue
|
|
if not isinstance(node.ctx, ast.Load):
|
|
out = True
|
|
elif only_read:
|
|
p = cm.parents.get(node)
|
|
out = not (
|
|
isinstance(p, ast.Subscript)
|
|
and p.value is node
|
|
and isinstance(p.ctx, ast.Load)
|
|
or isinstance(p, ast.Attribute)
|
|
and p.attr in READS
|
|
)
|
|
self._changed[key] = out
|
|
return out
|
|
|
|
# --- 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:
|
|
if self.caught_as(m, e, e.id) is not None:
|
|
return Res(unread=[self.at(m, e)]) # the exception itself, not a reason
|
|
fn = self.binder(m, e)
|
|
if fn is None:
|
|
return Res(unread=[self.at(m, e)])
|
|
if not isinstance(fn, FUNCS):
|
|
return Res(unread=[f"{self.at(m, e)} (bound by a lambda or a comprehension)"])
|
|
if any(isinstance(n, (ast.Nonlocal, ast.Global)) and e.id in n.names for n in ast.walk(fn)):
|
|
return Res(unread=[f"{self.at(m, e)} (named by a nonlocal or a global)"])
|
|
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)
|
|
out = Res(unread=[] if assigned else [self.at(m, e)]) # `*args`, `**kwargs`
|
|
for s in assigned:
|
|
if not (isinstance(s, ast.Assign) and all(isinstance(t, ast.Name) for t in s.targets)):
|
|
out = out | Res(unread=[self.at(m, e)])
|
|
elif id(s) in self._resolving:
|
|
out = out | Res(unread=[f"{self.at(m, e)} (reached again while it is resolved)"])
|
|
else:
|
|
self._resolving.add(id(s))
|
|
out = out | self.resolve(m, s.value)
|
|
self._resolving.discard(id(s))
|
|
return out
|
|
|
|
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))
|
|
and isinstance(found[1].value, ast.Constant)
|
|
):
|
|
if self.changed(found[0], found[1], e.attr, only_read=False):
|
|
return Res(unread=[f"{self.at(m, e)} (bound again or deleted elsewhere)"])
|
|
return self.resolve(found[0], found[1].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)])
|
|
name = e.value.attr if isinstance(e.value, ast.Attribute) else getattr(e.value, "id", "")
|
|
if self.changed(found[0], found[1], name, only_read=True):
|
|
return Res(unread=[f"{self.at(m, e)} (the map is changed after it is built)"])
|
|
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
|
|
|
|
# --- sites ---------------------------------------------------------------------------------------
|
|
|
|
def is_detail(self, node: Node | None) -> bool:
|
|
if isinstance(node, ast.IfExp):
|
|
return self.is_detail(node.body) and self.is_detail(node.orelse)
|
|
return _code_dict(node)
|
|
|
|
def factory(self, m: Module, fn: Node) -> bool:
|
|
"""A module-level function each return of which is a detail, and each reference to which is a
|
|
call sent as one (`quota_detail`)."""
|
|
assert isinstance(fn, FUNCS)
|
|
if id(fn) in self._factory:
|
|
return self._factory[id(fn)]
|
|
returns = [n for n in _own_scope(fn) if isinstance(n, ast.Return)]
|
|
ok = (
|
|
m.top.get(fn.name) is fn # a name bound twice: its calls are unresolved below
|
|
and bool(returns)
|
|
and all(self.is_detail(r.value) for r in returns)
|
|
)
|
|
for cm, _, call, resolved in self.references(m, fn) if ok else []:
|
|
ok = resolved and id(call) in cm.slots # a bare reference is no call: never sent
|
|
if not ok:
|
|
break
|
|
self._factory[id(fn)] = ok
|
|
return ok
|
|
|
|
def placed(self, m: Module, d: Node) -> bool:
|
|
"""`d` is sent as a detail: from a sender's slot, through the arms of `a if c else b`, or as what a
|
|
factory returns."""
|
|
holder, p = d, m.parents.get(d)
|
|
while isinstance(p, ast.IfExp) and holder is not p.test:
|
|
holder, p = p, m.parents.get(p)
|
|
if id(holder) in m.slots:
|
|
return True
|
|
fn = self.enclosing(m, p, FUNCS) if isinstance(p, ast.Return) else None
|
|
return fn is not None and self.factory(m, fn)
|
|
|
|
def sayable(self, m: Module, d: Node | None) -> bool:
|
|
"""What a slot holds is a detail the client can say: a code dict, or a factory's call."""
|
|
if isinstance(d, ast.IfExp):
|
|
return self.sayable(m, d.body) and self.sayable(m, d.orelse)
|
|
if _code_dict(d):
|
|
return True
|
|
found = self.lookup_expr(m, d.func) if isinstance(d, ast.Call) else None
|
|
return bool(
|
|
found
|
|
and isinstance(found[0], Module)
|
|
and isinstance(found[1], FUNCS)
|
|
and self.factory(found[0], found[1])
|
|
)
|
|
|
|
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 spread in dict())"
|
|
for k, v in items
|
|
if k is None and isinstance(d, ast.Call)
|
|
] + [
|
|
f"{self.at(m, v)} (a spread or a computed key the code and the reason do not follow)"
|
|
for i, (k, v) in enumerate(items)
|
|
if k in (None, "?") and not {"code", "reason"} <= {k2 for k2, _ in items[i + 1 :]}
|
|
]
|
|
if not self.placed(m, d):
|
|
problems.append(f"{self.at(m, d)} (a detail put where the scan cannot follow it)")
|
|
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
|
|
res, has_reason = Res(), False
|
|
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 other_details(self) -> list[tuple[str, int, str | None]]:
|
|
"""Each slot that sends what the client cannot say — nothing, or anything but a code dict or a
|
|
factory's call: (its file, its line, the function it is sent from)."""
|
|
out = []
|
|
for m in self.mods.values():
|
|
for call, d in detail_slots(m.tree):
|
|
if self.sayable(m, d):
|
|
continue
|
|
fn = self.enclosing(m, call, FUNCS)
|
|
out.append((m.path, call.lineno, getattr(fn, "name", None)))
|
|
return sorted(out)
|
|
|
|
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):
|
|
if not _code_dict(node):
|
|
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 app_sources(app: pathlib.Path = APP) -> dict[str, str]:
|
|
return {
|
|
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
|
|
}
|
|
|
|
|
|
def scan_app(app: pathlib.Path = APP) -> dict:
|
|
return scan(app_sources(app))
|
|
|
|
|
|
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,
|
|
)
|