fix(i18n): read every refusal from the backend's syntax tree, not by regex
The contract read Python with regexes, and each guard added for a missed shape had shapes of its own
it missed: a reason after another key, a status passed by name, a second `e.reason` handler with the
same text, a pin naming a type nothing checked.
backend/tests/refusal_sites.py (stdlib `ast`, run by the contract on the host's Python) reads every
`{"code": ...}` detail and follows each reason to its values: literals, both arms of a conditional,
module constants, the `Literal` on the parameter or attribute it travels in (on the class an `except`
catches), every call of a `str` helper with its defaults, a map's values with its keys held to the
index's type. What it cannot follow fails the contract, unless no screen shows that code; no pins.
The reason attributes are typed with their Literal, so a subclass or an assignment cannot widen
them; InvalidSetting takes a SettingReason, and the settings map is held to it (a reason the map
lacked was a 500). The per-family reverse checks are one check by prefix (the admin actions
included), and the leaf-key filter applies only to the suffixed sign-in family.
This commit is contained in:
@@ -0,0 +1,625 @@
|
||||
"""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: a
|
||||
reason after another key, a status passed by name, a second handler with the same text all went past it.
|
||||
Here a reason is followed to where it comes from, and what cannot be followed is reported, never guessed:
|
||||
|
||||
- a literal; both arms of `a if c else b`, every operand of `a or b`;
|
||||
- a module constant (`urlguard.REASON_MALFORMED`), a local variable's assignments;
|
||||
- a parameter or attribute typed with a `Literal` (basedpyright holds every value passed in), the
|
||||
attribute found on the class an `except … as e` catches, or the class a parameter is typed with;
|
||||
- a `str` parameter of a helper (`refused`, `not_found`): followed to every call of it, its default
|
||||
included, and a helper passed around as a value is unread;
|
||||
- a call: its return annotation; a map lookup (`_SETTING_REASONS[exc.reason]`): the map's values, and
|
||||
its keys must be exactly what the index can be.
|
||||
|
||||
Anything else is UNREAD, 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 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",
|
||||
}
|
||||
|
||||
Node = ast.AST
|
||||
|
||||
|
||||
@dataclass
|
||||
class Res:
|
||||
values: set[str] = field(default_factory=set)
|
||||
unread: list[str] = field(default_factory=list)
|
||||
|
||||
def __or__(self, other: "Res") -> "Res":
|
||||
return Res(self.values | other.values, self.unread + other.unread)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Module:
|
||||
path: str # relative to app/, posix
|
||||
name: str # dotted, `app.…`
|
||||
tree: ast.Module
|
||||
parents: dict[Node, Node]
|
||||
imports: dict[
|
||||
str, tuple[str, str | None]
|
||||
] # local name → (module, symbol or None for the module)
|
||||
top: dict[str, Node] # module-level bindings: def, class, or the assigned value
|
||||
|
||||
|
||||
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)}
|
||||
imports: dict[str, 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:
|
||||
imports[a.asname or a.name] = (node.module, a.name)
|
||||
elif isinstance(node, ast.Import):
|
||||
for a in node.names:
|
||||
if a.asname:
|
||||
imports[a.asname] = (a.name, None)
|
||||
top: dict[str, Node] = {}
|
||||
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
|
||||
name = _module_name(path)
|
||||
mods[name] = Module(path, name, tree, parents, imports, top)
|
||||
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()
|
||||
|
||||
# --- names ---------------------------------------------------------------------------------------
|
||||
|
||||
def lookup(self, m: Module, name: str) -> 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.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:
|
||||
return self.lookup(self.mods[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 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 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
|
||||
|
||||
# --- types ---------------------------------------------------------------------------------------
|
||||
|
||||
def ann_values(self, m: Module, ann: Node | None) -> set[str] | None:
|
||||
"""The values a `Literal` annotation allows, or None when it is not one (`str`, a class, …)."""
|
||||
if ann is None:
|
||||
return None
|
||||
if isinstance(ann, ast.Constant):
|
||||
if ann.value is None:
|
||||
return set()
|
||||
if isinstance(ann.value, str):
|
||||
return self.ann_values(m, ast.parse(ann.value, mode="eval").body)
|
||||
return None
|
||||
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.BinOp) and isinstance(ann.op, ast.BitOr):
|
||||
left, right = self.ann_values(m, ann.left), self.ann_values(m, ann.right)
|
||||
return None if left is None or right is None else left | right
|
||||
found = self.lookup_expr(m, ann)
|
||||
if found and isinstance(found[1], ast.Assign) and isinstance(found[0], Module):
|
||||
if isinstance(found[1].value, (ast.Subscript, ast.BinOp, ast.Name, ast.Attribute)):
|
||||
return self.ann_values(found[0], found[1].value)
|
||||
return None
|
||||
|
||||
def classes(self, m: Module, e: Node | None) -> list[tuple[Module, ast.ClassDef]] | None:
|
||||
if e is None:
|
||||
return None
|
||||
if isinstance(e, ast.Tuple):
|
||||
out = [self.classes(m, x) for x in e.elts]
|
||||
return None if any(o is None for o in out) else [c for o in out for c in o]
|
||||
found = self.lookup_expr(m, e)
|
||||
if found and isinstance(found[0], Module) and isinstance(found[1], ast.ClassDef):
|
||||
return [(found[0], found[1])]
|
||||
return None
|
||||
|
||||
def attr_values(self, cm: Module, cls: ast.ClassDef, attr: str) -> set[str] | None:
|
||||
"""An attribute's `Literal` values, from its annotation on the class or `self.attr: T` in a method,
|
||||
else from a base class."""
|
||||
for node in ast.walk(cls):
|
||||
if isinstance(node, ast.AnnAssign):
|
||||
t = node.target
|
||||
if (isinstance(t, ast.Name) and t.id == attr and node in cls.body) or (
|
||||
isinstance(t, ast.Attribute)
|
||||
and t.attr == attr
|
||||
and getattr(t.value, "id", None) == "self"
|
||||
):
|
||||
return self.ann_values(cm, node.annotation)
|
||||
for base in cls.bases:
|
||||
for bm, bc in self.classes(cm, base) or []:
|
||||
v = self.attr_values(bm, bc, attr)
|
||||
if v is not None:
|
||||
return v
|
||||
return None
|
||||
|
||||
# --- reasons -------------------------------------------------------------------------------------
|
||||
|
||||
def resolve(self, m: Module, e: Node) -> Res:
|
||||
if isinstance(e, ast.Constant):
|
||||
if e.value is None:
|
||||
return Res()
|
||||
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.BoolOp) and isinstance(e.op, ast.Or):
|
||||
out = Res()
|
||||
for v in e.values:
|
||||
out = out | self.resolve(m, v)
|
||||
return out
|
||||
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], (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
):
|
||||
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, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
while fn is not None:
|
||||
assert isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
if self.caught_as(m, e, e.id) is not None:
|
||||
return Res(unread=[self.at(m, e)]) # the exception itself, not a reason
|
||||
params = {a.arg: a for a in [*fn.args.posonlyargs, *fn.args.args, *fn.args.kwonlyargs]}
|
||||
if e.id in params:
|
||||
return self.resolve_param(m, fn, params[e.id], e)
|
||||
if (
|
||||
fn.args.vararg
|
||||
and fn.args.vararg.arg == e.id
|
||||
or fn.args.kwarg
|
||||
and fn.args.kwarg.arg == e.id
|
||||
):
|
||||
return Res(unread=[self.at(m, e)])
|
||||
assigned = self.assignments(fn, e.id)
|
||||
if assigned:
|
||||
out = Res()
|
||||
for a in assigned:
|
||||
if (
|
||||
isinstance(a, ast.AnnAssign)
|
||||
and (v := self.ann_values(m, a.annotation)) is not None
|
||||
):
|
||||
out = out | Res(v)
|
||||
elif (
|
||||
isinstance(a, (ast.Assign, ast.AnnAssign))
|
||||
and a.value is not None
|
||||
and not any(
|
||||
isinstance(t, (ast.Tuple, ast.List)) for t in getattr(a, "targets", [])
|
||||
)
|
||||
):
|
||||
out = out | self.resolve(m, a.value)
|
||||
else:
|
||||
out = out | Res(unread=[self.at(m, e)])
|
||||
return out
|
||||
fn = self.enclosing(m, fn, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
found = self.lookup(m, e.id)
|
||||
if (
|
||||
found
|
||||
and isinstance(found[0], Module)
|
||||
and isinstance(found[1], (ast.Assign, ast.AnnAssign))
|
||||
):
|
||||
stmt = found[1]
|
||||
if stmt.value is not None and isinstance(stmt.value, ast.Constant):
|
||||
return self.resolve(found[0], stmt.value)
|
||||
return Res(unread=[self.at(m, e)])
|
||||
|
||||
def assignments(self, fn: Node, name: str) -> list[Node]:
|
||||
"""The statements that bind `name` in this function's own body (not in a nested one)."""
|
||||
out: list[Node] = []
|
||||
stack = list(ast.iter_child_nodes(fn))
|
||||
while stack:
|
||||
node = stack.pop()
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)):
|
||||
continue
|
||||
if isinstance(node, ast.Assign):
|
||||
if any(
|
||||
name in {n.id for n in ast.walk(t) if isinstance(n, ast.Name)}
|
||||
for t in node.targets
|
||||
):
|
||||
out.append(node)
|
||||
elif isinstance(node, ast.AnnAssign) and getattr(node.target, "id", None) == name:
|
||||
out.append(node)
|
||||
elif isinstance(
|
||||
node, (ast.For, ast.With, ast.AugAssign, ast.NamedExpr, ast.comprehension)
|
||||
):
|
||||
targets = [getattr(node, "target", None)] + [
|
||||
i.optional_vars
|
||||
for i in getattr(node, "items", [])
|
||||
if i.optional_vars is not None
|
||||
]
|
||||
if any(
|
||||
t is not None and name in {n.id for n in ast.walk(t) if isinstance(n, ast.Name)}
|
||||
for t in targets
|
||||
):
|
||||
out.append(
|
||||
node
|
||||
) # bound by a loop, a `with`, an augmented or a walrus: not followed
|
||||
stack.extend(ast.iter_child_nodes(node))
|
||||
return out
|
||||
|
||||
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, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
if isinstance(node, ast.ExceptHandler) and node.name == name:
|
||||
return node
|
||||
node = m.parents.get(node)
|
||||
return None
|
||||
|
||||
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, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
key = (m.name, fn.name, param.arg)
|
||||
if key in self._following:
|
||||
return Res()
|
||||
if m.top.get(fn.name) is not fn:
|
||||
return Res(
|
||||
unread=[
|
||||
f"{m.path}:{fn.lineno}: {fn.name}({param.arg}) is not a module-level helper"
|
||||
]
|
||||
)
|
||||
self._following.add(key)
|
||||
positional = [a.arg for a in [*fn.args.posonlyargs, *fn.args.args]]
|
||||
defaults = dict(
|
||||
zip(
|
||||
positional[len(positional) - len(fn.args.defaults) :], fn.args.defaults, strict=True
|
||||
)
|
||||
)
|
||||
defaults |= {
|
||||
a.arg: d
|
||||
for a, d in zip(fn.args.kwonlyargs, fn.args.kw_defaults, strict=True)
|
||||
if d is not None
|
||||
}
|
||||
out = Res()
|
||||
for cm in self.mods.values():
|
||||
for node in ast.walk(cm.tree):
|
||||
if not isinstance(node, (ast.Name, ast.Attribute)) or isinstance(
|
||||
cm.parents.get(node), ast.Attribute
|
||||
):
|
||||
continue
|
||||
found = self.lookup_expr(cm, node)
|
||||
if not (found and found[0] is m and found[1] is fn):
|
||||
continue
|
||||
call = cm.parents.get(node)
|
||||
if not (isinstance(call, ast.Call) and call.func is node):
|
||||
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 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: # given by position, a `**` spread cannot pass it again (a TypeError)
|
||||
out = out | self.resolve(cm, call.args[positional.index(param.arg)])
|
||||
elif None in kw:
|
||||
out = out | Res(unread=[self.at(cm, call)])
|
||||
elif param.arg in defaults:
|
||||
out = out | self.resolve(m, defaults[param.arg])
|
||||
return out
|
||||
|
||||
def resolve_attr(self, m: Module, e: ast.Attribute) -> Res:
|
||||
base = e.value
|
||||
if isinstance(base, ast.Name):
|
||||
handler = self.caught_as(m, e, base.id)
|
||||
fn = self.enclosing(m, e, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
typed = None
|
||||
if handler is not None:
|
||||
typed = self.classes(m, handler.type)
|
||||
elif fn is not None:
|
||||
params = {
|
||||
a.arg: a for a in [*fn.args.posonlyargs, *fn.args.args, *fn.args.kwonlyargs]
|
||||
}
|
||||
if base.id in params:
|
||||
typed = self.classes(m, params[base.id].annotation)
|
||||
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 index.unread 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 spread_is_safe(self, m: Module, e: Node, src: Node) -> bool:
|
||||
"""A `**x` in a detail, or `detail |= {…}` from it, cannot carry a reason when `x` is the `**kwargs`
|
||||
of a function that takes `reason` by name: Python binds that one to the parameter."""
|
||||
if isinstance(src, ast.Attribute):
|
||||
return self.spread_is_safe_attr(m, e, src)
|
||||
fn = self.enclosing(m, e, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
if fn is None or fn.args.kwarg is None:
|
||||
return False
|
||||
names = {a.arg for a in [*fn.args.posonlyargs, *fn.args.args, *fn.args.kwonlyargs]}
|
||||
kwarg = fn.args.kwarg.arg
|
||||
if isinstance(src, ast.Name):
|
||||
return src.id == kwarg and "reason" in names
|
||||
|
||||
if isinstance(src, ast.DictComp):
|
||||
it = src.generators[0].iter
|
||||
return (
|
||||
isinstance(it, ast.Call)
|
||||
and isinstance(it.func, ast.Attribute)
|
||||
and it.func.attr == "items"
|
||||
and getattr(it.func.value, "id", None) == kwarg
|
||||
and "reason" in names
|
||||
)
|
||||
return False
|
||||
|
||||
def spread_is_safe_attr(self, m: Module, e: Node, src: ast.Attribute) -> bool:
|
||||
"""`e.params` of a caught exception whose `__init__(self, reason, **params)` keeps `self.params = params`."""
|
||||
if not isinstance(src.value, ast.Name):
|
||||
return False
|
||||
handler = self.caught_as(m, e, src.value.id)
|
||||
typed = self.classes(m, handler.type) if handler else None
|
||||
return bool(typed) and all(self.keeps_kwargs(cls, src.attr) for _, cls in typed or [])
|
||||
|
||||
def keeps_kwargs(self, cls: ast.ClassDef, attr: str) -> bool:
|
||||
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
|
||||
names = {a.arg for a in [*init.args.posonlyargs, *init.args.args, *init.args.kwonlyargs]}
|
||||
kwarg = init.args.kwarg.arg
|
||||
return "reason" in names and any(
|
||||
isinstance(n, ast.Assign)
|
||||
and any(
|
||||
isinstance(t, ast.Attribute)
|
||||
and t.attr == attr
|
||||
and getattr(t.value, "id", None) == "self"
|
||||
for t in n.targets
|
||||
)
|
||||
and getattr(n.value, "id", None) == kwarg
|
||||
for n in ast.walk(init)
|
||||
)
|
||||
|
||||
def entries(self, d: Node) -> list[tuple[str | None, Node]]:
|
||||
if isinstance(d, ast.Dict):
|
||||
return [
|
||||
(k.value if isinstance(k, ast.Constant) 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 [(k.arg, k.value) for k in d.keywords]
|
||||
|
||||
def site(self, m: Module, d: Node, out: dict) -> None:
|
||||
items = self.entries(d)
|
||||
code_expr = next(v for k, v in items if k == "code")
|
||||
code = self.resolve(m, code_expr)
|
||||
if code.unread or len(code.values) != 1:
|
||||
out["unreadCodes"].append(self.at(m, code_expr))
|
||||
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
|
||||
elif k is None and not self.spread_is_safe(m, d, v):
|
||||
res = res | Res(unread=[f"{self.at(m, v)} (a spread that may carry a reason)"])
|
||||
res, more = self.mutations(m, d, res)
|
||||
has_reason = has_reason or more
|
||||
if not has_reason:
|
||||
entry["reasonless"].append(f"{m.path}:{d.lineno}")
|
||||
entry["reasons"] |= res.values
|
||||
entry["unread"] += res.unread
|
||||
|
||||
def mutations(self, m: Module, d: Node, res: Res) -> tuple[Res, bool]:
|
||||
"""`detail |= {…}`, `detail["reason"] = …`, `detail.update(…)` on the variable a detail is put in."""
|
||||
stmt = m.parents.get(d)
|
||||
if not (isinstance(stmt, (ast.Assign, ast.AnnAssign)) and stmt.value is d):
|
||||
return res, False
|
||||
target = stmt.targets[0] if isinstance(stmt, ast.Assign) else stmt.target
|
||||
fn = self.enclosing(m, d, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
if not isinstance(target, ast.Name) or fn is None:
|
||||
return res, False
|
||||
found = False
|
||||
for node in ast.walk(fn):
|
||||
if isinstance(node, ast.AugAssign) and getattr(node.target, "id", None) == target.id:
|
||||
v = node.value
|
||||
if isinstance(v, ast.Dict):
|
||||
for k, val in self.entries(v):
|
||||
if k == "reason":
|
||||
res, found = res | self.resolve(m, val), True
|
||||
elif k is None and not self.spread_is_safe(m, node, val):
|
||||
res = res | Res(
|
||||
unread=[f"{self.at(m, val)} (a spread that may carry a reason)"]
|
||||
)
|
||||
elif not self.spread_is_safe(m, node, v):
|
||||
res = res | Res(unread=[f"{self.at(m, node)} (may carry a reason)"])
|
||||
elif (
|
||||
isinstance(node, ast.Subscript)
|
||||
and getattr(node.value, "id", None) == target.id
|
||||
and isinstance(node.ctx, ast.Store)
|
||||
):
|
||||
if isinstance(node.slice, ast.Constant) and node.slice.value != "reason":
|
||||
continue
|
||||
assign = m.parents.get(node)
|
||||
if isinstance(node.slice, ast.Constant) and isinstance(assign, ast.Assign):
|
||||
res, found = res | self.resolve(m, assign.value), True
|
||||
else:
|
||||
res = res | Res(unread=[f"{self.at(m, node)} (may carry a reason)"])
|
||||
elif (
|
||||
isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Attribute)
|
||||
and getattr(node.func.value, "id", None) == target.id
|
||||
and node.func.attr in {"update", "setdefault"}
|
||||
):
|
||||
res = res | Res(unread=[f"{self.at(m, node)} (may carry a reason)"])
|
||||
return res, found
|
||||
|
||||
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,
|
||||
)
|
||||
@@ -0,0 +1,278 @@
|
||||
"""The refusal scan (`refusal_sites.py`) on small sources: every shape a reason travels in is followed to
|
||||
its values or reported unread. A scan that silently drops a shape is a guard green over nothing, so each
|
||||
shape that once went past the regex it replaced is here. Pure: no app import, no database."""
|
||||
|
||||
import textwrap
|
||||
|
||||
from tests.refusal_sites import scan, scan_app
|
||||
|
||||
|
||||
def _scan(**files: str) -> dict:
|
||||
return scan(
|
||||
{f"{name.replace('__', '/')}.py": textwrap.dedent(src) for name, src in files.items()}
|
||||
)
|
||||
|
||||
|
||||
def _code(result: dict, code: str) -> dict:
|
||||
return result["codes"][code]
|
||||
|
||||
|
||||
REFUSAL = """
|
||||
from fastapi import HTTPException
|
||||
|
||||
def refused(status: int, reason: str, **params: object) -> HTTPException:
|
||||
return HTTPException(status_code=status, detail={"code": "refused", "reason": reason, **params})
|
||||
"""
|
||||
|
||||
|
||||
def test_a_literal_wherever_it_sits_in_the_dict():
|
||||
r = _scan(
|
||||
a="""
|
||||
def f(x):
|
||||
raise HTTPException(409, {"code": "clip", "limit_s": 5, "reason": "too_long"})
|
||||
def g(x):
|
||||
raise HTTPException(409, {"reason": "before", "code": "clip"})
|
||||
def h(x):
|
||||
raise HTTPException(409, {
|
||||
"code": "clip", # a comment between
|
||||
"reason": "after_comment",
|
||||
})
|
||||
def i(x):
|
||||
raise HTTPException(409, detail=dict(code="clip", reason="by_call"))
|
||||
"""
|
||||
)
|
||||
assert _code(r, "clip")["reasons"] == ["after_comment", "before", "by_call", "too_long"]
|
||||
assert _code(r, "clip")["unread"] == []
|
||||
|
||||
|
||||
def test_every_arm_of_a_conditional_and_every_operand_of_or():
|
||||
r = _scan(
|
||||
a="""
|
||||
def f(x, e):
|
||||
raise HTTPException(409, {"code": "edit", "reason": "a" if x else "b"})
|
||||
def g(x):
|
||||
raise HTTPException(409, {"code": "edit", "reason": x.why or "c"})
|
||||
"""
|
||||
)
|
||||
assert _code(r, "edit")["reasons"] == ["a", "b", "c"]
|
||||
assert [u.split(": ", 1)[1] for u in _code(r, "edit")["unread"]] == ["x.why"]
|
||||
|
||||
|
||||
def test_a_helper_is_followed_to_every_call_however_it_is_written():
|
||||
r = _scan(
|
||||
refusal=REFUSAL,
|
||||
routes__a="""
|
||||
from http import HTTPStatus
|
||||
from app import refusal
|
||||
from app.refusal import refused
|
||||
|
||||
def f(status):
|
||||
raise refused(status, "by_name_status")
|
||||
def g():
|
||||
raise refused(HTTPStatus.CONFLICT, reason="by_keyword")
|
||||
def h():
|
||||
raise refusal.refused(409, "qualified")
|
||||
def i(x):
|
||||
raise refused(409, "a" if x else "second_arm")
|
||||
""",
|
||||
)
|
||||
assert _code(r, "refused")["reasons"] == [
|
||||
"a",
|
||||
"by_keyword",
|
||||
"by_name_status",
|
||||
"qualified",
|
||||
"second_arm",
|
||||
]
|
||||
assert _code(r, "refused")["unread"] == []
|
||||
|
||||
|
||||
def test_a_helper_passed_as_a_value_or_spread_into_is_unread():
|
||||
r = _scan(
|
||||
refusal=REFUSAL,
|
||||
routes__a="""
|
||||
from app.refusal import refused
|
||||
|
||||
def f(args):
|
||||
pick = refused
|
||||
raise refused(*args)
|
||||
""",
|
||||
)
|
||||
unread = _code(r, "refused")["unread"]
|
||||
assert any("the helper used as a value" in u for u in unread)
|
||||
assert any("refused(*args)" in u for u in unread)
|
||||
|
||||
|
||||
def test_a_default_is_what_a_call_that_omits_it_sends():
|
||||
r = _scan(
|
||||
routes___common="""
|
||||
def not_found(what: str = "generic"):
|
||||
return HTTPException(404, {"code": "not_found", "reason": what})
|
||||
|
||||
def get_or_404(db, id_, *, what: str = "generic"):
|
||||
raise not_found(what)
|
||||
""",
|
||||
routes__a="""
|
||||
from app.routes._common import get_or_404, not_found
|
||||
|
||||
def f(db):
|
||||
get_or_404(db, 1, what="user")
|
||||
get_or_404(db, 2)
|
||||
raise not_found("debug_session")
|
||||
""",
|
||||
)
|
||||
assert _code(r, "not_found")["reasons"] == ["debug_session", "generic", "user"]
|
||||
|
||||
|
||||
def test_a_caught_exception_answers_with_its_attribute_type_every_class_it_catches():
|
||||
r = _scan(
|
||||
errors="""
|
||||
from typing import Literal
|
||||
ClipReason = Literal["bad_range", "too_short"]
|
||||
EditReason = Literal["not_media"]
|
||||
|
||||
class ClipError(Exception):
|
||||
def __init__(self, reason: ClipReason, **params: float):
|
||||
self.reason: ClipReason = reason
|
||||
self.params = params
|
||||
|
||||
class EditError(Exception):
|
||||
def __init__(self, reason: EditReason):
|
||||
self.reason: EditReason = reason
|
||||
|
||||
class Untyped(Exception):
|
||||
def __init__(self, reason: str):
|
||||
self.reason = reason
|
||||
|
||||
class Narrower(ClipError):
|
||||
pass
|
||||
""",
|
||||
routes__a="""
|
||||
from app import errors
|
||||
|
||||
def f():
|
||||
try:
|
||||
pass
|
||||
except (errors.ClipError, errors.EditError) as e:
|
||||
raise HTTPException(422, {"code": "clip", "reason": e.reason, **e.params})
|
||||
except errors.Untyped as e:
|
||||
raise HTTPException(422, {"code": "clip", "reason": e.reason})
|
||||
except errors.Narrower as e:
|
||||
raise HTTPException(422, {"code": "edit", "reason": e.reason})
|
||||
""",
|
||||
)
|
||||
assert _code(r, "clip")["reasons"] == ["bad_range", "not_media", "too_short"]
|
||||
unread = _code(r, "clip")["unread"]
|
||||
assert any("Untyped.reason has no Literal type" in u for u in unread)
|
||||
# EditError keeps no `**params`, so its `e.params` could be anything
|
||||
assert any("e.params" in u for u in unread)
|
||||
assert _code(r, "edit")["reasons"] == ["bad_range", "too_short"]
|
||||
|
||||
|
||||
def test_a_typed_parameter_is_its_literal_and_an_untyped_one_is_unread_when_never_called():
|
||||
r = _scan(
|
||||
a="""
|
||||
from typing import Literal
|
||||
Why = Literal["one", "two"]
|
||||
|
||||
def typed(reason: Why):
|
||||
return HTTPException(409, {"code": "playlist", "reason": reason})
|
||||
|
||||
class C:
|
||||
def method(self, reason: str):
|
||||
return HTTPException(409, {"code": "playlist", "reason": reason})
|
||||
"""
|
||||
)
|
||||
assert _code(r, "playlist")["reasons"] == ["one", "two"]
|
||||
assert any("is not a module-level helper" in u for u in _code(r, "playlist")["unread"])
|
||||
|
||||
|
||||
def test_a_value_the_scan_cannot_follow_is_unread():
|
||||
r = _scan(
|
||||
a="""
|
||||
def f(svc):
|
||||
hit, code = svc.fetch()
|
||||
raise HTTPException(422, {"code": "subtitle", "reason": code})
|
||||
def g(svc):
|
||||
raise HTTPException(422, {"code": "subtitle", "reason": svc.reason()})
|
||||
def h(e):
|
||||
raise HTTPException(422, {"code": "subtitle", "reason": e.reason})
|
||||
"""
|
||||
)
|
||||
assert len(_code(r, "subtitle")["unread"]) == 3
|
||||
assert _code(r, "subtitle")["reasons"] == []
|
||||
|
||||
|
||||
def test_a_map_lookup_is_its_values_and_its_keys_must_be_what_the_index_can_be():
|
||||
src = """
|
||||
from typing import Literal
|
||||
SettingReason = Literal["too_low", "too_high"]
|
||||
|
||||
class InvalidSetting(ValueError):
|
||||
def __init__(self, key: str, reason: SettingReason, **params: object):
|
||||
self.reason: SettingReason = reason
|
||||
self.params = params
|
||||
|
||||
_MAP: dict[SettingReason, str] = {{{keys}}}
|
||||
|
||||
def refused(status: int, reason: str, **params: object):
|
||||
return HTTPException(status, {{"code": "refused", "reason": reason, **params}})
|
||||
|
||||
def setting_refused(exc: InvalidSetting):
|
||||
return refused(400, _MAP[exc.reason], **exc.params)
|
||||
"""
|
||||
whole = _scan(a=src.format(keys='"too_low": "setting_too_low", "too_high": "setting_too_high"'))
|
||||
assert _code(whole, "refused")["reasons"] == ["setting_too_high", "setting_too_low"]
|
||||
assert whole["errors"] == []
|
||||
short = _scan(a=src.format(keys='"too_low": "setting_too_low"'))
|
||||
assert len(short["errors"]) == 1 and "not what its index can be" in short["errors"][0]
|
||||
|
||||
|
||||
def test_a_detail_built_up_in_steps_is_read_through_its_variable():
|
||||
r = _scan(
|
||||
a="""
|
||||
def f(status: int, reason: str, user=None, **extra):
|
||||
detail: dict = {"code": "login", "reason": "invalid"}
|
||||
if user is not None:
|
||||
detail |= {"email": user.email, "reason": "suspended"}
|
||||
detail |= {k: v for k, v in extra.items() if v is not None}
|
||||
return HTTPException(status, detail)
|
||||
|
||||
def g(more):
|
||||
detail = {"code": "login"}
|
||||
detail["reason"] = "late"
|
||||
detail.update(more)
|
||||
detail |= more
|
||||
return HTTPException(401, detail)
|
||||
"""
|
||||
)
|
||||
entry = _code(r, "login")
|
||||
assert entry["reasons"] == ["invalid", "late", "suspended"]
|
||||
assert sum("may carry a reason" in u for u in entry["unread"]) == 2
|
||||
|
||||
|
||||
def test_a_code_must_be_one_the_scan_can_read():
|
||||
r = _scan(
|
||||
a="""
|
||||
BUSY = "busy"
|
||||
def f(code):
|
||||
raise HTTPException(409, {"code": BUSY})
|
||||
def g(code):
|
||||
raise HTTPException(409, {"code": code})
|
||||
"""
|
||||
)
|
||||
assert "busy" in r["codes"] and r["codes"]["busy"]["reasonless"]
|
||||
assert len(r["unreadCodes"]) == 1 and r["unreadCodes"][0].endswith(": code")
|
||||
|
||||
|
||||
def test_the_data_files_are_excluded_by_name_only():
|
||||
r = _scan(plex__tmdb='x = {"code": "HU", "name": "Hungary"}', plex__other='y = {"code": "HU"}')
|
||||
assert r["excluded"] == {"plex/tmdb.py": 1}
|
||||
assert "HU" in r["codes"]
|
||||
|
||||
|
||||
def test_the_app_scans():
|
||||
"""The real tree, on the interpreter the backend runs: every refusal it holds is read (the
|
||||
frontend's contract judges what is read)."""
|
||||
result = scan_app()
|
||||
assert {"refused", "not_found", "login", "clip"} <= set(result["codes"])
|
||||
assert result["errors"] == []
|
||||
Reference in New Issue
Block a user