diff --git a/backend/app/refusal.py b/backend/app/refusal.py index 337a1c8a..d996cff4 100644 --- a/backend/app/refusal.py +++ b/backend/app/refusal.py @@ -7,10 +7,11 @@ whatever sentence the route happened to hand-write. `refused` is the generic fam (`errors.refused.`, the params interpolated into the sentence); the older specific codes (`not_found`, `youtube`, `playlist`, the admin refusals, …) stay, since each is already translated. -A test (`test_http_details.py`) reads every `HTTPException` in the app and fails on a string detail. -`tests/refusal_sites.py` reads every `{"code": …}` detail from the syntax tree and follows each reason -to its values: a literal, a `Literal` type on the parameter or attribute it travels in, every call of -a helper like `refused` below, a map's values. The frontend's contract test +A test (`test_http_details.py`) reads every `HTTPException` and `JSONResponse` detail in the app and +fails on a sentence, a dict without a `code`, or none at all. `tests/refusal_sites.py` reads every +`{"code": …}` detail from the syntax tree and follows each reason along a closed list of shapes: a +literal, a `Literal` type on the parameter or attribute it travels in, every call of a helper like +`refused` below, a map's values; anything else is reported, never guessed. The frontend's contract test (`i18n/backendIds.contract.test.ts`) runs it and requires each code in one record of `lib/api/refusalCodes.ts`, and each reason's sentence in both languages; a code or reason it cannot follow fails it, unless no screen shows that code. So the rule is enforced, not described. @@ -23,7 +24,8 @@ from app.sysconfig import InvalidSetting, SettingReason def refused(status: int, reason: str, **params: object) -> HTTPException: """The refusal to raise: `raise refused(409, "view_name_taken", name=name)`. `params` travel in - the detail for the sentence to interpolate (`{{name}}`); keep them to values the reader may see.""" + the detail for the sentence to interpolate (`{{name}}`); keep them to values the reader may see, + and never a `code`: it would replace this one (the refusal scan reads such a call as an unread code).""" return HTTPException(status_code=status, detail={"code": "refused", "reason": reason, **params}) diff --git a/backend/tests/refusal_sites.py b/backend/tests/refusal_sites.py index edf79344..0fbe3bbd 100644 --- a/backend/tests/refusal_sites.py +++ b/backend/tests/refusal_sites.py @@ -2,27 +2,37 @@ 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: +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 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. +- 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: `; +- 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`. -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). +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" @@ -33,16 +43,34 @@ NOT_REFUSALS = { "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) + return Res(self.values | other.values, self.unread + other.unread, self.none or other.none) @dataclass @@ -51,10 +79,45 @@ class Module: 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 + # 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: @@ -69,16 +132,18 @@ def _load(sources: dict[str, str]) -> 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]] = {} + 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: - imports[a.asname or a.name] = (node.module, a.name) + 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: - imports[a.asname] = (a.name, None) + 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 @@ -88,8 +153,15 @@ def _load(sources: dict[str, str]) -> dict[str, Module]: 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) + mods[name] = Module(path, name, tree, parents, imports, top, ambiguous) return mods @@ -98,11 +170,17 @@ class Scan: 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) -> tuple[Module, Node] | tuple[str, None] | None: + 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: @@ -111,19 +189,33 @@ class Scan: 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) + 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 self.lookup(m, e.id) + 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)}" @@ -133,60 +225,106 @@ class Scan: 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) -> 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 + 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.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) + 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: - 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) + 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 its annotation on the class or `self.attr: T` in a method, - else from a base class.""" + """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): - 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 + 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 ------------------------------------------------------------------------------------- @@ -194,27 +332,18 @@ class Scan: def resolve(self, m: Module, e: Node) -> Res: if isinstance(e, ast.Constant): if e.value is None: - return Res() + 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.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)) - ): + 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) @@ -223,95 +352,31 @@ class Scan: 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)) + fn = self.enclosing(m, e, FUNCS) while fn is not None: - assert isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) + 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 - 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)]) + 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 a in assigned: - if ( - isinstance(a, ast.AnnAssign) - and (v := self.ann_values(m, a.annotation)) is not None + for s in assigned: + if isinstance(s, ast.Assign) and all( + isinstance(t, ast.Name) for t in s.targets ): - 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) + out = out | self.resolve(m, s.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) + fn = self.enclosing(m, fn, FUNCS) 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: @@ -322,70 +387,45 @@ class Scan: 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)) + 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() - 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" - ] - ) + 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]] - 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 - } + by_name = param not in fn.args.posonlyargs 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]) + 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): - 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) + typed = self.typed(m, e, base.id) if typed is not None: out = Res() for cm, cls in typed: @@ -426,144 +466,248 @@ class Scan: 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: + 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 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 + # --- 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): - it = src.generators[0].iter + 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(it, ast.Call) + 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 getattr(it.func.value, "id", None) == kwarg - and "reason" in names + 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 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: + 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 - 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) + 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) else None if k is None else "?", v) + ( + 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 [(k.arg, k.value) for k in d.keywords] + 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_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)) + 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 - 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: + 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, 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): + 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, 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": + 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 - assign = m.parents.get(node) - if isinstance(node.slice, ast.Constant) and isinstance(assign, ast.Assign): + 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 - 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"} + 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") ): - res = res | Res(unread=[f"{self.at(m, node)} (may carry a reason)"]) - return res, found + 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 = { diff --git a/backend/tests/test_http_details.py b/backend/tests/test_http_details.py index 4b896bd4..5bc98728 100644 --- a/backend/tests/test_http_details.py +++ b/backend/tests/test_http_details.py @@ -51,3 +51,94 @@ def test_the_scan_sees_a_sentence_when_there_is_one(): sample = ast.parse('HTTPException(status_code=400, detail=f"bad {x}")').body[0].value assert _is_text(sample.keywords[1].value) assert not _is_text(ast.parse('{"code": "refused", "reason": "x"}').body[0].value) + + +# Answers that are no refusal of the app's, by the function they are sent from: the client shows a +# detail only when it is an object (`lib/api/core.ts`), so these read as its generic line, and no screen +# needs more from them. +NO_REFUSAL = { + ("main.py", "setup_gate"): "the 503 before setup: the setup screen is what the user sees", + ("main.py", "spa_fallback"): "a bare 404 for an api path no route matches: no screen calls one", +} + + +def _detail_offenders(source: str, path: str) -> list[str]: + """Each detail the client could not say: a dict without a `code`, a sentence, or none at all. + + `HTTPException(status, detail)` and `JSONResponse({"detail": …})`; a variable or a call is the + refusal scan's to read (`refusal_sites.py`, which follows each `code` dict).""" + tree = ast.parse(source) + parents = {child: node for node in ast.walk(tree) for child in ast.iter_child_nodes(node)} + out = [] + for node in ast.walk(tree): + name = isinstance(node, ast.Call) and getattr( + node.func, "id", getattr(node.func, "attr", "") + ) + if name == "HTTPException": + kw = {k.arg: k.value for k in node.keywords} + detail = kw.get("detail", node.args[1] if len(node.args) > 1 else None) + elif name == "JSONResponse": + content = {k.arg: k.value for k in node.keywords}.get( + "content", node.args[0] if node.args else None + ) + if not isinstance(content, ast.Dict) or not any( + isinstance(k, ast.Constant) and k.value == "detail" for k in content.keys + ): + continue # an answer, not a refusal + detail = next( + v + for k, v in zip(content.keys, content.values, strict=True) + if isinstance(k, ast.Constant) and k.value == "detail" + ) + else: + continue + fn = parents.get(node) + while fn is not None and not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): + fn = parents.get(fn) + if (path, getattr(fn, "name", None)) in NO_REFUSAL: + continue + keys = ( + [k.value for k in detail.keys if isinstance(k, ast.Constant)] + if isinstance(detail, ast.Dict) + else [k.arg for k in detail.keywords] + if isinstance(detail, ast.Call) and getattr(detail.func, "id", "") == "dict" + else None + ) + if detail is None or _is_text(detail) or (keys is not None and "code" not in keys): + out.append(f"{path}:{node.lineno}") + return out + + +def test_every_detail_is_one_the_client_can_say(): + offenders = [ + o + for p in sorted(APP.rglob("*.py")) + for o in _detail_offenders(p.read_text(encoding="utf-8"), p.relative_to(APP).as_posix()) + ] + assert offenders == [], ( + "a refusal the client can only answer with its generic line — send `{code, reason}` " + f"(`refused(status, reason)`), or list it in NO_REFUSAL with why: {offenders}" + ) + + +def test_the_detail_scan_sees_each_shape_the_client_cannot_say(): + src = """ +def f(x): + raise HTTPException(409, detail={"reason": "view_name_taken"}) +def g(x): + raise HTTPException(409, dict(reason="taken")) +def h(x): + raise HTTPException(status_code=404) +def i(x): + return JSONResponse({"detail": "This instance isn't set up yet."}, status_code=503) +def j(x): + return JSONResponse(content={"detail": {"reason": "x"}}, status_code=409) +def ok(x): + raise HTTPException(409, {"code": "refused", "reason": "x"}) + raise HTTPException(409, detail(x)) + return JSONResponse({"detail": {"code": "too_large"}}, status_code=413) + return JSONResponse({"ok": True}) +def setup_gate(x): + return JSONResponse({"detail": "listed"}, status_code=503) +""" + assert _detail_offenders(src, "main.py") == [f"main.py:{n}" for n in (3, 5, 7, 9, 11)] diff --git a/backend/tests/test_refusal_sites.py b/backend/tests/test_refusal_sites.py index 9c6effbb..e9b74afc 100644 --- a/backend/tests/test_refusal_sites.py +++ b/backend/tests/test_refusal_sites.py @@ -1,6 +1,7 @@ -"""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.""" +"""The refusal scan (`refusal_sites.py`) on small sources: each shape on its closed list is followed to its +values, and each exit from that list is reported, unread or as an unread code. A scan that silently drops a +shape is a guard green over nothing, so every exit is pinned here, not only the paths the app takes. Pure: +no app import, no database.""" import textwrap @@ -17,6 +18,10 @@ def _code(result: dict, code: str) -> dict: return result["codes"][code] +def _unread(result: dict, code: str) -> list[str]: + return [u.split(": ", 1)[1] for u in _code(result, code)["unread"]] + + REFUSAL = """ from fastapi import HTTPException @@ -24,6 +29,14 @@ REFUSAL = """ return HTTPException(status_code=status, detail={"code": "refused", "reason": reason, **params}) """ +# The same helper without a `**params` spread: what a call passes it is about the reason alone. +PLAIN = """ + from fastapi import HTTPException + + def refused(status: int, reason: str) -> HTTPException: + return HTTPException(status_code=status, detail={"code": "refused", "reason": reason}) +""" + def test_a_literal_wherever_it_sits_in_the_dict(): r = _scan( @@ -45,17 +58,38 @@ def test_a_literal_wherever_it_sits_in_the_dict(): assert _code(r, "clip")["unread"] == [] -def test_every_arm_of_a_conditional_and_every_operand_of_or(): +def test_both_arms_of_a_conditional_and_a_none_arm_makes_the_site_reasonless(): r = _scan( a=""" - def f(x, e): + def f(x): raise HTTPException(409, {"code": "edit", "reason": "a" if x else "b"}) def g(x): - raise HTTPException(409, {"code": "edit", "reason": x.why or "c"}) + raise HTTPException(409, {"code": "playlist", "reason": "lit" if x else None}) + def h(x): + raise HTTPException(409, {"code": "busy", "reason": None}) """ ) - assert _code(r, "edit")["reasons"] == ["a", "b", "c"] - assert [u.split(": ", 1)[1] for u in _code(r, "edit")["unread"]] == ["x.why"] + assert _code(r, "edit")["reasons"] == ["a", "b"] and _code(r, "edit")["reasonless"] == [] + assert _code(r, "playlist")["reasons"] == ["lit"] + assert _code(r, "playlist")["reasonless"] == ["a.py:5"] + assert _code(r, "busy")["reasonless"] == ["a.py:7"] + + +def test_what_the_list_does_not_hold_is_unread(): + r = _scan( + a=""" + def f(x, svc): + raise HTTPException(409, {"code": "edit", "reason": x.why or "c"}) + def g(svc): + raise HTTPException(409, {"code": "edit", "reason": svc.reason()}) + def h(): + raise HTTPException(409, {"code": "edit", "reason": 5}) + def i(e): + raise HTTPException(409, {"code": "edit", "reason": f"x{e}"}) + """ + ) + assert _unread(r, "edit") == ["x.why or 'c'", "svc.reason()", "5", "f'x{e}'"] + assert _code(r, "edit")["reasons"] == [] def test_a_helper_is_followed_to_every_call_however_it_is_written(): @@ -86,23 +120,43 @@ def test_a_helper_is_followed_to_every_call_however_it_is_written(): assert _code(r, "refused")["unread"] == [] -def test_a_helper_passed_as_a_value_or_spread_into_is_unread(): +def test_a_call_the_helper_cannot_be_followed_through_is_unread(): r = _scan( - refusal=REFUSAL, + refusal=PLAIN, routes__a=""" from app.refusal import refused - def f(args): + def f(args, kw): pick = refused raise refused(*args) + def g(kw): + raise refused(409, **kw) + def h(): + raise refused(409) + """, + routes__b=""" + import app.refusal + + def f(): + raise app.refusal.refused(409, "dotted") + """, + routes__c=""" + from ..refusal import refused + + def f(): + raise refused(409, "relative") """, ) - 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) + unread = _unread(r, "refused") + assert len(unread) == 6 + assert sum("the helper used as a value" in u for u in unread) == 1 + assert sum("a reference the scan cannot resolve" in u for u in unread) == 2 + assert sum("does not pass reason" in u for u in unread) == 2 # `**kw`, and nothing at all + assert any(u == "refused(*args)" for u in unread) + assert _code(r, "refused")["reasons"] == [] -def test_a_default_is_what_a_call_that_omits_it_sends(): +def test_a_default_is_not_followed_a_call_must_pass_the_reason(): r = _scan( routes___common=""" def not_found(what: str = "generic"): @@ -118,12 +172,191 @@ def test_a_default_is_what_a_call_that_omits_it_sends(): get_or_404(db, 1, what="user") get_or_404(db, 2) raise not_found("debug_session") + def g(): + raise not_found() """, ) - assert _code(r, "not_found")["reasons"] == ["debug_session", "generic", "user"] + assert _code(r, "not_found")["reasons"] == ["debug_session", "user"] + assert [u.endswith("(does not pass what)") for u in _unread(r, "not_found")] == [True, True] -def test_a_caught_exception_answers_with_its_attribute_type_every_class_it_catches(): +def test_a_positional_only_reason_passed_by_name_is_not_bound_to_it(): + r = _scan( + a=""" + def deny(reason: str, /, **params): + return HTTPException(409, {"code": "playlist", "reason": reason}) + + def f(): + raise deny("by_position") + def g(): + raise deny(reason="lands_in_params") + """ + ) + assert _code(r, "playlist")["reasons"] == ["by_position"] + assert _unread(r, "playlist") == ["deny(reason='lands_in_params') (does not pass reason)"] + + +def test_a_helper_reaching_two_sites_gives_both_its_reasons_and_one_reached_again_is_unread(): + r = _scan( + a=""" + def deny(reason: str, admin): + if admin: + return HTTPException(409, {"code": "activate", "reason": reason}) + return HTTPException(409, {"code": "playlist", "reason": reason}) + + def again(reason: str, n): + if n: + return again(reason, n - 1) + return HTTPException(409, {"code": "edit", "reason": reason}) + + def f(): + raise deny("foo", True) + def g(): + raise again("bar", 3) + """ + ) + assert _code(r, "activate")["reasons"] == ["foo"] + assert _code(r, "playlist")["reasons"] == ["foo"] + assert _code(r, "edit")["reasons"] == ["bar"] + assert any("reached again while it is followed" in u for u in _unread(r, "edit")) + + +def test_a_parameter_its_body_binds_again_is_unread(): + r = _scan( + a=""" + from typing import Literal + + def refused(status, reason: str = ""): + reason = reason or "fallback" + return HTTPException(status, {"code": "refused", "reason": reason}) + + def typed(reason: Literal["a"]): + if not reason: + reason = "b" + return HTTPException(409, {"code": "edit", "reason": reason}) + + def f(): + raise refused(409, "x") + """ + ) + assert _code(r, "refused")["reasons"] == [] + assert _unread(r, "refused") == ["reason (a parameter its body binds again)"] + assert _unread(r, "edit") == ["reason (a parameter its body binds again)"] + + +def test_a_local_is_followed_only_through_plain_assignments(): + r = _scan( + a=""" + from typing import Literal + + def f(x): + why = "first" + if x: + why = "second" if x > 1 else "third" + raise HTTPException(409, {"code": "edit", "reason": why}) + def g(svc): + hit, why = svc.fetch() + raise HTTPException(409, {"code": "playlist", "reason": why}) + def h(items): + for why in items: + pass + raise HTTPException(409, {"code": "playlist", "reason": why}) + def i(): + why: Literal["typed"] = pick() + raise HTTPException(409, {"code": "playlist", "reason": why}) + def j(items): + LABELS[why] = "NotAReason" + raise HTTPException(409, {"code": "playlist", "reason": why}) + """ + ) + assert _code(r, "edit")["reasons"] == ["first", "second", "third"] + assert _code(r, "playlist")["reasons"] == [] + assert _unread(r, "playlist") == [ + "why", + "why", + "why", + "why", + ] # unpacked, a loop, annotated, unbound + + +def test_a_call_answers_with_its_literal_return_type(): + r = _scan( + a=""" + from typing import Literal + + def why(x) -> Literal["quota", "gone"]: + return "quota" + def how(x) -> str: + return "x" + + def f(x): + raise HTTPException(409, {"code": "youtube", "reason": why(x)}) + def g(x): + raise HTTPException(409, {"code": "youtube", "reason": how(x)}) + """ + ) + assert _code(r, "youtube")["reasons"] == ["gone", "quota"] + assert _unread(r, "youtube") == ["how(x)"] + + +def test_a_module_constant_is_read_only_as_mod_name_and_only_when_bound_once(): + r = _scan( + urlguard=""" + REASON_MALFORMED = "malformed" + TWICE = "a" + TWICE = "b" + if FLAG: + LATER = "c" + """, + routes__a=""" + from app import urlguard + + BARE = "bare" + + def f(): + raise HTTPException(409, {"code": "url_blocked", "reason": urlguard.REASON_MALFORMED}) + def g(): + raise HTTPException(409, {"code": "url_blocked", "reason": urlguard.TWICE}) + def h(): + raise HTTPException(409, {"code": "url_blocked", "reason": urlguard.LATER}) + def i(): + raise HTTPException(409, {"code": "url_blocked", "reason": BARE}) + """, + ) + assert _code(r, "url_blocked")["reasons"] == ["malformed"] + assert _unread(r, "url_blocked") == ["urlguard.TWICE", "urlguard.LATER", "BARE"] + + +def test_a_name_imported_from_two_places_or_under_an_if_is_not_looked_up(): + r = _scan( + m1='R = Literal["one"]', + m2='R = Literal["two"]', + a=""" + def f(): + from app.m1 import R + def g(): + from app.m2 import R + def h(reason: R): + return HTTPException(409, {"code": "edit", "reason": reason}) + """, + b=""" + if FLAG: + from app.m1 import R + def h(reason: R): + return HTTPException(409, {"code": "playlist", "reason": reason}) + """, + c=""" + from app.m1 import R + def h(reason: R): + return HTTPException(409, {"code": "clip", "reason": reason}) + """, + ) + assert _code(r, "edit")["reasons"] == [] and _unread(r, "edit") == ["reason"] + assert _code(r, "playlist")["reasons"] == [] and _unread(r, "playlist") == ["reason"] + assert _code(r, "clip")["reasons"] == ["one"] + + +def test_a_caught_exception_answers_with_the_literal_its_class_declares_on_self(): r = _scan( errors=""" from typing import Literal @@ -136,8 +369,7 @@ def test_a_caught_exception_answers_with_its_attribute_type_every_class_it_catch self.params = params class EditError(Exception): - def __init__(self, reason: EditReason): - self.reason: EditReason = reason + reason: EditReason class Untyped(Exception): def __init__(self, reason: str): @@ -152,57 +384,80 @@ def test_a_caught_exception_answers_with_its_attribute_type_every_class_it_catch def f(): try: pass + except errors.ClipError as e: + raise HTTPException(422, {"code": "clip", "reason": e.reason}) except (errors.ClipError, errors.EditError) as e: - raise HTTPException(422, {"code": "clip", "reason": e.reason, **e.params}) + raise HTTPException(422, {"code": "clip", "reason": e.reason}) + except errors.EditError as e: + raise HTTPException(422, {"code": "clip", "reason": e.reason}) 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}) + raise HTTPException(422, {"code": "clip", "reason": e.reason}) + except errors.ClipError as e: + raise HTTPException(422, {"code": "clip", "reason": e}) """, ) - 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"] + assert _code(r, "clip")["reasons"] == ["bad_range", "too_short"] + assert _unread(r, "clip") == [ + "e.reason", # a tuple of classes + "e.reason (EditError.reason has no Literal type)", # declared on the class body only + "e.reason (Untyped.reason has no Literal type)", + "e.reason (Narrower.reason has no Literal type)", # a subclass: not looked up the bases + "e", # the exception itself + ] -def test_a_typed_parameter_is_its_literal_and_an_untyped_one_is_unread_when_never_called(): +def test_a_parameter_answers_with_its_literal_or_its_class_and_any_other_type_is_unread(): r = _scan( a=""" from typing import Literal Why = Literal["one", "two"] + class Err(Exception): + def __init__(self, reason: Literal["three"]): + self.reason: Literal["three"] = reason + def typed(reason: Why): return HTTPException(409, {"code": "playlist", "reason": reason}) + def inline(reason: Literal["four"]): + return HTTPException(409, {"code": "playlist", "reason": reason}) + def by_class(exc: Err): + return HTTPException(409, {"code": "playlist", "reason": exc.reason}) + def rebound(exc: Err): + exc = other() + return HTTPException(409, {"code": "playlist", "reason": exc.reason}) + def optional(reason: Literal["five"] | None = None): + return HTTPException(409, {"code": "playlist", "reason": reason}) + def quoted(reason: "Why"): + return HTTPException(409, {"code": "playlist", "reason": reason}) + def number(reason: int): + 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"]) + assert _code(r, "playlist")["reasons"] == ["four", "one", "three", "two"] + unread = _unread(r, "playlist") + # a parameter bound again, `| None`, a quoted annotation, `int` + assert unread[:4] == ["exc.reason", "reason", "reason", "reason"] + assert "method(reason) is not a module-level helper" in unread[4] + assert len(unread) == 5 -def test_a_value_the_scan_cannot_follow_is_unread(): +def test_a_str_helper_nobody_calls_sends_nothing(): 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}) + def helper(reason: str): + return HTTPException(409, {"code": "playlist", "reason": reason}) """ ) - assert len(_code(r, "subtitle")["unread"]) == 3 - assert _code(r, "subtitle")["reasons"] == [] + assert _code(r, "playlist") == {"reasons": [], "unread": [], "sites": 1, "reasonless": []} -def test_a_map_lookup_is_its_values_and_its_keys_must_be_what_the_index_can_be(): +def test_a_map_lookup_is_its_values_and_its_keys_must_be_exactly_what_the_index_can_be(): src = """ from typing import Literal SettingReason = Literal["too_low", "too_high"] @@ -218,13 +473,114 @@ def test_a_map_lookup_is_its_values_and_its_keys_must_be_what_the_index_can_be() return HTTPException(status, {{"code": "refused", "reason": reason, **params}}) def setting_refused(exc: InvalidSetting): - return refused(400, _MAP[exc.reason], **exc.params) + return refused(400, _MAP[{index}], **exc.params) """ - whole = _scan(a=src.format(keys='"too_low": "setting_too_low", "too_high": "setting_too_high"')) + both = '"too_low": "setting_too_low", "too_high": "setting_too_high"' + whole = _scan(a=src.format(keys=both, index="exc.reason")) 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] + assert whole["errors"] == [] and whole["unreadCodes"] == [] + for keys, index in [ + ('"too_low": "setting_too_low"', "exc.reason"), # a key short + (both + ', "gone": "setting_gone"', "exc.reason"), # a key no index can be + (both, "exc.key"), # an index the scan cannot read + ]: + r = _scan(a=src.format(keys=keys, index=index)) + assert len(r["errors"]) == 1 and "not what its index can be" in r["errors"][0] + + +def test_a_lookup_in_anything_but_a_module_dict_of_literal_keys_is_unread(): + r = _scan( + a=""" + BUILT = make() + NUMBERED = {1: "one"} + + def f(table, k): + raise HTTPException(409, {"code": "edit", "reason": table[k]}) + def g(k): + raise HTTPException(409, {"code": "edit", "reason": BUILT[k]}) + def h(k): + raise HTTPException(409, {"code": "edit", "reason": NUMBERED[k]}) + """ + ) + assert _unread(r, "edit") == ["table[k]", "BUILT[k]", "NUMBERED[k]"] + assert r["errors"] == [] + + +def test_a_code_must_be_one_literal(): + r = _scan( + a=""" + BUSY = "busy" + def f(code): + raise HTTPException(409, {"code": "busy"}) + def g(code): + raise HTTPException(409, {"code": code}) + def h(): + raise HTTPException(409, {"code": BUSY}) + def i(x): + raise HTTPException(409, {"code": "a" if x else "b"}) + def j(x): + raise HTTPException(409, {"code": "a" if x else None}) + def k(key): + raise HTTPException(409, {"code": "edit", key: "late"}) + """ + ) + assert r["codes"]["busy"]["reasonless"] == ["a.py:4"] + assert set(r["codes"]) == {"busy"} + assert len(r["unreadCodes"]) == 5 + assert r["unreadCodes"][-1].endswith("(a key the scan cannot read)") + + +def test_a_spread_is_clean_only_from_an_untouched_kwargs_no_call_puts_a_code_in(): + def unread_codes(helper: str, call: str) -> list[str]: + return _scan(a=helper + "\n" + f"def caller(x, more):\n raise {call}\n")["unreadCodes"] + + helper = """ +def deny(status, reason: str, **params): + return HTTPException(status, {"code": "refused", "reason": reason, **params}) +""" + assert unread_codes(helper, 'deny(409, "a", limit=3)') == [] + assert unread_codes(helper, 'deny(409, "a", code=x)') # the spread puts its own code + assert unread_codes(helper, 'deny(409, "a", **more)') # a spread nothing vouches for + assert unread_codes(helper + "pick = deny\n", 'deny(409, "a")') # passed around + posonly = helper.replace("reason: str,", "reason: str, /,") # `reason=` lands in params + assert unread_codes(posonly, 'deny(409, "a")') + mutated = helper.replace(" return", ' params["reason"] = "zz"\n return') + assert unread_codes(mutated, 'deny(409, "a")') + renamed = helper.replace("**params})", "**{k.lstrip('_'): v for k, v in params.items()}})") + assert unread_codes(renamed, 'deny(409, "a")') + filtered = helper.replace("**params})", "**{k: v for k, v in params.items() if v}})") + assert unread_codes(filtered, 'deny(409, "a")') == [] + + +def test_a_caught_exceptions_params_are_clean_only_when_kept_from_its_init(): + def unread_codes(errors: str, construct: str = 'Err("a", limit=1)') -> list[str]: + return _scan( + errors=errors, + a=f""" + from app.errors import Err + def f(): + raise {construct} + def g(): + try: + pass + except Err as e: + raise HTTPException(422, {{"code": "clip", "reason": e.reason, **e.params}}) + """, + )["unreadCodes"] + + kept = """ + from typing import Literal + class Err(Exception): + def __init__(self, reason: Literal["a"], **params: float): + super().__init__(f"{reason} {params}") + self.reason: Literal["a"] = reason + self.params = params + """ + assert unread_codes(kept) == [] + assert unread_codes(kept, 'Err("a", code="other")') + assert unread_codes(kept + "\n class Sub(Err):\n pass\n") + assert unread_codes(kept.replace("self.params = params", "self.params = {'code': 'x'}")) + assert unread_codes(kept + "\n def more(self):\n self.params = {}\n") def test_a_detail_built_up_in_steps_is_read_through_its_variable(): @@ -235,37 +591,58 @@ def test_a_detail_built_up_in_steps_is_read_through_its_variable(): 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} + detail["email"] = "x" return HTTPException(status, detail) - def g(more): - detail = {"code": "login"} + def g(x): + detail = {"code": "login"} if x else {"code": "login", "reason": "other"} detail["reason"] = "late" - detail.update(more) - detail |= more - return HTTPException(401, detail) + return HTTPException(401, detail=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 + assert r["unreadCodes"] == [] + assert _code(r, "login")["reasons"] == ["invalid", "late", "other", "suspended"] -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}) - """ +def test_a_detail_changed_or_sent_any_other_way_is_an_unread_code(): + changes = [ + "detail.update(more)", + "detail |= more", + 'detail |= {"code": "other"}', + 'detail["code"] = "other"', + 'del detail["reason"]', + "detail[key] = more", + 'detail = {**detail, "reason": "b"}', + "detail = rebuilt()", + "copy = detail", + "add_extra(detail)", + "return detail", + 'return {**detail, "reason": "b"}', + 'return dict(detail, reason="b")', + ] + for change in changes: + r = _scan( + a=f""" + def f(key, more): + detail = {{"code": "login", "reason": "a"}} + {change} + raise HTTPException(401, detail) + """ + ) + assert r["unreadCodes"], change + assert "login" not in r["codes"], change + held = _scan(a='class C:\n detail = {"code": "login"}\n') + assert ( + held["unreadCodes"] + and "a detail put where the scan cannot follow it" in held["unreadCodes"][0] ) - 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"}') + r = _scan( + plex__tmdb='def f():\n return {"code": "HU", "name": "Hungary"}', + plex__other='def f():\n return {"code": "HU"}', + ) assert r["excluded"] == {"plex/tmdb.py": 1} assert "HU" in r["codes"] @@ -275,4 +652,4 @@ def test_the_app_scans(): frontend's contract judges what is read).""" result = scan_app() assert {"refused", "not_found", "login", "clip"} <= set(result["codes"]) - assert result["errors"] == [] + assert result["errors"] == [] and result["unreadCodes"] == [] diff --git a/frontend/src/i18n/backendIds.contract.test.ts b/frontend/src/i18n/backendIds.contract.test.ts index 436d5ce6..e060932c 100644 --- a/frontend/src/i18n/backendIds.contract.test.ts +++ b/frontend/src/i18n/backendIds.contract.test.ts @@ -288,9 +288,7 @@ describe("backend ids have their names in both languages", () => { expect(reasoned.flatMap((code) => scan.codes[code]?.reasonless ?? [])).toEqual([]); }); - it("read every refusal's reasons (a scan that matches nothing is green over nothing)", () => { - const coded = [...Object.keys(REFUSAL_FAMILIES), ...Object.keys(REFUSAL_BY_REASON)]; - expect(Object.keys(reasonsOf).sort()).toEqual(coded.sort()); + it("every code whose reason picks the sentence is read with a reason (a scan that matches nothing is green over nothing)", () => { expect(Object.entries(reasonsOf).filter(([, rs]) => rs.length === 0)).toEqual([]); });