From 31a75f49856c51ead21e6c48a85374ace1d7d441 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 31 Oct 2026 23:54:55 +0100 Subject: [PATCH] fix(i18n): write each refusal where it is sent, and read only that form A refusal's detail now stands where it is sent (HTTPException's detail, a JSONResponse's "detail") or is returned by a factory that only ever is, with any spread before its code and reason, so no spread can replace them. The scan follows no variable and no kwargs any more; what it cannot read in a reason (a lambda's or comprehension's name, a nonlocal or global, a local that reaches itself, a constant or map changed elsewhere, an app import it cannot resolve) is reported. The detail guard reads the same slots and fails on anything but a code dict or a factory's call. --- backend/app/refusal.py | 18 +- backend/app/routes/auth.py | 18 +- backend/app/routes/downloads.py | 5 +- backend/tests/refusal_sites.py | 509 +++++++++++++--------------- backend/tests/test_http_details.py | 178 ++++------ backend/tests/test_refusal_sites.py | 333 ++++++++++++------ 6 files changed, 546 insertions(+), 515 deletions(-) diff --git a/backend/app/refusal.py b/backend/app/refusal.py index d996cff4..1e47f547 100644 --- a/backend/app/refusal.py +++ b/backend/app/refusal.py @@ -7,11 +7,13 @@ 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` 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 +A test (`test_http_details.py`) reads every place an `HTTPException` or a `JSONResponse` sends a detail +and fails on one the client cannot say: a sentence, a dict without a `code`, a variable, none at all. +`tests/refusal_sites.py` reads every `{"code": …}` detail from the syntax tree, written where it is sent +(or returned by a factory that only ever is), with any `**` spread before its `code` and `reason`, 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. @@ -24,9 +26,9 @@ 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, - 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}) + the detail for the sentence to interpolate (`{{name}}`); keep them to values the reader may see. + They are spread first, so a `code` or `reason` among them could never replace these two.""" + return HTTPException(status_code=status, detail={**params, "code": "refused", "reason": reason}) # `sysconfig.InvalidSetting.reason` → the refusal it answers as. A dict of literals, so the refusal diff --git a/backend/app/routes/auth.py b/backend/app/routes/auth.py index 92c6a4ef..b90791dd 100644 --- a/backend/app/routes/auth.py +++ b/backend/app/routes/auth.py @@ -1001,15 +1001,15 @@ def _login_refused( `email_verified` is false, and vice versa — the SPA renders whichever is actually outstanding rather than assuming a sequence. """ - detail: dict = {"code": "login", "reason": reason} - if user is not None: - detail |= { - "email": user.email, - "email_verified": user.email_verified, - "approved": user.is_active, - } - detail |= {k: v for k, v in extra.items() if v is not None} - return HTTPException(status_code=status, detail=detail) + account = ( + {"email": user.email, "email_verified": user.email_verified, "approved": user.is_active} + if user is not None + else {} + ) + given = {k: v for k, v in extra.items() if v is not None} + return HTTPException( + status_code=status, detail={**account, **given, "code": "login", "reason": reason} + ) @router.post("/password-login") diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index b7e672d8..380bb337 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -359,8 +359,9 @@ def enqueue_clip( job = service.enqueue_clip(db, user.id, spec, display_name=display_name) except clipmod.ClipError as e: status = 404 if e.reason == "not_found" else 422 - detail = {"code": "clip", "reason": e.reason, **e.params} - raise HTTPException(status_code=status, detail=detail) from e + raise HTTPException( + status_code=status, detail={**e.params, "code": "clip", "reason": e.reason} + ) from e except quota.QuotaExceeded as e: raise HTTPException(status_code=422, detail=quota_detail(e)) from e db.commit() diff --git a/backend/tests/refusal_sites.py b/backend/tests/refusal_sites.py index 0fbe3bbd..9f8453e9 100644 --- a/backend/tests/refusal_sites.py +++ b/backend/tests/refusal_sites.py @@ -2,12 +2,19 @@ with every reason its code can be sent with. The frontend's contract (`i18n/backendIds.contract.test.ts`) runs this and requires a sentence for each -code and reason in both languages. A regex over the source read only the shapes it was written for, and a -scan that tries to follow every shape opens a silent gap with each one it learns. So this one reads a -CLOSED list, the shapes the app writes, and reports everything else, never guesses: +code and reason in both languages. A scan that tries to follow every shape opens a silent gap with each one +it learns, and following how a value flows through Python (variables, scopes, aliases, a dict changed in +place) never closes. So the CODE keeps to a narrow form, and this reads a CLOSED list of it, reporting +everything else, never guessing: +- where a detail stands: where it is sent — `HTTPException`'s second argument or `detail=`, the `"detail"` + value of a `JSONResponse`'s content dict — through the arms of `a if c else b`, or as what a module-level + factory returns when every return of it is a detail and every reference to it is a call sent as one. A + detail held in a variable is not followed; +- the code: one literal. A `**` spread, or a key that is not a literal, stands BEFORE a literal `code` and + `reason`, both of which the detail then carries: the later key wins, so no spread can change them; - a reason: a string literal (`None` sends none: the site counts as reasonless); both arms of - `a if c else b`; a local variable bound only by plain `x = …`; a module constant reached as `mod.NAME`; + `a if c else b`; a function's local bound only by plain `x = …`; a module constant reached as `mod.NAME`; a call's `Literal` return annotation; - a parameter typed with a `Literal` (basedpyright holds every value passed in), or a `str` one of a module-level helper (`refused`, `not_found`), followed to every call, each of which must pass it by @@ -15,17 +22,15 @@ CLOSED list, the shapes the app writes, and reports everything else, never guess - 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`. + index can be. -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). +A name bound twice at module level, inside a module-level `if`/`try`/loop, named by a `global`, or imported +from two places is not looked up; nor is a local a lambda or a comprehension binds, or one a `nonlocal` or +`global` names. A module constant or map that anything stores into, deletes from or (a map) uses but by +`map[k]` or a read method is unread. A reference named like a followed helper that the scan cannot resolve +(`import app.x`, a relative import, an app module that does not bind it) is unread. Anything else is UNREAD, +or an unread code, and the contract fails on it unless no screen shows that code. Stdlib only: it runs on +the host, beside the frontend's tests (`python -I backend/tests/refusal_sites.py` → JSON). """ import ast @@ -43,13 +48,13 @@ 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"} +# What a map may be used by besides `map[k]`: a method that only reads it. +READS = {"get", "items", "keys", "values"} Node = ast.AST FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef) SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef) +COMPREHENSIONS = (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp) COMPOUND = ( ast.If, ast.Try, @@ -82,7 +87,10 @@ class Module: # 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 + ambiguous: set[ + str + ] # bound more than once, conditionally, by a `global`, or imported from two places + slots: set[int] # the ids of the nodes sent as a detail def _own_scope(node: Node) -> Iterator[Node]: @@ -120,6 +128,39 @@ def _binds(node: Node) -> set[str]: return set() +def _sender(call: ast.Call) -> str | None: + return getattr(call.func, "id", None) or getattr(call.func, "attr", None) + + +def detail_slots(tree: Node) -> Iterator[tuple[ast.Call, Node | None]]: + """Every place a detail is sent from, with what it holds: `HTTPException`'s second argument or + `detail=` (None: sent without one), and the `"detail"` value of a `JSONResponse`'s content dict (a + content without one is an answer, not a refusal).""" + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + kw = {k.arg: k.value for k in node.keywords} + if _sender(node) == "HTTPException": + yield node, kw.get("detail", node.args[1] if len(node.args) > 1 else None) + elif _sender(node) == "JSONResponse": + content = kw.get("content", node.args[0] if node.args else None) + if isinstance(content, ast.Dict): + for k, v in zip(content.keys, content.values, strict=True): + if isinstance(k, ast.Constant) and k.value == "detail": + yield node, v + + +def _code_dict(node: Node) -> bool: + """A `{"code": …}` dict, or a `dict(code=…)` call.""" + if isinstance(node, ast.Dict): + return any(isinstance(k, ast.Constant) and k.value == "code" for k in node.keys) + return ( + isinstance(node, ast.Call) + and getattr(node.func, "id", None) == "dict" + and any(k.arg == "code" for k in node.keywords) + ) + + def _module_name(path: str) -> str: parts = path.removesuffix(".py").split("/") if parts[-1] == "__init__": @@ -133,6 +174,7 @@ def _load(sources: dict[str, str]) -> dict[str, Module]: tree = ast.parse(src) parents = {child: node for node in ast.walk(tree) for child in ast.iter_child_nodes(node)} origins: dict[str, set[tuple[str, str | None]]] = {} + ambiguous: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module and node.level == 0: for a in node.names: @@ -141,9 +183,11 @@ def _load(sources: dict[str, str]) -> dict[str, Module]: for a in node.names: if a.asname: origins.setdefault(a.asname, set()).add((a.name, None)) + elif isinstance(node, ast.Global): + ambiguous |= set(node.names) top: dict[str, Node] = {} count: dict[str, int] = {} - ambiguous = {name for name, o in origins.items() if len(o) > 1} + 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 @@ -160,8 +204,9 @@ def _load(sources: dict[str, str]) -> dict[str, Module]: ambiguous.add(name) ambiguous |= {name for name, c in count.items() if c > 1} imports = {name: next(iter(o)) for name, o in origins.items() if name not in ambiguous} + slots = {id(d) for _, d in detail_slots(tree) if d is not None} name = _module_name(path) - mods[name] = Module(path, name, tree, parents, imports, top, ambiguous) + mods[name] = Module(path, name, tree, parents, imports, top, ambiguous, slots) return mods @@ -171,7 +216,9 @@ class Scan: 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] = {} + self._resolving: set[int] = set() + self._factory: dict[int, bool] = {} + self._changed: dict[int, bool] = {} # --- names --------------------------------------------------------------------------------------- @@ -195,26 +242,31 @@ class Scan: def lookup_expr(self, m: Module, e: Node) -> tuple[Module, Node] | tuple[str, None] | None: if isinstance(e, ast.Name): - return None if self.local(m, e) else self.lookup(m, e.id) + return None if self.binder(m, e) is not None else self.lookup(m, e.id) if isinstance(e, ast.Attribute): base = self.lookup_expr(m, e.value) if base and base[1] is None and isinstance(base[0], str): return self.lookup(self.mods[base[0]], e.attr) return None - def 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 binder(self, m: Module, e: ast.Name) -> Node | None: + """The innermost scope `e` is bound in: a function (a parameter, or a binding in its own body), a + lambda (a parameter), a comprehension (its target); None: a module-level name.""" + node = m.parents.get(e) + while node is not None: + if isinstance(node, (*FUNCS, ast.Lambda)): + a = node.args + params = [*a.posonlyargs, *a.args, *a.kwonlyargs, a.vararg, a.kwarg] + if e.id in {p.arg for p in params if p}: + return node + if isinstance(node, FUNCS) and self.assignments(node, e.id): + return node + elif isinstance(node, COMPREHENSIONS) and any( + e.id in _stored(g.target) for g in node.generators + ): + return node + node = m.parents.get(node) + return None def at(self, m: Module, e: Node) -> str: return f"{m.path}:{getattr(e, 'lineno', '?')}: {ast.unparse(e)}" @@ -269,17 +321,56 @@ class Scan: root = node while isinstance(root, ast.Attribute): root = root.value - if not isinstance(root, ast.Name) or self.local(m, root): + if not isinstance(root, ast.Name) or self.binder(m, root) is not None: return False if root.id in m.ambiguous or (root.id not in m.top and root.id not in m.imports): return True - if root is node or root.id in m.top: + if root.id in m.top: return False + if root is node: # imported from an app module that does not bind it where the scan can see + mod = m.imports[root.id][0] + return (mod == "app" or mod.startswith("app.")) and self.lookup(m, root.id) is None found = self.lookup(m, root.id) return ( found is not None and found[1] is None ) # an app module whose dotted path went nowhere + def changed(self, m: Module, stmt: Node, name: str, *, only_read: bool) -> bool: + """Something in the app stores into, deletes or rebinds the module-level `name` that `stmt` binds, + or (`only_read`, a map) uses it but by `map[k]` or a method that only reads it.""" + key = id(stmt) * 2 + only_read + if key in self._changed: + return self._changed[key] + own = stmt.targets if isinstance(stmt, ast.Assign) else [getattr(stmt, "target", None)] + out = False + for cm in self.mods.values(): + for node in ast.walk(cm.tree): + if out: + break + if not ( + isinstance(node, ast.Name) + and node.id == name + or isinstance(node, ast.Attribute) + and node.attr == name + ) or (cm is m and node in own): + continue + found = self.lookup_expr(cm, node) + if not (found and found[0] is m and found[1] is stmt): + continue + if not isinstance(node.ctx, ast.Load): + out = True + elif only_read: + p = cm.parents.get(node) + out = not ( + isinstance(p, ast.Subscript) + and p.value is node + and isinstance(p.ctx, ast.Load) + or isinstance(p, ast.Attribute) + and p.attr in READS + ) + self._changed[key] = out + return out + # --- types --------------------------------------------------------------------------------------- def ann_values(self, m: Module, ann: Node | None, depth: int = 0) -> set[str] | None: @@ -352,30 +443,33 @@ class Scan: return Res(unread=[self.at(m, e)]) def resolve_name(self, m: Module, e: ast.Name) -> Res: - fn = self.enclosing(m, e, FUNCS) - while fn is not None: - assert isinstance(fn, FUNCS) - if self.caught_as(m, e, e.id) is not None: - return Res(unread=[self.at(m, e)]) # the exception itself, not a reason - a = fn.args - params = {p.arg: p for p in [*a.posonlyargs, *a.args, *a.kwonlyargs]} - assigned = self.assignments(fn, e.id) - if e.id in params: - if assigned: - return Res(unread=[f"{self.at(m, e)} (a parameter its body binds again)"]) - return self.resolve_param(m, fn, params[e.id], e) + if self.caught_as(m, e, e.id) is not None: + return Res(unread=[self.at(m, e)]) # the exception itself, not a reason + fn = self.binder(m, e) + if fn is None: + return Res(unread=[self.at(m, e)]) + if not isinstance(fn, FUNCS): + return Res(unread=[f"{self.at(m, e)} (bound by a lambda or a comprehension)"]) + if any(isinstance(n, (ast.Nonlocal, ast.Global)) and e.id in n.names for n in ast.walk(fn)): + return Res(unread=[f"{self.at(m, e)} (named by a nonlocal or a global)"]) + a = fn.args + params = {p.arg: p for p in [*a.posonlyargs, *a.args, *a.kwonlyargs]} + assigned = self.assignments(fn, e.id) + if e.id in params: if assigned: - out = Res() - for s in assigned: - if isinstance(s, ast.Assign) and all( - isinstance(t, ast.Name) for t in s.targets - ): - out = out | self.resolve(m, s.value) - else: - out = out | Res(unread=[self.at(m, e)]) - return out - fn = self.enclosing(m, fn, FUNCS) - return Res(unread=[self.at(m, e)]) + return Res(unread=[f"{self.at(m, e)} (a parameter its body binds again)"]) + return self.resolve_param(m, fn, params[e.id], e) + out = Res(unread=[] if assigned else [self.at(m, e)]) # `*args`, `**kwargs` + for s in assigned: + if not (isinstance(s, ast.Assign) and all(isinstance(t, ast.Name) for t in s.targets)): + out = out | Res(unread=[self.at(m, e)]) + elif id(s) in self._resolving: + out = out | Res(unread=[f"{self.at(m, e)} (reached again while it is resolved)"]) + else: + self._resolving.add(id(s)) + out = out | self.resolve(m, s.value) + self._resolving.discard(id(s)) + return out def resolve_param(self, m: Module, fn: Node, param: ast.arg, e: Node) -> Res: v = self.ann_values(m, param.annotation) @@ -443,10 +537,11 @@ class Scan: found and isinstance(found[0], Module) and isinstance(found[1], (ast.Assign, ast.AnnAssign)) + and isinstance(found[1].value, ast.Constant) ): - stmt = found[1] - if isinstance(stmt.value, ast.Constant): - return self.resolve(found[0], stmt.value) + if self.changed(found[0], found[1], e.attr, only_read=False): + return Res(unread=[f"{self.at(m, e)} (bound again or deleted elsewhere)"]) + return self.resolve(found[0], found[1].value) return Res(unread=[self.at(m, e)]) def resolve_lookup(self, m: Module, e: ast.Subscript) -> Res: @@ -462,6 +557,9 @@ class Scan: isinstance(k, ast.Constant) and isinstance(k.value, str) for k in table.keys ): return Res(unread=[self.at(m, e)]) + name = e.value.attr if isinstance(e.value, ast.Attribute) else getattr(e.value, "id", "") + if self.changed(found[0], found[1], name, only_read=True): + return Res(unread=[f"{self.at(m, e)} (the map is changed after it is built)"]) out = Res() for v in table.values: out = out | self.resolve(found[0], v) @@ -474,119 +572,57 @@ class Scan: ) return out - # --- spreads ------------------------------------------------------------------------------------- - - def clean_spread(self, m: Module, at: Node, src: Node) -> bool: - """`**src` cannot carry a `reason` or a `code`: it is a function's `**kwargs` (or a caught - exception's `e.params` kept from its `__init__`'s), or `{k: … for k, v in that.items()}`: a - value may change, a key may not.""" - if isinstance(src, ast.DictComp): - g = src.generators - if len(g) != 1 or not isinstance(g[0].target, ast.Tuple) or len(g[0].target.elts) != 2: - return False - k, v = g[0].target.elts - it = g[0].iter - return ( - isinstance(k, ast.Name) - and isinstance(v, ast.Name) - and getattr(src.key, "id", None) == k.id - and isinstance(it, ast.Call) - and isinstance(it.func, ast.Attribute) - and it.func.attr == "items" - and not it.args - and self.clean_spread(m, at, it.func.value) - ) - if isinstance(src, ast.Attribute) and isinstance(src.value, ast.Name): - typed = self.typed(m, at, src.value.id) - return bool(typed) and all(self.kept(cm, cls, src.attr) for cm, cls in typed or []) - if isinstance(src, ast.Name): - fn = self.enclosing(m, at, FUNCS) - return ( - fn is not None - and fn.args.kwarg is not None - and src.id == fn.args.kwarg.arg - and self.kwargs_clean(m, fn, fn) - ) - return False - - def kept(self, cm: Module, cls: ast.ClassDef, attr: str) -> bool: - """`self.attr` is only ever the `**kwargs` of the class's `__init__`, and that one is clean.""" - init = next( - (s for s in cls.body if isinstance(s, ast.FunctionDef) and s.name == "__init__"), None - ) - if init is None or init.args.kwarg is None: - return False - stores = [ - n - for n in ast.walk(cls) - if isinstance(n, ast.Attribute) - and n.attr == attr - and getattr(n.value, "id", None) == "self" - and not isinstance(n.ctx, ast.Load) - ] - kept = [ - s - for s in stores - if isinstance(p := cm.parents.get(s), ast.Assign) - and getattr(p.value, "id", None) == init.args.kwarg.arg - and self.enclosing(cm, p, FUNCS) is init - ] - return len(stores) == 1 and len(kept) == 1 and self.kwargs_clean(cm, init, cls) - - def kwargs_clean(self, m: Module, fn: Node, target: Node) -> bool: - """`fn`'s `**kwargs` holds no `reason` (`fn` takes it by name) and no `code`: it is used for - nothing but `**kwargs`, `kwargs.items()`, `self.x = kwargs` or an f-string, and every call of - `target` (the helper, or the class `fn` is the `__init__` of) passes no `code=` and no unclean - `**`.""" - assert isinstance(fn, FUNCS) and fn.args.kwarg is not None - if id(fn) in self._clean: - return self._clean[id(fn)] - self._clean[id(fn)] = False # a call that reaches it again while it is checked is not clean - kwarg = fn.args.kwarg.arg - clean = "reason" in {a.arg for a in [*fn.args.args, *fn.args.kwonlyargs]} and not ( - self.assignments(fn, kwarg) - ) - for n in _own_scope(fn): - if not clean: - break - if isinstance(n, ast.Name) and n.id == kwarg: - p = m.parents.get(n) - clean = ( - isinstance(p, ast.FormattedValue) # printed in a message - or isinstance(p, ast.keyword) - and p.arg is None - or isinstance(p, ast.Dict) - and any(k is None and v is n for k, v in zip(p.keys, p.values, strict=True)) - or isinstance(p, ast.Attribute) - and p.attr == "items" - and isinstance(m.parents.get(p), ast.Call) - or isinstance(p, ast.Assign) - and p.value is n - and all( - isinstance(t, ast.Attribute) and getattr(t.value, "id", None) == "self" - for t in p.targets - ) - ) - refs = list(self.references(m, target)) if clean else [] - for cm, node, call, resolved in refs: - if call is not None: - clean = not any( - k.arg == "code" or k.arg is None and not self.clean_spread(cm, call, k.value) - for k in call.keywords - ) - elif not resolved or not isinstance(target, ast.ClassDef): - clean = False # a helper passed around, or a name the scan cannot resolve - else: - # `except C`, an annotation: not a construction; a subclass is built some other way - cls = self.enclosing(cm, node, ast.ClassDef) - clean = not (cls is not None and any(node in ast.walk(b) for b in cls.bases)) - if not clean: - break - self._clean[id(fn)] = clean - return clean - # --- sites --------------------------------------------------------------------------------------- + def is_detail(self, node: Node | None) -> bool: + if isinstance(node, ast.IfExp): + return self.is_detail(node.body) and self.is_detail(node.orelse) + return _code_dict(node) + + def factory(self, m: Module, fn: Node) -> bool: + """A module-level function each return of which is a detail, and each reference to which is a + call sent as one (`quota_detail`).""" + assert isinstance(fn, FUNCS) + if id(fn) in self._factory: + return self._factory[id(fn)] + returns = [n for n in _own_scope(fn) if isinstance(n, ast.Return)] + ok = ( + m.top.get(fn.name) is fn # a name bound twice: its calls are unresolved below + and bool(returns) + and all(self.is_detail(r.value) for r in returns) + ) + for cm, _, call, resolved in self.references(m, fn) if ok else []: + ok = resolved and id(call) in cm.slots # a bare reference is no call: never sent + if not ok: + break + self._factory[id(fn)] = ok + return ok + + def placed(self, m: Module, d: Node) -> bool: + """`d` is sent as a detail: from a sender's slot, through the arms of `a if c else b`, or as what a + factory returns.""" + holder, p = d, m.parents.get(d) + while isinstance(p, ast.IfExp) and holder is not p.test: + holder, p = p, m.parents.get(p) + if id(holder) in m.slots: + return True + fn = self.enclosing(m, p, FUNCS) if isinstance(p, ast.Return) else None + return fn is not None and self.factory(m, fn) + + def sayable(self, m: Module, d: Node | None) -> bool: + """What a slot holds is a detail the client can say: a code dict, or a factory's call.""" + if isinstance(d, ast.IfExp): + return self.sayable(m, d.body) and self.sayable(m, d.orelse) + if _code_dict(d): + return True + found = self.lookup_expr(m, d.func) if isinstance(d, ast.Call) else None + return bool( + found + and isinstance(found[0], Module) + and isinstance(found[1], FUNCS) + and self.factory(found[0], found[1]) + ) + def entries(self, d: Node) -> list[tuple[str | None, Node]]: """A dict's (key, value) pairs: None for a `**` spread, "?" for a key that is not a literal.""" if isinstance(d, ast.Dict): @@ -610,14 +646,16 @@ class Scan: 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)" + f"{self.at(m, v)} (a spread in dict())" for k, v in items - if k is None and not self.clean_spread(m, d, v) + if k is None and isinstance(d, ast.Call) + ] + [ + f"{self.at(m, v)} (a spread or a computed key the code and the reason do not follow)" + for i, (k, v) in enumerate(items) + if k in (None, "?") and not {"code", "reason"} <= {k2 for k2, _ in items[i + 1 :]} ] - res, has_reason, changed = self.mutations(m, d) - problems += changed + if not self.placed(m, d): + problems.append(f"{self.at(m, d)} (a detail put where the scan cannot follow it)") if code.unread or code.none or len(code.values) != 1 or problems: out["unreadCodes"] += problems or [self.at(m, d)] return @@ -625,6 +663,7 @@ class Scan: 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 @@ -633,81 +672,17 @@ class Scan: entry["reasons"] |= res.values entry["unread"] += res.unread - def mutations(self, m: Module, d: Node) -> tuple[Res, bool, list[str]]: - """A detail put in a variable: what `|= {…}` and `["reason"] = …` add, and every other use that - could change it or send it elsewhere.""" - holder, stmt = d, m.parents.get(d) - while isinstance(stmt, ast.IfExp) and holder is not stmt.test: - holder, stmt = stmt, m.parents.get(stmt) - if not (isinstance(stmt, (ast.Assign, ast.AnnAssign)) and stmt.value is holder): - return Res(), False, [] - targets = stmt.targets if isinstance(stmt, ast.Assign) else [stmt.target] - fn = self.enclosing(m, d, FUNCS) - if fn is None or len(targets) != 1 or not isinstance(targets[0], ast.Name): - return ( - Res(), - False, - [f"{self.at(m, stmt)} (a detail put where the scan cannot follow it)"], - ) - name = targets[0].id - res, found, problems = Res(), False, [] - for n in _own_scope(fn): - if n is stmt: - continue - if name in _binds(n): - if not (isinstance(n, ast.AugAssign) and isinstance(n.op, ast.BitOr)): - problems.append(f"{self.at(m, n)} (binds the detail's variable again)") - elif isinstance(n.value, ast.Dict): - for k, v in self.entries(n.value): - if k == "reason": - res, found = res | self.resolve(m, v), True - elif k in ("code", "?") or k is None and not self.clean_spread(m, n, v): - problems.append(f"{self.at(m, v)} (may change the code)") - elif not self.clean_spread(m, n, n.value): - problems.append(f"{self.at(m, n)} (may change the code)") - continue - if not (isinstance(n, ast.Name) and n.id == name and isinstance(n.ctx, ast.Load)): - continue - p = m.parents.get(n) - if isinstance(p, ast.AugAssign) and p.target is n: - continue - if isinstance(p, ast.Subscript) and p.value is n: - key = p.slice.value if isinstance(p.slice, ast.Constant) else None - assign = m.parents.get(p) - if isinstance(p.ctx, ast.Load): + def other_details(self) -> list[tuple[str, int, str | None]]: + """Each slot that sends what the client cannot say — nothing, or anything but a code dict or a + factory's call: (its file, its line, the function it is sent from).""" + out = [] + for m in self.mods.values(): + for call, d in detail_slots(m.tree): + if self.sayable(m, d): continue - if ( - key == "reason" - and isinstance(p.ctx, ast.Store) - and isinstance(assign, ast.Assign) - and assign.targets == [p] - ): - res, found = res | self.resolve(m, assign.value), True - elif not ( - isinstance(key, str) - and key not in ("code", "reason") - and isinstance(p.ctx, ast.Store) - ): - problems.append(f"{self.at(m, p)} (may change the code or drop the reason)") - continue - if isinstance(p, ast.Attribute) and p.attr in READS: - continue - sender = ( - p - if isinstance(p, ast.Call) - else m.parents.get(p) - if isinstance(p, ast.keyword) - else None - ) - func = getattr(sender, "func", None) - if ( - isinstance(sender, ast.Call) - and (getattr(func, "id", None) or getattr(func, "attr", None)) in SENDERS - and (n in sender.args or isinstance(p, ast.keyword) and p.arg == "detail") - ): - continue - problems.append(f"{self.at(m, n)} (the detail goes where the scan cannot follow it)") - return res, found, problems + fn = self.enclosing(m, call, FUNCS) + out.append((m.path, call.lineno, getattr(fn, "name", None))) + return sorted(out) def run(self) -> dict: out: dict = { @@ -727,15 +702,7 @@ class Scan: ): 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): + if not _code_dict(node): continue if m.path in NOT_REFUSALS: out["excluded"][m.path] = out["excluded"].get(m.path, 0) + 1 @@ -750,14 +717,16 @@ def scan(sources: dict[str, str]) -> dict: return Scan(sources).run() +def app_sources(app: pathlib.Path = APP) -> dict[str, str]: + return { + p.relative_to(app).as_posix(): p.read_text(encoding="utf-8") + for p in sorted(app.rglob("*.py")) + if "__pycache__" not in p.parts + } + + def scan_app(app: pathlib.Path = APP) -> dict: - return scan( - { - 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 - } - ) + return scan(app_sources(app)) if __name__ == "__main__": diff --git a/backend/tests/test_http_details.py b/backend/tests/test_http_details.py index 5bc98728..b9cda470 100644 --- a/backend/tests/test_http_details.py +++ b/backend/tests/test_http_details.py @@ -1,57 +1,16 @@ -"""No refusal answers in a hand-written English sentence. +"""No refusal answers in a way the client cannot say. The UI is bilingual and translates a structured `detail` (`{code, reason, …}`), but a string detail it can only print as it came — so ~150 of them reached the Hungarian UI verbatim, in the blocking error dialog and on every inline surface that shows `HttpError.detail`. `app/refusal.py` is the family they -became; this test is what keeps a new one from being written the old way. Pure: it reads the sources. +became; this test is what keeps a new one from being written the old way. It reads the same slots the +refusal scan does (`refusal_sites.detail_slots`), so the two cannot disagree on where a detail is sent. +Pure: it reads the sources. """ -import ast -import pathlib - -APP = pathlib.Path(__file__).resolve().parents[1] / "app" - - -def _is_text(node: ast.AST) -> bool: - """A detail that is a sentence: a literal, an f-string, `str(exc)`, or `"…" + x`.""" - if isinstance(node, ast.Constant) and isinstance(node.value, str): - return True - if isinstance(node, ast.JoinedStr): - return True - if isinstance(node, ast.Call) and getattr(node.func, "id", "") == "str": - return True - if isinstance(node, ast.BoolOp): - return any(_is_text(v) for v in node.values) - if isinstance(node, ast.BinOp): - return _is_text(node.left) or _is_text(node.right) - return False - - -def test_no_http_exception_carries_a_sentence(): - offenders = [] - for path in sorted(APP.rglob("*.py")): - for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): - if not ( - isinstance(node, ast.Call) - and getattr(node.func, "id", getattr(node.func, "attr", "")) == "HTTPException" - ): - continue - kw = {k.arg: k.value for k in node.keywords} - detail = kw.get("detail", node.args[1] if len(node.args) > 1 else None) - if detail is not None and _is_text(detail): - offenders.append(f"{path.relative_to(APP).as_posix()}:{node.lineno}") - assert offenders == [], ( - "a refusal with an English sentence for a detail — raise `refused(status, reason)` " - f"(app/refusal.py) and name the reason in both locales: {offenders}" - ) - - -def test_the_scan_sees_a_sentence_when_there_is_one(): - """A guard that matches nothing is green over nothing.""" - 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) +import textwrap +from tests.refusal_sites import Scan, app_sources # 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 @@ -59,62 +18,21 @@ def test_the_scan_sees_a_sentence_when_there_is_one(): 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", + ("main.py", "_validation_error"): "FastAPI's own 422 list: a body the client itself got wrong", + ("routes/diagnostics.py", "_parsed_events"): "the client-events beacon's 422: nothing shows it", } -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 _offenders(sources: dict[str, str]) -> list[str]: + return [ + f"{path}:{line}" + for path, line, fn in Scan(sources).other_details() + if (path, fn) not in NO_REFUSAL + ] 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()) - ] + offenders = _offenders(app_sources()) 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}" @@ -122,23 +40,49 @@ def test_every_detail_is_one_the_client_can_say(): 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)] + """A guard that matches nothing is green over nothing.""" + src = textwrap.dedent( + """ + 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 k(x): + raise HTTPException(400, detail=f"bad {x}") + def l(x): + raise HTTPException(404, None) + def n(x, msg): + raise HTTPException(400, detail=msg) + def o(x): + raise HTTPException(400, detail=["a sentence"]) + def p(x): + raise HTTPException(400, detail=describe(x)) + def q(x): + raise HTTPException(400, detail={"code": "a"} if x else "text") + def r(x): + raise HTTPException(400, detail=nothing(x)) + def nothing(x): + pass + def ok(x): + raise HTTPException(409, {"code": "refused", "reason": "x"}) + raise HTTPException(409, detail=quota(x)) + raise HTTPException(409, detail={"code": "a"} if x else {"code": "b"}) + return JSONResponse({"detail": {"code": "too_large"}}, status_code=413) + return JSONResponse({"ok": True}) + def quota(x): + return {"code": "quota", "reason": "files"} + def setup_gate(x): + return JSONResponse({"detail": "listed"}, status_code=503) + """ + ) + lines = (3, 5, 7, 9, 11, 13, 15, 17, 19, 21, 23, 25) + assert _offenders({"main.py": src}) == [f"main.py:{n}" for n in lines] + # A file whose code dicts are data still answers its refusals in a way the client can say. + data = 'def f():\n raise HTTPException(400, "text")\n' + assert _offenders({"plex/tmdb.py": data}) == ["plex/tmdb.py:2"] diff --git a/backend/tests/test_refusal_sites.py b/backend/tests/test_refusal_sites.py index e9b74afc..1c3592fd 100644 --- a/backend/tests/test_refusal_sites.py +++ b/backend/tests/test_refusal_sites.py @@ -26,15 +26,7 @@ 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}) -""" - -# 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}) + return HTTPException(status_code=status, detail={**params, "code": "refused", "reason": reason}) """ @@ -109,6 +101,12 @@ def test_a_helper_is_followed_to_every_call_however_it_is_written(): def i(x): raise refused(409, "a" if x else "second_arm") """, + routes__b=""" + from other.lib import refused + + def f(): + refused("an outside library's, not the app's") + """, ) assert _code(r, "refused")["reasons"] == [ "a", @@ -122,7 +120,7 @@ def test_a_helper_is_followed_to_every_call_however_it_is_written(): def test_a_call_the_helper_cannot_be_followed_through_is_unread(): r = _scan( - refusal=PLAIN, + refusal=REFUSAL, routes__a=""" from app.refusal import refused @@ -267,6 +265,8 @@ def test_a_local_is_followed_only_through_plain_assignments(): def j(items): LABELS[why] = "NotAReason" raise HTTPException(409, {"code": "playlist", "reason": why}) + def k(*why): + raise HTTPException(409, {"code": "playlist", "reason": why}) """ ) assert _code(r, "edit")["reasons"] == ["first", "second", "third"] @@ -276,7 +276,8 @@ def test_a_local_is_followed_only_through_plain_assignments(): "why", "why", "why", - ] # unpacked, a loop, annotated, unbound + "why", + ] # unpacked, a loop, annotated, unbound, `*args` def test_a_call_answers_with_its_literal_return_type(): @@ -470,7 +471,7 @@ def test_a_map_lookup_is_its_values_and_its_keys_must_be_exactly_what_the_index_ _MAP: dict[SettingReason, str] = {{{keys}}} def refused(status: int, reason: str, **params: object): - return HTTPException(status, {{"code": "refused", "reason": reason, **params}}) + return HTTPException(status, {{**params, "code": "refused", "reason": reason}}) def setting_refused(exc: InvalidSetting): return refused(400, _MAP[{index}], **exc.params) @@ -527,115 +528,229 @@ def test_a_code_must_be_one_literal(): 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)") + assert r["unreadCodes"][-1].endswith( + "(a spread or a computed key the code and the reason do not follow)" + ) -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"] +def test_a_spread_or_a_computed_key_stands_before_both_the_code_and_the_reason(): + def unread_codes(detail: str) -> list[str]: + return _scan(a=f"def f(x, more, key):\n raise HTTPException(409, {detail})\n")[ + "unreadCodes" + ] - helper = """ -def deny(status, reason: str, **params): - return HTTPException(status, {"code": "refused", "reason": reason, **params}) + assert unread_codes('{**more, "code": "refused", "reason": "a"}') == [] + assert unread_codes('{key: 1, **more, "reason": "a", "code": "refused"}') == [] + for late in [ + '{"code": "refused", "reason": "a", **more}', # the spread may replace both + '{"code": "refused", **more, "reason": "a"}', # … the code + '{"reason": "a", **more, "code": "refused"}', # … the reason + '{**more, "code": "refused"}', # no reason after it: the spread may carry one + '{"code": "refused", "reason": "a", key: "x"}', # a key that may be either + 'dict(**more, code="refused", reason="a")', # dict() raises on a repeated key + 'dict(more, code="refused", reason="a")', + ]: + assert unread_codes(late), late + + +def test_a_detail_stands_only_where_it_is_sent(): + def result(body: str, extra: str = "") -> dict: + return _scan(a=f"def f(x, cached, fix):\n{textwrap.indent(body, ' ')}\n{extra}") + + for body in [ + 'detail = {"code": "login", "reason": "a"}\nraise HTTPException(401, detail)', + 'if (detail := {"code": "login", "reason": "a"}):\n raise HTTPException(401, detail)', + 'raise HTTPException(401, {"code": "login", "reason": "a"}.copy())', + 'raise HTTPException(401, fix({"code": "login", "reason": "a"}))', + 'raise HTTPException(401, cached or {"code": "login", "reason": "a"})', + 'return JSONResponse({"code": "login", "reason": "a"}, status_code=401)', # the whole body + 'raise HTTPException(401, None, {"code": "login", "reason": "a"})', # the headers + 'raise HTTPException(401, headers={"code": "login", "reason": "a"})', + 'raise HTTPException(401, "a" if {"code": "login", "reason": "a"} else "b")', + ]: + r = result(body) + assert r["unreadCodes"] and "login" not in r["codes"], body + assert "(a detail put where the scan cannot follow it)" in r["unreadCodes"][0], body + held = _scan(a='class C:\n detail = {"code": "login"}\n') + assert "(a detail put where the scan cannot follow it)" in held["unreadCodes"][0] + + sent = result( + 'raise HTTPException(401, {"code": "login", "reason": "a"} if x else {"code": "login"})\n' + 'raise HTTPException(401, detail=dict(code="login", reason="b"))\n' + 'return JSONResponse(content={"detail": {"code": "login", "reason": "c"}})' + ) + assert sent["unreadCodes"] == [] and _code(sent, "login")["reasons"] == ["a", "b", "c"] + + +def test_a_factory_is_a_detail_only_when_each_return_is_one_and_each_call_is_sent(): + factory = """ +def quota(e) -> dict: + return {"code": "quota", "reason": "files"} """ - 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")') == [] + sent = "def f(e):\n raise HTTPException(422, detail=quota(e))\n" + assert _scan(a=factory + sent)["codes"]["quota"]["reasons"] == ["files"] + for change in [ + "def g(e):\n d = quota(e)\n d['code'] = 'other'\n raise HTTPException(422, d)\n", + "def g(e):\n raise HTTPException(422, detail=quota(e).copy())\n", + "pick = quota\n", + ]: + r = _scan(a=factory + sent + change) + assert r["unreadCodes"] and "quota" not in r["codes"], change + twice = _scan(a="quota = None\n" + factory + sent) # the def is the last of two bindings + assert twice["unreadCodes"] and "quota" not in twice["codes"] + for other in [ + " if e:\n return None\n", + " if e:\n return cached\n", + " if e:\n return {'code': 'quota', 'reason': 'x'} if e > 1 else None\n", + " log({'code': 'quota', 'reason': 'logged'})\n", # a detail it does not return + ]: + bent = factory.replace("-> dict:\n", "-> dict:\n" + other) + assert _scan(a=bent + sent)["unreadCodes"], other + method = "class C:\n def quota(self):\n return {'code': 'quota', 'reason': 'm'}\n" + assert _scan(a=method)["unreadCodes"] + dotted = _scan( + a=factory, + routes__b="import app.a\n\ndef g(e):\n raise HTTPException(422, detail=app.a.quota(e))\n", + ) + assert dotted["unreadCodes"] and "quota" not in dotted["codes"] -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(): +def test_a_name_a_lambda_a_comprehension_a_nonlocal_or_a_global_binds_is_unread(): 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} - detail["email"] = "x" - return HTTPException(status, detail) - - def g(x): - detail = {"code": "login"} if x else {"code": "login", "reason": "other"} - detail["reason"] = "late" - return HTTPException(401, detail=detail) + def f(svc): + why = "invalid" + make = lambda why: HTTPException(401, {"code": "login", "reason": why}) + raise make(svc.state) + def g(svc): + why = "invalid" + raise [HTTPException(401, {"code": "login", "reason": why}) for why in svc.all][0] + def h(svc): + why = "invalid" + def inner(): + nonlocal why + why = svc.reason + inner() + raise HTTPException(401, {"code": "login", "reason": why}) + def i(svc): + global why + why = "invalid" + raise HTTPException(401, {"code": "login", "reason": why}) """ ) - assert r["unreadCodes"] == [] - assert _code(r, "login")["reasons"] == ["invalid", "late", "other", "suspended"] - - -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")', + assert _code(r, "login")["reasons"] == [] + assert sorted(_unread(r, "login")) == [ + "why (bound by a lambda or a comprehension)", + "why (bound by a lambda or a comprehension)", + "why (named by a nonlocal or a global)", + "why (named by a nonlocal or a global)", ] - 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] + + +def test_a_local_that_reaches_itself_is_unread(): + r = _scan( + a=""" + def f(x): + why = "a" + if x: + why = why if x > 1 else "b" + raise HTTPException(409, {"code": "edit", "reason": why}) + def g(x): + a = "z" + b = a + a = b + raise HTTPException(409, {"code": "playlist", "reason": a}) + def h(x): + why = "c" + if x: + raise HTTPException(409, {"code": "busy", "reason": why}) + raise HTTPException(409, {"code": "busy", "reason": why}) + """ ) + assert _code(r, "busy")["reasons"] == ["c"] and _code(r, "busy")["unread"] == [] + assert _code(r, "edit")["reasons"] == ["a", "b"] + assert _unread(r, "edit") == ["why (reached again while it is resolved)"] + assert _code(r, "playlist")["reasons"] == ["z"] + assert any("reached again while it is resolved" in u for u in _unread(r, "playlist")) + + +def test_a_module_constant_or_a_map_changed_elsewhere_is_unread(): + def reasons(urlguard: str, route: str = "", **more: str) -> tuple[list[str], list[str]]: + r = _scan( + **more, + urlguard="REASON = 'malformed'\n_MAP = {'a': 'map_a'}\n" + urlguard, + routes__a="from app import urlguard\n" + + route + + "\ndef f(k: Literal['a']):\n" + + " raise HTTPException(409, {'code': 'url_blocked', 'reason': urlguard.REASON})\n" + + " raise HTTPException(409, {'code': 'url_blocked', 'reason': urlguard._MAP[k]})\n", + ) + return _code(r, "url_blocked")["reasons"], _unread(r, "url_blocked") + + assert reasons("def get(k):\n return _MAP.get(k)\n") == (["malformed", "map_a"], []) + other = "from app import other\ndef g():\n other.REASON = 'x'\n" # another module's REASON + assert reasons("", other, other="REASON = 'y'\n") == (["malformed", "map_a"], []) + shadow = ( + "def h(*urlguard):\n" + " raise HTTPException(409, {'code': 'url_blocked', 'reason': urlguard.REASON})\n" + ) + assert reasons("", shadow)[1] == ["urlguard.REASON"] + for urlguard, route in [ + ("def reset():\n global REASON\n REASON = 'x'\n", ""), + ("", "def g():\n urlguard.REASON = 'x'\n"), + ("", "def g():\n del urlguard.REASON\n"), + ]: + assert reasons(urlguard, route)[0] == ["map_a"], (urlguard, route) + for urlguard in [ + "_MAP['a'] = compute()\n", + "def init(x):\n _MAP['a'] = x\n", + "_MAP.update(a='other')\n", + "def drop():\n del _MAP['a']\n", + "register(_MAP)\n", + ]: + got, unread = reasons(urlguard) + assert got == ["malformed"] and "(the map is changed after it is built)" in unread[0] + + +def test_a_helper_bound_twice_is_not_followed(): + r = _scan( + refusal="refused = None\n" + textwrap.dedent(REFUSAL), + routes__a="from app.refusal import refused\n\ndef f():\n raise refused(409, 'lost')\n", + ) + assert _code(r, "refused")["reasons"] == [] + assert _unread(r, "refused")[0].endswith("refused(reason) is not a module-level helper") + + +def test_a_dotted_path_through_an_app_module_that_goes_nowhere_is_unread(): + r = scan( + { + "__init__.py": "", + "refusal.py": textwrap.dedent(REFUSAL), + "routes/a.py": "import app as a\n\ndef f():\n raise a.refusal.refused(409, 'dotted')\n", + } + ) + assert _unread(r, "refused") == ["a.refusal.refused (a reference the scan cannot resolve)"] + + +def test_a_helper_reached_through_an_app_module_that_does_not_bind_it_is_unread(): + for shim in [ + "from app.refusal import *\n", + "try:\n from app.refusal import refused\nexcept ImportError:\n refused = None\n", + ]: + r = _scan( + refusal=REFUSAL, + shim=shim, + routes__a="from app.shim import refused\n\ndef f():\n raise refused(409, 'hidden')\n", + ) + assert _unread(r, "refused") == ["refused (a reference the scan cannot resolve)"], shim + package = scan( + { + "pkg/refusal.py": textwrap.dedent(REFUSAL), + "pkg/__init__.py": "from .refusal import refused\n", + "routes/a.py": "from app.pkg import refused\n\ndef f():\n raise refused(409, 'hidden')\n", + } + ) + assert _unread(package, "refused") == ["refused (a reference the scan cannot resolve)"] def test_the_data_files_are_excluded_by_name_only():