Compare commits

..
10 Commits
Author SHA1 Message Date
peter c430558ab3 release: 0.114.3 (improvement/e49-refusal-contract) 2026-11-01 00:30:51 +01:00
peter be38d652a5 chore(release): 0.114.3 notes 2026-11-01 00:18:59 +01:00
peter 31a75f4985 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.
2026-10-31 23:54:55 +01:00
peter f8a8e905da fix(i18n): read refusals along a closed list of shapes, report the rest
The scan followed many shapes and called the rest unread, but each shape it learned opened a silent
branch: a parameter rebound in the body, a second visit of a followed helper, an `import app.x`, a name
bound twice or under an `if`, a spread or a store that replaces the code, a detail rebuilt from its
variable, a None reason. It now reads only what the app writes and reports everything else: a
branch the app never takes is gone (an `or`, a default, a base class's attribute, string and union
annotations, a bare module constant), a spread is clean only from an untouched `**kwargs` no call puts
a `code` in, and the detail's variable may only grow by `|=` or `["reason"]`. The real tree reads
exactly as before; every exit and every kept branch has a self-test that a mutant of it fails.

The detail test also fails a dict without a `code`, a JSONResponse sentence and a detail-less
HTTPException; the two that are no refusal (the setup gate's 503, an unmatched api path's 404) are
listed with why. The contract's assertion that compared the reason map with itself is gone.
2026-10-31 18:45:02 +01:00
peter 5e8589bbf8 fix(i18n): read every refusal from the backend's syntax tree, not by regex
The contract read Python with regexes, and each guard added for a missed shape had shapes of its own
it missed: a reason after another key, a status passed by name, a second `e.reason` handler with the
same text, a pin naming a type nothing checked.

backend/tests/refusal_sites.py (stdlib `ast`, run by the contract on the host's Python) reads every
`{"code": ...}` detail and follows each reason to its values: literals, both arms of a conditional,
module constants, the `Literal` on the parameter or attribute it travels in (on the class an `except`
catches), every call of a `str` helper with its defaults, a map's values with its keys held to the
index's type. What it cannot follow fails the contract, unless no screen shows that code; no pins.

The reason attributes are typed with their Literal, so a subclass or an assignment cannot widen
them; InvalidSetting takes a SettingReason, and the settings map is held to it (a reason the map
lacked was a 500). The per-family reverse checks are one check by prefix (the admin actions
included), and the leaf-key filter applies only to the suffixed sign-in family.
2026-10-31 15:07:38 +01:00
peter ceadf69e43 fix(i18n): hold every refusal reason by type, and refuse what the scan cannot read
Review round 1 of the refusal contract: it claimed more than it held.

- The other reason carriers get their Literal too (BlockedUrl, ClipError,
  FolderError, the playlist and sign-in refusal helpers), so the type
  checker refuses a reason passed by variable or an unknown literal, and
  the contract reads one shape.
- A literal reason in a detail dict built anywhere counts as a reason of
  its code; a code the scan cannot read, a refused() whose reason it
  cannot read, and a new reason passed by variable each fail the contract
  until listed with what holds them.
- A sign-in refusal key the backend no longer sends is caught too.
- localizeDetail's numbers (download cap in GB, upload cap in bytes) and
  the shared-sentence fallbacks are tested in both languages.
- One Literal parser in the contract; a stale comment fixed.
2026-10-31 01:00:26 +01:00
peter 4379f76747 feat(i18n): every server refusal is said by a record the contract holds
A refusal code or reason with no sentence fell through to the generic
"couldn't carry out that action" unnoticed: the parity test is green when
EN and HU miss the same key. One did (an edit refused for a file that is
not video or audio).

- lib/api/refusalCodes.ts: which sentence answers each code, as data;
  localizeDetail says a refusal through it, unchanged for every reader.
- backendIds contract: every detail code the backend sends is in exactly
  one record, every reason it can carry has its sentence in EN and HU,
  and no family names a reason the backend cannot send.
- The reasons that travel in a variable are Literal types (edit, move,
  cookies, quota), so the type checker refuses an unknown one too.
- downloads.errors.edit.not_media in EN and HU.
2026-10-30 23:36:25 +01:00
peter b3be568e36 release: 0.114.2 (bug/joserfc-cve) 2026-10-30 22:39:07 +01:00
peter a8cbc7b338 chore(release): 0.114.2 2026-10-30 22:28:39 +01:00
peter cf8dc80428 fix(deps): joserfc 1.7.4 -> 1.7.6 for CVE-2026-119283 and CVE-2026-119284
One pin bumped in place (Authlib's dependency); pip check passes and pip-audit finds nothing.
2026-10-30 22:28:39 +01:00
29 changed files with 2176 additions and 332 deletions
+12
View File
@@ -7,6 +7,18 @@ edit the release notes and re-run the generator.
Upgrading, pinning a version and rolling back: [docs/self-hosting.md](docs/self-hosting.md#upgrading).
## 0.114.3 — 2026-11-01
### Fixed
- Editing a file that is neither video nor audio now says why it can't be done, instead of a generic error.
## 0.114.2 — 2026-10-30
### Fixed
- Security: a library the sign-in relies on is updated, closing two published vulnerabilities.
## 0.114.1 — 2026-10-30
### Fixed
+1 -1
View File
@@ -105,7 +105,7 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory
| Ownership check | `userscope.owned_by` (a filter) / `owned_row` (a row by id): a unit's `queries.own_x` returns None and its route answers `not_found`; a core route `owned_or_404`. A composite rule (own or shared, built-in or own) says so in its `queries` | an inline owner filter — `Model.user_id == …`, an alias's, `.in_()`, `filter_by(user_id=…)` (fence-held, every module) — or a `db.get` and a compare where `owned_row` fits |
| Admin gate import | `app.auth` | the `app.routes.admin` re-export |
| Upstream failure status | `422` with JSON detail | `502` (reserve 502 for raw byte-proxies) |
| User-facing error | structured `detail` (`{code, reason}`) | a bare English string under a bilingual UI |
| User-facing error | structured `detail` (`{code, reason}`), its sentence named in `lib/api/refusalCodes.ts` (contract-held) | a bare English string under a bilingual UI |
| Mutation response | `{"ok": true, …}` (`{"deleted": id}`), test-held | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok` |
| Transaction commit | the handler commits; a unit's `service` and every helper `flush` (fence-held, every module) | a helper that commits — only a listed one (what owns its session, or a core step that must land whatever its caller does next), each with its reason: the fence's `UNIT_COMMITS` / `CORE_COMMITS`, a route's `COMMITTING_HELPERS` (`test_route_conventions`) |
| Admin knob storage | `sysconfig` (values) | env / `app_state` / a table of its own (the job intervals had one) |
+1 -1
View File
@@ -1 +1 @@
0.114.1
0.114.3
+18 -2
View File
@@ -22,6 +22,7 @@ import json
import math
import re
from pathlib import Path
from typing import Literal
from app.downloads.edit import AAC_ARGS, EVEN_FRAME, INPUT_GUARD, X264_ARGS
from app.plex.sync import HDR_TRANSFERS
@@ -57,12 +58,27 @@ _CROP_ASPECTS = (0.25, 4.0)
_EDGE_SLACK = 1e-3
# Every reason a clip is refused; the UI names each one (`clip.errors.<reason>`, held by the frontend's
# backendIds contract, which reads this Literal).
ClipReason = Literal[
"bad_source",
"bad_range",
"too_short",
"too_long",
"bad_audio",
"bad_crop",
"not_found",
"media_missing",
"past_end",
]
class ClipError(Exception):
"""A clip request that cannot be cut; `reason` is an i18n-mappable key, `params` the numbers its
message names (the cap itself, so the client never says a stale one)."""
def __init__(self, reason: str, **params: float):
self.reason = reason
def __init__(self, reason: ClipReason, **params: float):
self.reason: ClipReason = reason
self.params = params
super().__init__(reason)
+10
View File
@@ -25,6 +25,7 @@ Security controls enforced here:
import os
from pathlib import Path
from typing import Literal
from sqlalchemy import select
from sqlalchemy.orm import Session
@@ -163,9 +164,18 @@ def get_credential(db: Session, user_id: int, host_key: str) -> DownloadCredenti
).scalar_one_or_none()
# Every reason a pasted cookies.txt is refused; the UI names each one (`downloads.errors.cookies.<reason>`,
# held by the frontend's backendIds contract, which reads this Literal).
CookieReason = Literal["empty", "format"]
class InvalidCookies(ValueError):
"""The pasted cookies aren't a usable Netscape cookies.txt — surfaced as a 400 by the route."""
def __init__(self, reason: CookieReason):
super().__init__(reason)
self.reason: CookieReason = reason
def validate_cookies(raw: str) -> str:
"""Sanity-check a pasted cookies.txt (Netscape format) and return the cleaned text. yt-dlp reads
+13 -2
View File
@@ -24,6 +24,7 @@ kept it — and its `ready_at`, the moment it finished (`stamp_done`).
from datetime import datetime, timedelta, timezone
from enum import StrEnum
from typing import Literal
from urllib.parse import urlsplit
from sqlalchemy import select
@@ -181,12 +182,22 @@ def stamp_done(db: Session, job: DownloadJob) -> None:
job.expiry_warned = False
# Every folder change the rules refuse, answered as a generic refusal; the UI names each one
# (`errors.refused.<reason>`, held by the frontend's backendIds contract, which reads this Literal).
FolderReason = Literal[
"file_folder_gone",
"file_folder_name_taken",
"file_folder_not_its_source",
"file_folder_not_own",
]
class FolderError(Exception):
"""A folder change the rules refuse; `reason` is the refusal the route answers with."""
def __init__(self, reason: str) -> None:
def __init__(self, reason: FolderReason) -> None:
super().__init__(reason)
self.reason = reason
self.reason: FolderReason = reason
def own_folder(db: Session, user_id: int, folder_id: int, *, hold: bool = True) -> FileFolder:
+10 -4
View File
@@ -20,6 +20,7 @@ their output's real size before it is placed. Edit jobs do not consult it at all
"""
from dataclasses import dataclass
from typing import Literal
from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session
@@ -50,11 +51,16 @@ class QuotaLimits:
unlimited: bool
class QuotaExceeded(Exception):
"""Raised by check_enqueue; `reason` is one of max_jobs|max_bytes (i18n key on the client)."""
# Every cap a download can hit; the UI names each one (`downloads.errors.quota.<reason>`, held by the
# frontend's backendIds contract, which reads this Literal).
QuotaReason = Literal["max_jobs", "max_bytes"]
def __init__(self, reason: str, limit: int, current: int):
self.reason = reason
class QuotaExceeded(Exception):
"""Raised by check_enqueue; `reason` names the cap it hit (an i18n key on the client)."""
def __init__(self, reason: QuotaReason, limit: int, current: int):
self.reason: QuotaReason = reason
self.limit = limit
self.current = current
super().__init__(f"download quota exceeded: {reason} ({current}/{limit})")
+10 -4
View File
@@ -10,6 +10,7 @@ handler); the folder, link and quota writes are `folders`, `links` and `quota`.
import uuid
from datetime import datetime, timezone
from typing import Literal
from sqlalchemy import delete as delete_rows
from sqlalchemy import func, select, update
@@ -39,11 +40,16 @@ from app.plex import paths as plex_paths
from app.userscope import owned_by, owned_row
class EditError(Exception):
"""Bad edit request (empty spec / source not ready); `reason` is an i18n-mappable key."""
# Every reason an edit is refused; the UI names each one (`downloads.errors.edit.<reason>`, held by
# the frontend's backendIds contract, which reads this Literal).
EditReason = Literal["source_not_ready", "not_media", "empty_edit"]
def __init__(self, reason: str):
self.reason = reason
class EditError(Exception):
"""Bad edit request (empty spec / source not ready / not media); `reason` names the sentence."""
def __init__(self, reason: EditReason):
self.reason: EditReason = reason
super().__init__(reason)
+17 -8
View File
@@ -33,24 +33,33 @@ user-controlled destination to guard.
import ipaddress
import socket
from typing import Literal
from urllib.parse import urlparse
# Reason codes travel to the client inside the structured 422 detail, so they are part of the API
# vocabulary (frontend `localizeDetail`) — keep them stable.
REASON_SCHEME = "scheme" # not http(s)
REASON_NO_HOST = "no_host" # URL carries no hostname
REASON_MALFORMED = "malformed" # unparseable authority or port; urlparse raises late, so guard it
REASON_UNRESOLVABLE = "unresolvable" # DNS said nothing (honest: not the same as "blocked")
REASON_PRIVATE = "private" # resolves into a network the server must not fetch from
# vocabulary — keep them stable. The UI names each one (`downloads.errors.urlBlocked.<reason>`, held by
# the frontend's backendIds contract, which reads this Literal).
UrlBlockedReason = Literal["scheme", "no_host", "malformed", "unresolvable", "private"]
REASON_SCHEME: UrlBlockedReason = "scheme" # not http(s)
REASON_NO_HOST: UrlBlockedReason = "no_host" # URL carries no hostname
REASON_MALFORMED: UrlBlockedReason = (
"malformed" # unparseable authority or port; urlparse raises late
)
REASON_UNRESOLVABLE: UrlBlockedReason = (
"unresolvable" # DNS said nothing (not the same as "blocked")
)
REASON_PRIVATE: UrlBlockedReason = (
"private" # resolves into a network the server must not fetch from
)
class BlockedUrl(Exception):
"""A user-supplied URL that must not be fetched server-side. ``reason`` is one of the
REASON_* codes above."""
def __init__(self, reason: str):
def __init__(self, reason: UrlBlockedReason):
super().__init__(reason)
self.reason = reason
self.reason: UrlBlockedReason = reason
def _is_forbidden(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+8 -3
View File
@@ -9,7 +9,7 @@ unique key, a row lock taken before a presence check. The comfort shelf's writes
from datetime import datetime, timezone
from typing import cast
from typing import Literal, cast
from sqlalchemy import Table, bindparam, delete, exists, func, select, text, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
@@ -301,12 +301,17 @@ def set_progress(db: Session, user: User, video_id: str, position: int) -> None:
# --- playlists -------------------------------------------------------------------------------
# Every reason a move is refused; the UI names each one (`playlists.moveErrors.<reason>`, held by the
# frontend's backendIds contract, which reads this Literal).
MoveReason = Literal["same_playlist", "not_in_playlist"]
class MoveRefused(Exception):
"""A move between playlists that cannot happen: `reason` is the code the UI translates."""
def __init__(self, reason: str):
def __init__(self, reason: MoveReason):
super().__init__(reason)
self.reason = reason
self.reason: MoveReason = reason
def touch(pl: Playlist) -> None:
+19 -9
View File
@@ -7,23 +7,33 @@ whatever sentence the route happened to hand-write. `refused` is the generic fam
(`errors.refused.<reason>`, 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,
and the frontend's contract test requires every `refused(…, "<reason>")` to be named in both
languages — so the rule is enforced, not described.
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.
"""
from fastapi import HTTPException
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."""
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 frontend's
# contract test can read every reason this can produce.
_SETTING_REASONS = {
# `sysconfig.InvalidSetting.reason` → the refusal it answers as. A dict of literals, so the refusal
# scan reads every reason this can produce, and holds its keys to `SettingReason`.
_SETTING_REASONS: dict[SettingReason, str] = {
"not_a_number": "setting_not_a_number",
"too_low": "setting_too_low",
"too_high": "setting_too_high",
@@ -32,6 +42,6 @@ _SETTING_REASONS = {
}
def setting_refused(exc) -> HTTPException:
def setting_refused(exc: InvalidSetting) -> HTTPException:
"""A value the admin entered that `sysconfig.set_value` refused (`exc`: its `InvalidSetting`)."""
return refused(400, _SETTING_REASONS[exc.reason], key=exc.key, **exc.params)
+19 -10
View File
@@ -976,7 +976,16 @@ def _resend_verify_if_eligible(email: str) -> None:
send_verification_link(db, user)
def _login_refused(status: int, reason: str, user: User | None = None, **extra) -> HTTPException:
# Every refusal a password sign-in answers with; the UI names each one (`welcome.auth.blocked.<reason>`,
# held by the frontend's backendIds contract, which reads this Literal).
LoginRefusal = Literal[
"rate_limited", "invalid", "suspended", "email_unverified", "pending_approval"
]
def _login_refused(
status: int, reason: LoginRefusal, user: User | None = None, **extra
) -> HTTPException:
"""Every refusal password_login can answer with, in the codebase's structured `{code, reason}`
shape so the SPA can say it in the user's own language (the old bare English strings were
unreadable under a bilingual UI, and unusable as anything but a one-line error).
@@ -992,15 +1001,15 @@ def _login_refused(status: int, reason: str, user: User | None = None, **extra)
`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")
+4 -3
View File
@@ -208,7 +208,7 @@ def set_credential(
try:
cleaned = dlcreds.validate_cookies(body.cookies)
except dlcreds.InvalidCookies as e:
raise HTTPException(status_code=400, detail={"code": "cookies", "reason": str(e)}) from e
raise HTTPException(status_code=400, detail={"code": "cookies", "reason": e.reason}) from e
dlcreds.upsert_credential(db, user.id, host_key, cleaned)
audit.record(
db,
@@ -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()
+17 -2
View File
@@ -3,7 +3,7 @@ CRUD (create / rename / delete, add / remove / reorder items) plus YouTube two-w
mirror the user's own YouTube playlists in (`/sync-youtube`) and push/revert local edits back
out (`/{id}/push-plan`, `/{id}/revert-youtube`), behind the optional write scope."""
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Literal
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
@@ -53,7 +53,22 @@ def _check_pull_cooldown(user: User) -> None:
raise HTTPException(status_code=429, detail={"code": "sync_cooldown"})
def _refused(status: int, reason: str) -> HTTPException:
# Every playlist action the server refuses; the UI names each one (`playlists.errors.<reason>`, held by
# the frontend's backendIds contract, which reads this Literal).
PlaylistRefusal = Literal[
"name_required",
"not_on_youtube",
"quota_exceeded",
"sync_all_busy",
"sync_busy",
"watch_later_not_pushable",
"watch_later_undeletable",
"youtube_editing_off",
"youtube_read_scope",
]
def _refused(status: int, reason: PlaylistRefusal) -> HTTPException:
"""A playlist action the server refuses, as a reason the UI translates
(`playlists.errors.<reason>`) — a bare English detail reached the Hungarian UI verbatim."""
return HTTPException(status_code=status, detail={"code": "playlist", "reason": reason})
+7 -2
View File
@@ -443,15 +443,20 @@ def _default(s: ConfigSpec):
return getattr(settings, s.default_attr)
# Every reason `set_value` refuses with. `refusal.setting_refused` looks each one up in its map, which
# the refusal scan holds to exactly these keys: a reason the map lacks would answer a 500.
SettingReason = Literal["not_a_number", "too_low", "too_high", "not_a_choice", "no_encryption_key"]
class InvalidSetting(ValueError):
"""A value `set_value` refuses, as a reason the UI translates (`errors.refused.setting_<reason>`)
plus what the sentence needs — the English message is for logs only. A ValueError still, so a
caller that only needs "refused or not" keeps working."""
def __init__(self, key: str, reason: str, **params: object):
def __init__(self, key: str, reason: SettingReason, **params: object):
super().__init__(f"{key}: {reason} {params}")
self.key = key
self.reason = reason
self.reason: SettingReason = reason
self.params = params
+2 -2
View File
@@ -7,7 +7,7 @@
# `pip install -r requirements.lock && pip check` must pass. In place so far: urllib3 2.7.0 -> 2.8.0 (PYSEC-2026-4175/4176/4177); yt-dlp 2026.7.4 -> 2026.8.19
# (it marks the clients that need a stream PO token, so the downloader stops fetching into a 403);
# certifi 2026.7.22 -> 2026.10.19 (GHSA-6q48-6c7w-9gw2); brotli 1.2.0 -> 1.2.1 (PYSEC-2026-4322);
# Authlib 1.7.2 -> 1.9.1 (CVE-2026-96760, CVE-2026-114616).
# Authlib 1.7.2 -> 1.9.1 (CVE-2026-96760, CVE-2026-114616); joserfc 1.7.4 -> 1.7.6 (CVE-2026-119283, CVE-2026-119284).
APScheduler==3.11.3
Authlib==1.9.1
Mako==1.4.1
@@ -39,7 +39,7 @@ httptools==0.8.0
httpx==0.28.1
idna==3.18
itsdangerous==2.2.0
joserfc==1.7.4
joserfc==1.7.6
markdown-it-py==4.2.0
mdurl==0.1.2
mutagen==1.48.1
+738
View File
@@ -0,0 +1,738 @@
"""Every refusal the backend can answer, read from its syntax tree: each `{"code": …, "reason": …}` detail,
with every reason its code can be sent with.
The frontend's contract (`i18n/backendIds.contract.test.ts`) runs this and requires a sentence for each
code and reason in both languages. A scan that tries to follow every shape opens a silent gap with each one
it learns, and following how a value flows through Python (variables, scopes, aliases, a dict changed in
place) never closes. So the CODE keeps to a narrow form, and this reads a CLOSED list of it, reporting
everything else, never guessing:
- where a detail stands: where it is sent — `HTTPException`'s second argument or `detail=`, the `"detail"`
value of a `JSONResponse`'s content dict — through the arms of `a if c else b`, or as what a module-level
factory returns when every return of it is a detail and every reference to it is a call sent as one. A
detail held in a variable is not followed;
- the code: one literal. A `**` spread, or a key that is not a literal, stands BEFORE a literal `code` and
`reason`, both of which the detail then carries: the later key wins, so no spread can change them;
- a reason: a string literal (`None` sends none: the site counts as reasonless); both arms of
`a if c else b`; a function's local bound only by plain `x = …`; a module constant reached as `mod.NAME`;
a call's `Literal` return annotation;
- a parameter typed with a `Literal` (basedpyright holds every value passed in), or a `str` one of a
module-level helper (`refused`, `not_found`), followed to every call, each of which must pass it by
position or by name; a parameter its body binds again is unread;
- an attribute: `e.x` of the exception an `except C as e` catches, or of a parameter typed `C`, when a
method of `C` declares `self.x: <Literal>`;
- a map lookup (`_SETTING_REASONS[exc.reason]`): the map's values, and its keys must be exactly what the
index can be.
A name bound twice at module level, inside a module-level `if`/`try`/loop, named by a `global`, or imported
from two places is not looked up; nor is a local a lambda or a comprehension binds, or one a `nonlocal` or
`global` names. A module constant or map that anything stores into, deletes from or (a map) uses but by
`map[k]` or a read method is unread. A reference named like a followed helper that the scan cannot resolve
(`import app.x`, a relative import, an app module that does not bind it) is unread. Anything else is UNREAD,
or an unread code, and the contract fails on it unless no screen shows that code. Stdlib only: it runs on
the host, beside the frontend's tests (`python -I backend/tests/refusal_sites.py` → JSON).
"""
import ast
import json
import pathlib
import sys
from collections.abc import Iterator
from dataclasses import dataclass, field
APP = pathlib.Path(__file__).resolve().parents[1] / "app"
# Files whose `"code"` dicts are data, not refusals: a subtitle track's status payload, TMDB country codes.
NOT_REFUSALS = {
"plex/subtitles.py": "a subtitle track's status payload (its own `code`)",
"plex/tmdb.py": "a production country's ISO code",
}
# What a map may be used by besides `map[k]`: a method that only reads it.
READS = {"get", "items", "keys", "values"}
Node = ast.AST
FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef)
SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)
COMPREHENSIONS = (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)
COMPOUND = (
ast.If,
ast.Try,
ast.TryStar,
ast.For,
ast.AsyncFor,
ast.While,
ast.With,
ast.AsyncWith,
ast.Match,
)
@dataclass
class Res:
values: set[str] = field(default_factory=set)
unread: list[str] = field(default_factory=list)
none: bool = False # an arm sends no reason
def __or__(self, other: "Res") -> "Res":
return Res(self.values | other.values, self.unread + other.unread, self.none or other.none)
@dataclass
class Module:
path: str # relative to app/, posix
name: str # dotted, `app.…`
tree: ast.Module
parents: dict[Node, Node]
# local name → (module, symbol or None for the module)
imports: dict[str, tuple[str, str | None]]
top: dict[str, Node] # module-level bindings: def, class, or the assignment
ambiguous: set[
str
] # bound more than once, conditionally, by a `global`, or imported from two places
slots: set[int] # the ids of the nodes sent as a detail
def _own_scope(node: Node) -> Iterator[Node]:
"""`node` and what it holds in its own scope: a nested def, class or lambda is itself, not its body."""
stack = [node]
while stack:
n = stack.pop()
yield n
if n is not node and isinstance(n, SCOPES):
continue
stack.extend(ast.iter_child_nodes(n))
def _stored(target: Node) -> set[str]:
"""The names an assignment target binds, not the ones it reads (`LABELS[k] = v` binds none)."""
return {
n.id for n in ast.walk(target) if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store)
}
def _binds(node: Node) -> set[str]:
"""The names one node binds in the scope it sits in."""
if isinstance(node, ast.Assign):
return set().union(*(_stored(t) for t in node.targets))
if isinstance(node, (ast.AnnAssign, ast.AugAssign, ast.For, ast.AsyncFor, ast.NamedExpr)):
return _stored(node.target)
if isinstance(node, (ast.With, ast.AsyncWith)):
return set().union(*(_stored(i.optional_vars) for i in node.items if i.optional_vars))
if isinstance(node, (ast.Import, ast.ImportFrom)):
return {(a.asname or a.name).split(".")[0] for a in node.names}
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
return {node.name}
if isinstance(node, (ast.MatchAs, ast.MatchStar, ast.ExceptHandler)) and node.name:
return {node.name}
return set()
def _sender(call: ast.Call) -> str | None:
return getattr(call.func, "id", None) or getattr(call.func, "attr", None)
def detail_slots(tree: Node) -> Iterator[tuple[ast.Call, Node | None]]:
"""Every place a detail is sent from, with what it holds: `HTTPException`'s second argument or
`detail=` (None: sent without one), and the `"detail"` value of a `JSONResponse`'s content dict (a
content without one is an answer, not a refusal)."""
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
kw = {k.arg: k.value for k in node.keywords}
if _sender(node) == "HTTPException":
yield node, kw.get("detail", node.args[1] if len(node.args) > 1 else None)
elif _sender(node) == "JSONResponse":
content = kw.get("content", node.args[0] if node.args else None)
if isinstance(content, ast.Dict):
for k, v in zip(content.keys, content.values, strict=True):
if isinstance(k, ast.Constant) and k.value == "detail":
yield node, v
def _code_dict(node: Node) -> bool:
"""A `{"code": …}` dict, or a `dict(code=…)` call."""
if isinstance(node, ast.Dict):
return any(isinstance(k, ast.Constant) and k.value == "code" for k in node.keys)
return (
isinstance(node, ast.Call)
and getattr(node.func, "id", None) == "dict"
and any(k.arg == "code" for k in node.keywords)
)
def _module_name(path: str) -> str:
parts = path.removesuffix(".py").split("/")
if parts[-1] == "__init__":
parts = parts[:-1]
return ".".join(["app", *parts])
def _load(sources: dict[str, str]) -> dict[str, Module]:
mods: dict[str, Module] = {}
for path, src in sources.items():
tree = ast.parse(src)
parents = {child: node for node in ast.walk(tree) for child in ast.iter_child_nodes(node)}
origins: dict[str, set[tuple[str, str | None]]] = {}
ambiguous: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
for a in node.names:
origins.setdefault(a.asname or a.name, set()).add((node.module, a.name))
elif isinstance(node, ast.Import):
for a in node.names:
if a.asname:
origins.setdefault(a.asname, set()).add((a.name, None))
elif isinstance(node, ast.Global):
ambiguous |= set(node.names)
top: dict[str, Node] = {}
count: dict[str, int] = {}
ambiguous |= {name for name, o in origins.items() if len(o) > 1}
for stmt in tree.body:
if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
top[stmt.name] = stmt
elif isinstance(stmt, ast.Assign):
for t in stmt.targets:
if isinstance(t, ast.Name):
top[t.id] = stmt
elif isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name):
top[stmt.target.id] = stmt
for n in [stmt] if isinstance(stmt, SCOPES) else _own_scope(stmt):
for name in _binds(n):
count[name] = count.get(name, 0) + 1
if isinstance(stmt, COMPOUND):
ambiguous.add(name)
ambiguous |= {name for name, c in count.items() if c > 1}
imports = {name: next(iter(o)) for name, o in origins.items() if name not in ambiguous}
slots = {id(d) for _, d in detail_slots(tree) if d is not None}
name = _module_name(path)
mods[name] = Module(path, name, tree, parents, imports, top, ambiguous, slots)
return mods
class Scan:
def __init__(self, sources: dict[str, str]):
self.mods = _load(sources)
self.errors: list[str] = []
self._following: set[tuple[str, str, str]] = set()
self._followed: dict[tuple[str, str, str], Res] = {}
self._resolving: set[int] = set()
self._factory: dict[int, bool] = {}
self._changed: dict[int, bool] = {}
# --- names ---------------------------------------------------------------------------------------
def lookup(
self, m: Module, name: str, seen: frozenset[tuple[str, str]] = frozenset()
) -> tuple[Module, Node] | tuple[str, None] | None:
"""What a module-level name is: (its module, its binding), or (a module's name, None)."""
if name in m.ambiguous:
return None
if name in m.top:
return m, m.top[name]
if name in m.imports:
mod, sym = m.imports[name]
if sym is None:
return (mod, None) if mod in self.mods else None
if f"{mod}.{sym}" in self.mods:
return f"{mod}.{sym}", None
if mod in self.mods and (mod, sym) not in seen:
return self.lookup(self.mods[mod], sym, seen | {(mod, sym)})
return None
def lookup_expr(self, m: Module, e: Node) -> tuple[Module, Node] | tuple[str, None] | None:
if isinstance(e, ast.Name):
return None if self.binder(m, e) is not None else self.lookup(m, e.id)
if isinstance(e, ast.Attribute):
base = self.lookup_expr(m, e.value)
if base and base[1] is None and isinstance(base[0], str):
return self.lookup(self.mods[base[0]], e.attr)
return None
def binder(self, m: Module, e: ast.Name) -> Node | None:
"""The innermost scope `e` is bound in: a function (a parameter, or a binding in its own body), a
lambda (a parameter), a comprehension (its target); None: a module-level name."""
node = m.parents.get(e)
while node is not None:
if isinstance(node, (*FUNCS, ast.Lambda)):
a = node.args
params = [*a.posonlyargs, *a.args, *a.kwonlyargs, a.vararg, a.kwarg]
if e.id in {p.arg for p in params if p}:
return node
if isinstance(node, FUNCS) and self.assignments(node, e.id):
return node
elif isinstance(node, COMPREHENSIONS) and any(
e.id in _stored(g.target) for g in node.generators
):
return node
node = m.parents.get(node)
return None
def at(self, m: Module, e: Node) -> str:
return f"{m.path}:{getattr(e, 'lineno', '?')}: {ast.unparse(e)}"
def enclosing(self, m: Module, e: Node, kind: type | tuple[type, ...]) -> Node | None:
node = m.parents.get(e)
while node is not None and not isinstance(node, kind):
node = m.parents.get(node)
return node
def assignments(self, fn: Node, name: str) -> list[Node]:
"""What binds `name` in this function's own body (not in a nested one)."""
return [n for n in _own_scope(fn) if n is not fn and name in _binds(n)]
def caught_as(self, m: Module, e: Node, name: str) -> ast.ExceptHandler | None:
node = m.parents.get(e)
while node is not None and not isinstance(node, FUNCS):
if isinstance(node, ast.ExceptHandler) and node.name == name:
return node
node = m.parents.get(node)
return None
def references(
self, m: Module, target: Node
) -> Iterator[tuple[Module, Node, ast.Call | None, bool]]:
"""Every reference to a module-level def or class: the call it makes (None: not a call), and
whether it resolved. A reference named like it that the scan cannot resolve is yielded too."""
name = getattr(target, "name", None)
for cm in self.mods.values():
for node in ast.walk(cm.tree):
if not isinstance(node, (ast.Name, ast.Attribute)) or not isinstance(
node.ctx, ast.Load
):
continue
if (node.id if isinstance(node, ast.Name) else node.attr) != name:
continue
parent = cm.parents.get(node)
if isinstance(parent, ast.Attribute) and parent.value is node:
continue
found = self.lookup_expr(cm, node)
if found:
if not (found[0] is m and found[1] is target):
continue
elif not self.unresolved(cm, node):
continue
call = parent if isinstance(parent, ast.Call) and parent.func is node else None
yield cm, node, call, found is not None
def unresolved(self, m: Module, node: Node) -> bool:
"""A name, or a dotted path, the app could mean that the scan cannot resolve: not a local, nor an
outside module's (an attribute of an object the function holds is not one)."""
root = node
while isinstance(root, ast.Attribute):
root = root.value
if not isinstance(root, ast.Name) or self.binder(m, root) is not None:
return False
if root.id in m.ambiguous or (root.id not in m.top and root.id not in m.imports):
return True
if root.id in m.top:
return False
if root is node: # imported from an app module that does not bind it where the scan can see
mod = m.imports[root.id][0]
return (mod == "app" or mod.startswith("app.")) and self.lookup(m, root.id) is None
found = self.lookup(m, root.id)
return (
found is not None and found[1] is None
) # an app module whose dotted path went nowhere
def changed(self, m: Module, stmt: Node, name: str, *, only_read: bool) -> bool:
"""Something in the app stores into, deletes or rebinds the module-level `name` that `stmt` binds,
or (`only_read`, a map) uses it but by `map[k]` or a method that only reads it."""
key = id(stmt) * 2 + only_read
if key in self._changed:
return self._changed[key]
own = stmt.targets if isinstance(stmt, ast.Assign) else [getattr(stmt, "target", None)]
out = False
for cm in self.mods.values():
for node in ast.walk(cm.tree):
if out:
break
if not (
isinstance(node, ast.Name)
and node.id == name
or isinstance(node, ast.Attribute)
and node.attr == name
) or (cm is m and node in own):
continue
found = self.lookup_expr(cm, node)
if not (found and found[0] is m and found[1] is stmt):
continue
if not isinstance(node.ctx, ast.Load):
out = True
elif only_read:
p = cm.parents.get(node)
out = not (
isinstance(p, ast.Subscript)
and p.value is node
and isinstance(p.ctx, ast.Load)
or isinstance(p, ast.Attribute)
and p.attr in READS
)
self._changed[key] = out
return out
# --- types ---------------------------------------------------------------------------------------
def ann_values(self, m: Module, ann: Node | None, depth: int = 0) -> set[str] | None:
"""The values a `Literal[…]` annotation, or an alias of one, allows; None when it is not one."""
if isinstance(ann, ast.Subscript) and getattr(ann.value, "id", None) == "Literal":
items = ann.slice.elts if isinstance(ann.slice, ast.Tuple) else [ann.slice]
if all(isinstance(i, ast.Constant) and isinstance(i.value, str) for i in items):
return {i.value for i in items}
return None
if isinstance(ann, (ast.Name, ast.Attribute)) and depth < 8:
found = self.lookup_expr(m, ann)
if found and isinstance(found[0], Module) and isinstance(found[1], ast.Assign):
return self.ann_values(found[0], found[1].value, depth + 1)
return None
def classes(self, m: Module, e: Node | None) -> list[tuple[Module, ast.ClassDef]] | None:
found = self.lookup_expr(m, e) if e is not None else None
if found and isinstance(found[0], Module) and isinstance(found[1], ast.ClassDef):
return [(found[0], found[1])]
return None
def typed(self, m: Module, at: Node, name: str) -> list[tuple[Module, ast.ClassDef]] | None:
"""The class `name` holds where `at` sits: the one an `except C as name` catches, or a parameter's
annotation (a parameter its body binds again holds no known class)."""
handler = self.caught_as(m, at, name)
if handler is not None:
return self.classes(m, handler.type)
fn = self.enclosing(m, at, FUNCS)
if fn is None:
return None
params = {a.arg: a for a in [*fn.args.posonlyargs, *fn.args.args, *fn.args.kwonlyargs]}
if name in params and not self.assignments(fn, name):
return self.classes(m, params[name].annotation)
return None
def attr_values(self, cm: Module, cls: ast.ClassDef, attr: str) -> set[str] | None:
"""An attribute's `Literal` values, from `self.attr: T` in a method of the class."""
for node in ast.walk(cls):
if (
isinstance(node, ast.AnnAssign)
and isinstance(node.target, ast.Attribute)
and node.target.attr == attr
and getattr(node.target.value, "id", None) == "self"
):
return self.ann_values(cm, node.annotation)
return None
# --- reasons -------------------------------------------------------------------------------------
def resolve(self, m: Module, e: Node) -> Res:
if isinstance(e, ast.Constant):
if e.value is None:
return Res(none=True)
if isinstance(e.value, str):
return Res({e.value})
if isinstance(e, ast.IfExp):
return self.resolve(m, e.body) | self.resolve(m, e.orelse)
if isinstance(e, ast.Name):
return self.resolve_name(m, e)
if isinstance(e, ast.Attribute):
return self.resolve_attr(m, e)
if isinstance(e, ast.Call):
found = self.lookup_expr(m, e.func)
if found and isinstance(found[0], Module) and isinstance(found[1], FUNCS):
v = self.ann_values(found[0], found[1].returns)
if v is not None:
return Res(v)
if isinstance(e, ast.Subscript):
return self.resolve_lookup(m, e)
return Res(unread=[self.at(m, e)])
def resolve_name(self, m: Module, e: ast.Name) -> Res:
if self.caught_as(m, e, e.id) is not None:
return Res(unread=[self.at(m, e)]) # the exception itself, not a reason
fn = self.binder(m, e)
if fn is None:
return Res(unread=[self.at(m, e)])
if not isinstance(fn, FUNCS):
return Res(unread=[f"{self.at(m, e)} (bound by a lambda or a comprehension)"])
if any(isinstance(n, (ast.Nonlocal, ast.Global)) and e.id in n.names for n in ast.walk(fn)):
return Res(unread=[f"{self.at(m, e)} (named by a nonlocal or a global)"])
a = fn.args
params = {p.arg: p for p in [*a.posonlyargs, *a.args, *a.kwonlyargs]}
assigned = self.assignments(fn, e.id)
if e.id in params:
if assigned:
return Res(unread=[f"{self.at(m, e)} (a parameter its body binds again)"])
return self.resolve_param(m, fn, params[e.id], e)
out = Res(unread=[] if assigned else [self.at(m, e)]) # `*args`, `**kwargs`
for s in assigned:
if not (isinstance(s, ast.Assign) and all(isinstance(t, ast.Name) for t in s.targets)):
out = out | Res(unread=[self.at(m, e)])
elif id(s) in self._resolving:
out = out | Res(unread=[f"{self.at(m, e)} (reached again while it is resolved)"])
else:
self._resolving.add(id(s))
out = out | self.resolve(m, s.value)
self._resolving.discard(id(s))
return out
def resolve_param(self, m: Module, fn: Node, param: ast.arg, e: Node) -> Res:
v = self.ann_values(m, param.annotation)
if v is not None:
return Res(v)
if param.annotation is not None and getattr(param.annotation, "id", None) != "str":
return Res(unread=[self.at(m, e)])
return self.follow(m, fn, param)
def follow(self, m: Module, fn: Node, param: ast.arg) -> Res:
"""A `str` parameter of a helper: what every call of the helper passes it."""
assert isinstance(fn, FUNCS)
key = (m.name, fn.name, param.arg)
if key in self._followed:
return self._followed[key]
where = f"{m.path}:{fn.lineno}: {fn.name}({param.arg})"
if key in self._following:
return Res(unread=[f"{where} reached again while it is followed"])
if m.top.get(fn.name) is not fn or fn.name in m.ambiguous:
return Res(unread=[f"{where} is not a module-level helper"])
self._following.add(key)
positional = [a.arg for a in [*fn.args.posonlyargs, *fn.args.args]]
by_name = param not in fn.args.posonlyargs
out = Res()
for cm, node, call, resolved in self.references(m, fn):
if not resolved:
out = out | Res(
unread=[f"{self.at(cm, node)} (a reference the scan cannot resolve)"]
)
continue
if call is None:
out = out | Res(unread=[f"{self.at(cm, node)} (the helper used as a value)"])
continue
kw = {k.arg: k.value for k in call.keywords}
given = param.arg in positional and positional.index(param.arg) < len(call.args)
if by_name and param.arg in kw:
out = out | self.resolve(cm, kw[param.arg])
elif any(isinstance(a, ast.Starred) for a in call.args):
out = out | Res(unread=[self.at(cm, call)])
elif given:
out = out | self.resolve(cm, call.args[positional.index(param.arg)])
else:
out = out | Res(unread=[f"{self.at(cm, call)} (does not pass {param.arg})"])
self._followed[key] = out # once followed, its key answers from here: never reached again
return out
def resolve_attr(self, m: Module, e: ast.Attribute) -> Res:
base = e.value
if isinstance(base, ast.Name):
typed = self.typed(m, e, base.id)
if typed is not None:
out = Res()
for cm, cls in typed:
v = self.attr_values(cm, cls, e.attr)
out = out | (
Res(v)
if v is not None
else Res(
unread=[f"{self.at(m, e)} ({cls.name}.{e.attr} has no Literal type)"]
)
)
return out
found = self.lookup_expr(m, e)
if (
found
and isinstance(found[0], Module)
and isinstance(found[1], (ast.Assign, ast.AnnAssign))
and isinstance(found[1].value, ast.Constant)
):
if self.changed(found[0], found[1], e.attr, only_read=False):
return Res(unread=[f"{self.at(m, e)} (bound again or deleted elsewhere)"])
return self.resolve(found[0], found[1].value)
return Res(unread=[self.at(m, e)])
def resolve_lookup(self, m: Module, e: ast.Subscript) -> Res:
found = self.lookup_expr(m, e.value)
if not (
found
and isinstance(found[0], Module)
and isinstance(found[1], (ast.Assign, ast.AnnAssign))
):
return Res(unread=[self.at(m, e)])
table = found[1].value
if not isinstance(table, ast.Dict) or not all(
isinstance(k, ast.Constant) and isinstance(k.value, str) for k in table.keys
):
return Res(unread=[self.at(m, e)])
name = e.value.attr if isinstance(e.value, ast.Attribute) else getattr(e.value, "id", "")
if self.changed(found[0], found[1], name, only_read=True):
return Res(unread=[f"{self.at(m, e)} (the map is changed after it is built)"])
out = Res()
for v in table.values:
out = out | self.resolve(found[0], v)
index = self.resolve(m, e.slice)
keys = {k.value for k in table.keys if isinstance(k, ast.Constant)}
if index.unread or index.none or index.values != keys:
self.errors.append(
f"{self.at(m, e)}: the map's keys {sorted(keys)} are not what its index can be "
f"{sorted(index.values)}{' + unread ' + str(index.unread) if index.unread else ''}"
)
return out
# --- sites ---------------------------------------------------------------------------------------
def is_detail(self, node: Node | None) -> bool:
if isinstance(node, ast.IfExp):
return self.is_detail(node.body) and self.is_detail(node.orelse)
return _code_dict(node)
def factory(self, m: Module, fn: Node) -> bool:
"""A module-level function each return of which is a detail, and each reference to which is a
call sent as one (`quota_detail`)."""
assert isinstance(fn, FUNCS)
if id(fn) in self._factory:
return self._factory[id(fn)]
returns = [n for n in _own_scope(fn) if isinstance(n, ast.Return)]
ok = (
m.top.get(fn.name) is fn # a name bound twice: its calls are unresolved below
and bool(returns)
and all(self.is_detail(r.value) for r in returns)
)
for cm, _, call, resolved in self.references(m, fn) if ok else []:
ok = resolved and id(call) in cm.slots # a bare reference is no call: never sent
if not ok:
break
self._factory[id(fn)] = ok
return ok
def placed(self, m: Module, d: Node) -> bool:
"""`d` is sent as a detail: from a sender's slot, through the arms of `a if c else b`, or as what a
factory returns."""
holder, p = d, m.parents.get(d)
while isinstance(p, ast.IfExp) and holder is not p.test:
holder, p = p, m.parents.get(p)
if id(holder) in m.slots:
return True
fn = self.enclosing(m, p, FUNCS) if isinstance(p, ast.Return) else None
return fn is not None and self.factory(m, fn)
def sayable(self, m: Module, d: Node | None) -> bool:
"""What a slot holds is a detail the client can say: a code dict, or a factory's call."""
if isinstance(d, ast.IfExp):
return self.sayable(m, d.body) and self.sayable(m, d.orelse)
if _code_dict(d):
return True
found = self.lookup_expr(m, d.func) if isinstance(d, ast.Call) else None
return bool(
found
and isinstance(found[0], Module)
and isinstance(found[1], FUNCS)
and self.factory(found[0], found[1])
)
def entries(self, d: Node) -> list[tuple[str | None, Node]]:
"""A dict's (key, value) pairs: None for a `**` spread, "?" for a key that is not a literal."""
if isinstance(d, ast.Dict):
return [
(
k.value
if isinstance(k, ast.Constant) and isinstance(k.value, str)
else None
if k is None
else "?",
v,
)
for k, v in zip(d.keys, d.values, strict=True)
]
assert isinstance(d, ast.Call)
return [(None, a) for a in d.args] + [(k.arg, k.value) for k in d.keywords]
def site(self, m: Module, d: Node, out: dict) -> None:
items = self.entries(d)
code = Res()
for v in (v for k, v in items if k == "code"):
code = code | self.resolve(m, v)
problems = [
f"{self.at(m, v)} (a spread in dict())"
for k, v in items
if k is None and isinstance(d, ast.Call)
] + [
f"{self.at(m, v)} (a spread or a computed key the code and the reason do not follow)"
for i, (k, v) in enumerate(items)
if k in (None, "?") and not {"code", "reason"} <= {k2 for k2, _ in items[i + 1 :]}
]
if not self.placed(m, d):
problems.append(f"{self.at(m, d)} (a detail put where the scan cannot follow it)")
if code.unread or code.none or len(code.values) != 1 or problems:
out["unreadCodes"] += problems or [self.at(m, d)]
return
entry = out["codes"].setdefault(
next(iter(code.values)), {"reasons": set(), "unread": [], "sites": 0, "reasonless": []}
)
entry["sites"] += 1
res, has_reason = Res(), False
for k, v in items:
if k == "reason":
res, has_reason = res | self.resolve(m, v), True
if not has_reason or res.none:
entry["reasonless"].append(f"{m.path}:{d.lineno}")
entry["reasons"] |= res.values
entry["unread"] += res.unread
def other_details(self) -> list[tuple[str, int, str | None]]:
"""Each slot that sends what the client cannot say — nothing, or anything but a code dict or a
factory's call: (its file, its line, the function it is sent from)."""
out = []
for m in self.mods.values():
for call, d in detail_slots(m.tree):
if self.sayable(m, d):
continue
fn = self.enclosing(m, call, FUNCS)
out.append((m.path, call.lineno, getattr(fn, "name", None)))
return sorted(out)
def run(self) -> dict:
out: dict = {
"codes": {},
"unreadCodes": [],
"errors": self.errors,
"excluded": {},
"aliases": {},
"notRefusals": sorted(NOT_REFUSALS),
}
for m in self.mods.values():
for name, stmt in m.top.items():
if (
isinstance(stmt, ast.Assign)
and isinstance(stmt.value, ast.Subscript)
and (v := self.ann_values(m, stmt.value))
):
out["aliases"].setdefault(m.path, {})[name] = sorted(v)
for node in ast.walk(m.tree):
if not _code_dict(node):
continue
if m.path in NOT_REFUSALS:
out["excluded"][m.path] = out["excluded"].get(m.path, 0) + 1
continue
self.site(m, node, out)
for entry in out["codes"].values():
entry["reasons"] = sorted(entry["reasons"])
return out
def scan(sources: dict[str, str]) -> dict:
return Scan(sources).run()
def app_sources(app: pathlib.Path = APP) -> dict[str, str]:
return {
p.relative_to(app).as_posix(): p.read_text(encoding="utf-8")
for p in sorted(app.rglob("*.py"))
if "__pycache__" not in p.parts
}
def scan_app(app: pathlib.Path = APP) -> dict:
return scan(app_sources(app))
if __name__ == "__main__":
json.dump(
scan_app(pathlib.Path(sys.argv[1]) if len(sys.argv) > 1 else APP),
sys.stdout,
indent=1,
sort_keys=True,
)
+72 -37
View File
@@ -1,53 +1,88 @@
"""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
import textwrap
APP = pathlib.Path(__file__).resolve().parents[1] / "app"
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
# 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",
("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 _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 _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_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}")
def test_every_detail_is_one_the_client_can_say():
offenders = _offenders(app_sources())
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}"
"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_scan_sees_a_sentence_when_there_is_one():
def test_the_detail_scan_sees_each_shape_the_client_cannot_say():
"""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)
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"]
+770
View File
@@ -0,0 +1,770 @@
"""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
from tests.refusal_sites import scan, scan_app
def _scan(**files: str) -> dict:
return scan(
{f"{name.replace('__', '/')}.py": textwrap.dedent(src) for name, src in files.items()}
)
def _code(result: dict, code: str) -> dict:
return result["codes"][code]
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
def refused(status: int, reason: str, **params: object) -> HTTPException:
return HTTPException(status_code=status, detail={**params, "code": "refused", "reason": reason})
"""
def test_a_literal_wherever_it_sits_in_the_dict():
r = _scan(
a="""
def f(x):
raise HTTPException(409, {"code": "clip", "limit_s": 5, "reason": "too_long"})
def g(x):
raise HTTPException(409, {"reason": "before", "code": "clip"})
def h(x):
raise HTTPException(409, {
"code": "clip", # a comment between
"reason": "after_comment",
})
def i(x):
raise HTTPException(409, detail=dict(code="clip", reason="by_call"))
"""
)
assert _code(r, "clip")["reasons"] == ["after_comment", "before", "by_call", "too_long"]
assert _code(r, "clip")["unread"] == []
def test_both_arms_of_a_conditional_and_a_none_arm_makes_the_site_reasonless():
r = _scan(
a="""
def f(x):
raise HTTPException(409, {"code": "edit", "reason": "a" if x else "b"})
def g(x):
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"] 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():
r = _scan(
refusal=REFUSAL,
routes__a="""
from http import HTTPStatus
from app import refusal
from app.refusal import refused
def f(status):
raise refused(status, "by_name_status")
def g():
raise refused(HTTPStatus.CONFLICT, reason="by_keyword")
def h():
raise refusal.refused(409, "qualified")
def i(x):
raise refused(409, "a" if x else "second_arm")
""",
routes__b="""
from other.lib import refused
def f():
refused("an outside library's, not the app's")
""",
)
assert _code(r, "refused")["reasons"] == [
"a",
"by_keyword",
"by_name_status",
"qualified",
"second_arm",
]
assert _code(r, "refused")["unread"] == []
def test_a_call_the_helper_cannot_be_followed_through_is_unread():
r = _scan(
refusal=REFUSAL,
routes__a="""
from app.refusal import refused
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 = _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_not_followed_a_call_must_pass_the_reason():
r = _scan(
routes___common="""
def not_found(what: str = "generic"):
return HTTPException(404, {"code": "not_found", "reason": what})
def get_or_404(db, id_, *, what: str = "generic"):
raise not_found(what)
""",
routes__a="""
from app.routes._common import get_or_404, not_found
def f(db):
get_or_404(db, 1, what="user")
get_or_404(db, 2)
raise not_found("debug_session")
def g():
raise not_found()
""",
)
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_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})
def k(*why):
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",
"why",
] # unpacked, a loop, annotated, unbound, `*args`
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
ClipReason = Literal["bad_range", "too_short"]
EditReason = Literal["not_media"]
class ClipError(Exception):
def __init__(self, reason: ClipReason, **params: float):
self.reason: ClipReason = reason
self.params = params
class EditError(Exception):
reason: EditReason
class Untyped(Exception):
def __init__(self, reason: str):
self.reason = reason
class Narrower(ClipError):
pass
""",
routes__a="""
from app import errors
def f():
try:
pass
except errors.ClipError 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})
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": "clip", "reason": e.reason})
except errors.ClipError as e:
raise HTTPException(422, {"code": "clip", "reason": e})
""",
)
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_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"] == ["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_str_helper_nobody_calls_sends_nothing():
r = _scan(
a="""
def helper(reason: str):
return HTTPException(409, {"code": "playlist", "reason": reason})
"""
)
assert _code(r, "playlist") == {"reasons": [], "unread": [], "sites": 1, "reasonless": []}
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"]
class InvalidSetting(ValueError):
def __init__(self, key: str, reason: SettingReason, **params: object):
self.reason: SettingReason = reason
self.params = params
_MAP: dict[SettingReason, str] = {{{keys}}}
def refused(status: int, reason: str, **params: object):
return HTTPException(status, {{**params, "code": "refused", "reason": reason}})
def setting_refused(exc: InvalidSetting):
return refused(400, _MAP[{index}], **exc.params)
"""
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"] == [] 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 spread or a computed key the code and the reason do not follow)"
)
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"
]
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"}
"""
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_name_a_lambda_a_comprehension_a_nonlocal_or_a_global_binds_is_unread():
r = _scan(
a="""
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 _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)",
]
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():
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"]
def test_the_app_scans():
"""The real tree, on the interpreter the backend runs: every refusal it holds is read (the
frontend's contract judges what is read)."""
result = scan_app()
assert {"refused", "not_found", "login", "clip"} <= set(result["codes"])
assert result["errors"] == [] and result["unreadCodes"] == []
+2 -1
View File
@@ -144,7 +144,8 @@ npm run typecheck # tsc x3 (app / node / e2e)
npm run lint # eslint (typed rules included; warnings fail)
npm run format:check # prettier
npm run knip # unused files/exports/deps
npm run test # vitest (incl. the compiler and gate-register contracts)
npm run test # vitest (incl. the compiler and gate-register contracts; the refusal contract runs
# the host's `python` on backend/tests/refusal_sites.py, stdlib only)
npx vite build # the production bundle (devOnlyGuard fails it on dev-only code, chunkGuard on a lazy page loaded at boot)
npm audit --omit=dev --audit-level=high # what ships to the browser, at high or above
```
@@ -1,4 +1,20 @@
[
{
"version": "0.114.3",
"date": "2026-11-01",
"new": [],
"fixed": [
"Editing a file that is neither video nor audio now says why it can't be done, instead of a generic error."
]
},
{
"version": "0.114.2",
"date": "2026-10-30",
"new": [],
"fixed": [
"Security: a library the sign-in relies on is updated, closing two published vulnerabilities."
]
},
{
"version": "0.114.1",
"date": "2026-10-30",
@@ -1,4 +1,20 @@
[
{
"version": "0.114.3",
"date": "2026-11-01",
"new": [],
"fixed": [
"Ha nem videó- vagy hangfájlt próbálsz szerkeszteni, az alkalmazás most megmondja, miért nem lehet, általános hibaüzenet helyett."
]
},
{
"version": "0.114.2",
"date": "2026-10-30",
"new": [],
"fixed": [
"Biztonság: frissült egy könyvtár, amelyre a bejelentkezés épül, így két nyilvános sebezhetőség záródott be."
]
},
{
"version": "0.114.1",
"date": "2026-10-30",
+144 -137
View File
@@ -1,7 +1,15 @@
import { execFileSync } from "node:child_process";
import { readdirSync, readFileSync } from "node:fs";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
import { RUN_STATUSES } from "../lib/activityUrlState";
import {
REFUSAL_BY_REASON,
REFUSAL_FAMILIES,
REFUSAL_FIXED,
REFUSAL_UNSHOWN,
} from "../lib/api/refusalCodes";
import { vocabKey } from "../lib/vocab";
// Every id the BACKEND owns and the UI names must have its name in both languages. The UI looks these
@@ -21,6 +29,25 @@ import { vocabKey } from "../lib/vocab";
const backend = (file: string) =>
readFileSync(new URL(`../../../backend/app/${file}`, import.meta.url), "utf-8");
// Every refusal the backend can answer, read from its syntax tree by `backend/tests/refusal_sites.py`
// (stdlib `ast`, on the host's Python, as the ruff lane is): each code with every reason it can be sent
// with — a reason in a variable followed to its `Literal` type, its helper's every call, its map — and
// what it could not follow, which fails below. It also hands back every `Literal` alias it read.
type RefusalScan = {
codes: Record<string, { reasons: string[]; unread: string[]; reasonless: string[] }>;
unreadCodes: string[];
errors: string[];
notRefusals: string[];
excluded: Record<string, number>;
aliases: Record<string, Record<string, string[]>>;
};
const scan = JSON.parse(
execFileSync("python", ["-I", "backend/tests/refusal_sites.py"], {
cwd: fileURLToPath(new URL("../../../", import.meta.url)),
encoding: "utf-8",
}),
) as RefusalScan;
const mods = import.meta.glob("./locales/*/*.json", { eager: true }) as Record<
string,
{ default: Record<string, unknown> }
@@ -48,12 +75,11 @@ const keysOf = (lang: string): Set<string> => {
const specs = [
...backend("sysconfig.py").matchAll(/ConfigSpec\(\s*"(\w+)",\s*"(\w+)",\s*"(\w+)",\s*"(\w+)"/g),
].map((m) => ({ key: m[1]!, type: m[2]!, group: m[3]!, attr: m[4]! }));
// A choice key's options, as config.py spells them: `<attr>: <Alias> = …` → `<Alias> = Literal[…]`.
// A choice key's options, as config.py spells them: `<attr>: <Alias> = …`, the alias's values from the scan.
const configPy = backend("config.py");
const choicesOf = (attr: string): string[] => {
const alias = configPy.match(new RegExp(`^\\s+${attr}: (\\w+) =`, "m"))?.[1];
const body = alias && configPy.match(new RegExp(`^${alias} = Literal\\[([^\\]]*)\\]`, "m"))?.[1];
return [...(body ?? "").matchAll(/"(\w+)"/g)].map((m) => m[1]!);
return (alias && scan.aliases["config.py"]?.[alias]) || [];
};
const choiceIds = specs
.filter((s) => s.type === "choice")
@@ -84,20 +110,10 @@ const runEnum = (name: string) =>
const runStatuses = runEnum("RunStatus");
const runTriggers = runEnum("RunTrigger");
const quotaActions = enumValues("quota.py");
// The admin actions' refusal reasons → `errors.admin.<reason>`. The codes are read from
// `localizeDetail`'s own ADMIN_REFUSALS, so a code added there is scanned here too; a detail dict
// ruff wraps over several lines still matches.
const adminCodes = [
...(
readFileSync(new URL("../lib/api/refusalText.ts", import.meta.url), "utf-8").match(
/ADMIN_REFUSALS = new Set\(\[([^\]]*)\]/,
)?.[1] ?? ""
).matchAll(/"(\w+)"/g),
].map((m) => m[1]!);
const adminRefusals = [
...backend("routes/admin.py").matchAll(/\{\s*"code":\s*"(\w+)",\s*"reason":\s*"(\w+)"/g),
].filter((m) => adminCodes.includes(m[1]!));
const adminReasons = [...new Set(adminRefusals.map((m) => m[2]!))];
// The admin actions' refusals → `errors.admin.<reason>`: the codes recorded under that prefix.
const adminCodes = Object.entries(REFUSAL_FAMILIES)
.filter(([, f]) => f.prefix === "errors.admin")
.map(([code]) => code);
// The download failure catalog's host-specific entries → `downloads.errors.fail.catalog.<detail>`.
// A code holds a dot (`facebook.login_wall`), so the reverse check strips the prefix and the part
// rather than splitting on dots.
@@ -118,102 +134,46 @@ const failCategories = [
]),
];
// What a structured 404 says is gone → `errors.notFound.<what>`: every `not_found("x")` and every
// `owned_or_404(..., what="x")` / `get_or_404(..., what="x")` under routes/, read from the sources.
const routeFiles = (dir = "routes"): string[] =>
readdirSync(new URL(`../../../backend/app/${dir}/`, import.meta.url), {
// Every backend source, by its path under app/.
const appPaths = (dir = ""): string[] =>
readdirSync(new URL(`../../../backend/app/${dir}`, import.meta.url), {
withFileTypes: true,
}).flatMap((e) =>
e.isDirectory()
? routeFiles(`${dir}/${e.name}`)
? e.name === "__pycache__"
? []
: appPaths(`${dir}${e.name}/`)
: e.name.endsWith(".py")
? [`${dir}/${e.name}`]
? [`${dir}${e.name}`]
: [],
);
const routeSources = (): string[] => routeFiles().map(backend);
// The words a job's result gives for not doing its work → `scheduler.skip.<vocabKey(word)>` (the Scheduler
// page, the run notice and the audit row name them; lib/jobResult). Read as the `"skipped": "…"` /
// `"reason": "…"` literals of the non-HTTP backend: under routes/ (and main.py) a `reason` is a refusal
// code, which `errors.*` names. A word from an exception's text is not a literal and is not covered.
const appFiles = (dir: string): string[] =>
readdirSync(new URL(`../../../backend/app/${dir}`, import.meta.url), {
withFileTypes: true,
}).flatMap((e) =>
e.isDirectory()
? e.name === "routes" || e.name === "__pycache__"
? []
: appFiles(`${dir}${e.name}/`)
: e.name.endsWith(".py") && `${dir}${e.name}` !== "main.py"
? [`${dir}${e.name}`]
: [],
);
const skipWords = [
...new Set(
appFiles("").flatMap((f) =>
[...backend(f).matchAll(/"(?:skipped|reason)": "([^"]+)"/g)].map((m) => m[1]!),
),
),
];
// A target the scan cannot read — `what=SOME_CONSTANT` — is a 404 no locale is checked for (one hid
// `debug_session`). Only the helpers' own pass-through in `_common.py` may name it by variable.
const opaqueWhats = routeFiles()
.filter((f) => !f.endsWith("/_common.py"))
.flatMap((f) =>
[...backend(f).matchAll(/not_found\((?!"|\))[^)]*\)|what=(?!")\S+/g)].map(
(m) => `${f}: ${m[0]}`,
),
);
const goneWhats = [
...new Set(
routeSources().flatMap((src) =>
[...src.matchAll(/not_found\("(\w+)"\)|what="(\w+)"/g)].map((m) => (m[1] ?? m[2])!),
),
),
];
// Every generic refusal → `errors.refused.<reason>`: each `refused(<status>, "<reason>")` anywhere in
// the backend (not `_refused`, a router's own family), plus the values of the settings map, which
// `setting_refused` picks from at runtime, and every `FolderError` reason.
const appSources = (dir = ""): string[] =>
readdirSync(new URL(`../../../backend/app/${dir}`, import.meta.url), {
withFileTypes: true,
}).flatMap((e) =>
e.isDirectory()
? appSources(`${dir}${e.name}/`)
: e.name.endsWith(".py")
? [backend(`${dir}${e.name}`)]
: [],
);
const refusedReasons = [
...new Set([
...appSources().flatMap((src) =>
[...src.matchAll(/(?<![\w.])refused\(\s*\d+,\s*"(\w+)"/g)].map((m) => m[1]!),
),
// A folder change the rules refuse travels as `FolderError("<reason>")` and is answered with
// `refused(409, e.reason)`, which the pattern above cannot read.
...appSources().flatMap((src) =>
[...src.matchAll(/FolderError\("(\w+)"\)/g)].map((m) => m[1]!),
),
...[
...(backend("refusal.py").split("_SETTING_REASONS = {")[1]?.split("}")[0] ?? "").matchAll(
/:\s*"(\w+)"/g,
appPaths()
.filter((f) => !f.startsWith("routes/") && f !== "main.py")
.flatMap((f) =>
[...backend(f).matchAll(/"(?:skipped|reason)": "([^"]+)"/g)].map((m) => m[1]!),
),
].map((m) => m[1]!),
]),
];
// A refused YouTube action → `errors.youtube.<reason>`, the reasons read from their Literal.
const youtubeReasons = [
...(
backend("youtube/client.py").match(/^YouTubeRefusal = Literal\[([^\]]*)\]/m)?.[1] ?? ""
).matchAll(/"(\w+)"/g),
].map((m) => m[1]!);
// A clip the server will not cut → `clip.errors.<reason>`: every `ClipError("<reason>"…)` in the backend
// (a reason may carry its numbers after it: `ClipError("too_long", limit_s=…)`).
const clipReasons = [
...new Set(
appSources().flatMap((src) => [...src.matchAll(/ClipError\("(\w+)"[,)]/g)].map((m) => m[1]!)),
),
];
const backendCodes = Object.keys(scan.codes);
const recordedCodes = [
...Object.keys(REFUSAL_FAMILIES),
...Object.keys(REFUSAL_BY_REASON),
...Object.keys(REFUSAL_FIXED),
...Object.keys(REFUSAL_UNSHOWN),
];
// The codes whose reason picks the sentence, and what each can carry as its reason.
const reasoned = [...Object.keys(REFUSAL_FAMILIES), ...Object.keys(REFUSAL_BY_REASON)];
const reasonsOf: Record<string, string[]> = Object.fromEntries(
reasoned.map((code) => [code, scan.codes[code]?.reasons ?? []]),
);
// The closed English vocabularies the backend WRITES into data, named through `vocab.<group>`
// (lib/vocab) by their letters-and-digits key. A word the backend adds shows in English until named —
// so each list is read from its source here: the TMDB crew jobs kept, every auto-tag the tagger can
@@ -275,10 +235,11 @@ describe("backend ids have their names in both languages", () => {
expect(runTriggers).toEqual(expect.arrayContaining(["scheduled", "ad_hoc"]));
expect(adminCodes).toEqual(expect.arrayContaining(["activate", "resend"]));
expect(failDetails).toEqual(expect.arrayContaining(["facebook.login_wall"]));
expect(goneWhats).toEqual(expect.arrayContaining(["playlist", "download", "job"]));
expect(opaqueWhats).toEqual([]);
expect(youtubeReasons).toEqual(["quota", "reconnect", "unavailable", "gone"]);
expect(clipReasons).toEqual(expect.arrayContaining(["too_long", "past_end", "bad_crop"]));
expect(reasonsOf.not_found).toEqual(
expect.arrayContaining(["playlist", "download", "job", "user", "debug_session"]),
);
expect(reasonsOf.youtube).toEqual(["gone", "quota", "reconnect", "unavailable"]);
expect(reasonsOf.clip).toEqual(expect.arrayContaining(["too_long", "past_end", "bad_crop"]));
expect(crewJobs).toEqual(
expect.arrayContaining(["Director", "Creator", "Director of Photography"]),
);
@@ -294,11 +255,46 @@ describe("backend ids have their names in both languages", () => {
"no service user",
]),
);
expect(refusedReasons).toEqual(
expect.arrayContaining(["weak_password", "setting_too_low", "plex_write_failed"]),
expect(reasonsOf.refused).toEqual(
expect.arrayContaining([
"weak_password",
"setting_too_low",
"plex_write_failed",
"file_folder_gone",
]),
);
// Every code the frontend localizes is raised somewhere the scan can see.
expect(adminCodes.filter((c) => !adminRefusals.some((m) => m[1] === c))).toEqual([]);
});
it("every refusal code the backend answers is said by exactly one record, and no record names another", () => {
expect(backendCodes).toEqual(expect.arrayContaining(["refused", "playlist", "too_large"]));
expect(backendCodes.filter((c) => recordedCodes.filter((r) => r === c).length !== 1)).toEqual(
[],
);
expect(recordedCodes.filter((c) => !backendCodes.includes(c))).toEqual([]);
});
it("every refusal code and reason is one the scan can follow", () => {
// The modules set aside as not refusals still use the key.
expect(Object.keys(scan.excluded).sort()).toEqual(scan.notRefusals);
expect(scan.unreadCodes).toEqual([]);
expect(scan.errors).toEqual([]);
// A reason it could not follow is a sentence nobody is asked for, unless no screen says that code.
expect(
Object.entries(scan.codes)
.filter(([code]) => !(code in REFUSAL_UNSHOWN))
.flatMap(([, c]) => c.unread),
).toEqual([]);
// A code whose reason picks the sentence, sent without one, says only its fallback.
expect(reasoned.flatMap((code) => scan.codes[code]?.reasonless ?? [])).toEqual([]);
});
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([]);
});
it("a code whose reasons share sentences names every reason it is sent with, and no other", () => {
for (const [code, { reasons }] of Object.entries(REFUSAL_BY_REASON))
expect(Object.keys(reasons).sort(), code).toEqual([...reasonsOf[code]!].sort());
});
it("the run list's status chips are the server's run statuses", () => {
@@ -418,33 +414,6 @@ describe("backend ids have their names in both languages", () => {
expect([...named].filter((d) => !failDetails.includes(d))).toEqual([]);
});
it(`${lang}: every "that is gone" a 404 names, and every YouTube refusal`, () => {
expect(missing(goneWhats.map((w) => `errors.notFound.${w}`))).toEqual([]);
expect(
[...idsUnder(keys, "errors.notFound")].filter(
(w) => w !== "generic" && !goneWhats.includes(w),
),
).toEqual([]);
expect(missing(youtubeReasons.map((r) => `errors.youtube.${r}`))).toEqual([]);
expect(
[...idsUnder(keys, "errors.youtube")].filter((r) => !youtubeReasons.includes(r)),
).toEqual([]);
});
it(`${lang}: every generic refusal the backend can answer, and none it cannot`, () => {
expect(missing(refusedReasons.map((r) => `errors.refused.${r}`))).toEqual([]);
expect(
[...idsUnder(keys, "errors.refused")].filter((r) => !refusedReasons.includes(r)),
).toEqual([]);
});
it(`${lang}: every reason a clip is refused, and none it cannot be`, () => {
expect(missing(clipReasons.map((r) => `clip.errors.${r}`))).toEqual([]);
expect([...idsUnder(keys, "clip.errors")].filter((r) => !clipReasons.includes(r))).toEqual(
[],
);
});
it(`${lang}: every crew job, auto-tag and built-in profile the backend writes, and no other`, () => {
for (const [group, words] of Object.entries(VOCAB)) {
const wanted = words.map((w) => `vocab.${group}.${vocabKey(w)}`);
@@ -456,11 +425,49 @@ describe("backend ids have their names in both languages", () => {
}
});
it(`${lang}: every admin refusal reason`, () => {
expect(missing(adminReasons.map((r) => `errors.admin.${r}`))).toEqual([]);
expect([...idsUnder(keys, "errors.admin")].filter((r) => !adminReasons.includes(r))).toEqual(
[],
);
it(`${lang}: every refusal the backend can answer has its sentence`, () => {
const wanted = [
...Object.entries(REFUSAL_FAMILIES).flatMap(([code, f]) => [
...reasonsOf[code]!.map((r) => `${f.prefix}.${r}${"suffix" in f ? f.suffix : ""}`),
f.fallback,
]),
...Object.values(REFUSAL_BY_REASON).flatMap((f) => [
...Object.values(f.reasons),
f.fallback,
]),
...Object.values(REFUSAL_FIXED),
"errors.generic",
];
expect(missing([...new Set(wanted)])).toEqual([]);
});
it(`${lang}: no sentence for a refusal reason the backend cannot send`, () => {
// By prefix: the admin actions share `errors.admin`. A suffixed family shares its prefix with more
// than its refusals — the sign-in screens' `gate`, `addressLabel` — so only an id with the
// refusal's own key counts as a reason there.
const byPrefix = new Map<
string,
{ suffix: string; reasons: Set<string>; spare: Set<string> }
>();
for (const [code, f] of Object.entries(REFUSAL_FAMILIES)) {
const p = byPrefix.get(f.prefix) ?? {
suffix: "suffix" in f ? f.suffix : "",
reasons: new Set<string>(),
spare: new Set<string>(),
};
for (const r of reasonsOf[code]!) p.reasons.add(r);
if (f.fallback.startsWith(`${f.prefix}.`))
p.spare.add(f.fallback.slice(f.prefix.length + 1));
byPrefix.set(f.prefix, p);
}
for (const [prefix, { suffix, reasons, spare }] of byPrefix)
expect(
[...idsUnder(keys, prefix)].filter(
(r) =>
(!suffix || keys.has(`${prefix}.${r}${suffix}`)) && !spare.has(r) && !reasons.has(r),
),
prefix,
).toEqual([]);
});
}
});
@@ -219,6 +219,7 @@
"edit": {
"empty_edit": "Choose a trim range or crop area first",
"source_not_ready": "The source download isn't ready to edit",
"not_media": "Only video and audio can be edited",
"generic": "Couldn't create that edit"
},
"upload": {
@@ -219,6 +219,7 @@
"edit": {
"empty_edit": "Előbb válassz vágási tartományt vagy körbevágási területet",
"source_not_ready": "A forrásletöltés még nem szerkeszthető",
"not_media": "Csak videó és hang szerkeszthető",
"generic": "Nem sikerült létrehozni a vágást"
},
"upload": {
@@ -3,6 +3,8 @@ import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
import { REFUSAL_FAMILIES, type RefusalFamily } from "../lib/api/refusalCodes";
// Every locale key is used by the app. The parity test proves EN and HU agree; nothing proved either
// one is still READ — 75 keys had outlived the code that showed them (a whole sidebar sort vocabulary,
// the old login page, a pager the DataTable replaced), each a string a translator would keep in step
@@ -49,9 +51,14 @@ const code = sources(SRC).join("\n");
const literals = new Set([...code.matchAll(/["'`]([\w.-]+)["'`]/g)].map((m) => m[1]!));
// Each dynamic key as a SHAPE, every interpolation a `*`: `plex.sort.*`, `editor.*.hint`,
// `config.choices.*.*` — from a template literal that opens with a key head, or a head concatenated.
// A refusal's sentence is built from its record (`lib/api/refusalCodes`) rather than a template the
// scan could read, so each family's prefix is a shape of its own.
const shapes = [
...[...code.matchAll(/`([\w.]+\.(?:\$\{[^}`]*\}|[\w.-])*)`/g)].map((m) => m[1]!),
...[...code.matchAll(/["'`]([\w.]+\.)["'`]\s*\+/g)].map((m) => `${m[1]!}\${}`),
...Object.values(REFUSAL_FAMILIES).map(
(f: RefusalFamily) => `${f.prefix}.\${reason}${f.suffix ?? ""}`,
),
]
.filter((s) => s.includes("${"))
.map((s) => s.replace(/\$\{[^}]*\}/g, "*"));
+104
View File
@@ -0,0 +1,104 @@
// Which sentence answers each server refusal (`{ code, reason, … }`), as data: `localizeDetail` says a
// refusal through these records, and `i18n/backendIds.contract.test.ts` reads every code and reason the
// backend can send and requires each to land here with its sentence in both languages. A refusal that
// fell through to the generic "couldn't carry out that action" used to stay unnoticed — the language
// parity test is green when EN and HU miss the same key.
//
// No imports: the contract test reads this module in a plain node run, without i18n.
/** A code whose reasons each name their own sentence: `<prefix>.<reason><suffix>`. */
export type RefusalFamily = { prefix: string; suffix?: string; fallback: string };
export const REFUSAL_FAMILIES = {
// Every other refusal (backend `app/refusal.py`): one family, the reason naming the sentence and the
// rest of the detail its values.
refused: { prefix: "errors.refused", fallback: "errors.generic" },
// An action YouTube refused (routes/_common.youtube_refusal): what the reader can do about it — wait
// for the budget, reconnect, or retry. An unknown reason reads as the retryable one.
youtube: { prefix: "errors.youtube", fallback: "errors.youtube.unavailable" },
// Any row that is gone — deleted in another tab or device, or never the caller's (a stranger's row
// 404s the same way); `reason` names WHAT is gone.
not_found: { prefix: "errors.notFound", fallback: "errors.notFound.generic" },
// A clip the server will not cut (downloads/clip): the range, the frame, the source; a cap comes
// with the refusal (`limit_s`).
clip: { prefix: "clip.errors", fallback: "clip.failed" },
// An admin action on an account or an invite: an address no longer approved, a link or invitation
// with nothing to go to or no mail to go by. The toast's title names the button; this is the WHY,
// which used to be the generic "try again" no retry could fix. One namespace for the three codes.
activate: { prefix: "errors.admin", fallback: "errors.generic" },
resend: { prefix: "errors.admin", fallback: "errors.generic" },
invite_send: { prefix: "errors.admin", fallback: "errors.generic" },
// A playlist action the server refused (Watch later can't be deleted, YouTube said no, …).
playlist: { prefix: "playlists.errors", fallback: "errors.generic" },
// The one-step "move to another list": both reasons mean the list moved under you, named apart.
playlist_move: { prefix: "playlists.moveErrors", fallback: "playlists.moveFailed" },
edit: { prefix: "downloads.errors.edit", fallback: "downloads.errors.edit.generic" },
cookies: { prefix: "downloads.errors.cookies", fallback: "downloads.errors.cookies.generic" },
// A download URL the server refuses to fetch. The reasons are deliberately distinct — "we could not
// look that host up" is a different thing to tell someone than "we will not fetch from that address".
url_blocked: {
prefix: "downloads.errors.urlBlocked",
fallback: "downloads.errors.urlBlocked.generic",
},
// Password sign-in refusals, in one line. The sign-in card renders the account-shaped ones as a full
// panel off `detailData`; this is the form every other consumer (and the card's fallback) gets.
login: {
prefix: "welcome.auth.blocked",
suffix: ".short",
fallback: "welcome.auth.genericError",
},
} satisfies Record<string, RefusalFamily>;
/** A code whose few reasons share sentences: each reason the backend sends is named here. */
export const REFUSAL_BY_REASON = {
quota: {
reasons: {
max_bytes: "downloads.errors.quota.max_bytes",
max_jobs: "downloads.errors.quota.max_jobs",
},
fallback: "downloads.errors.quota.generic",
},
// The set-password link. Its failure is one a user meets by simply clicking an old link, so it may
// not answer in hardcoded English under a bilingual UI.
reset: {
reasons: { weak_password: "welcome.auth.weakPassword", invalid: "welcome.auth.deadLink.short" },
fallback: "welcome.auth.deadLink.short",
},
// A Plex collection name already taken in that library (the one collection-edit refusal).
collection: {
reasons: { name_taken: "plex.collection.nameTaken" },
fallback: "plex.collection.editFailed",
},
} satisfies Record<string, { reasons: Record<string, string>; fallback: string }>;
/** A code with one sentence, whatever else it carries. */
export const REFUSAL_FIXED = {
// The email confirmation link — like `reset`, met by clicking an old link.
verify: "welcome.auth.verifyInvalid",
// Local-upload ingest: the per-file cap carries the limit in bytes; empty is a bare code.
upload_too_large: "downloads.errors.upload.tooLarge",
upload_empty: "downloads.errors.upload.empty",
// The EDGE ceiling's refusal (`bodylimit.TOO_LARGE_BODY`), shared by every route and carrying no
// number. On a DEFAULT install it is what an over-cap upload gets, because the ceiling and the
// route's own cap are the same figure and the edge answers first — the numbered `upload_too_large`
// answers only once an admin has LOWERED the knob (bodylimit.ROUTE_MAX_BODY_BYTES).
// A BACKSTOP, not the plan: the middleware answers without draining, so whether the browser reads
// this at all depends on the client having finished sending (readable up to 200 MiB over loopback,
// a bare `onerror` through a dev proxy). That is why the upload dialog checks the file against
// `upload_max_bytes` BEFORE sending. The sentence stays PAYLOAD-NEUTRAL: a JSON form gets it too
// (`debug_sessions` refuses an over-size draft with the same code), and advice about a file told the
// issue composer's user to pick a smaller one — of nothing.
too_large: "errors.tooLarge",
// A manual sync trigger on cooldown: a pace limit rather than an error, and the copy says so.
sync_cooldown: "errors.syncCooldown",
} satisfies Record<string, string>;
/** Codes no screen prints, so they read as the generic failure — each with why nobody reads it. */
export const REFUSAL_UNSHOWN = {
subtitle: "answers a <track> src; the player polls the track's status and never reads this body",
json_required: "answers the client log beacon, which the UI always sends as JSON",
diagnostics_off: "answers the client log beacon; the client stops streaming on it, silently",
rate_limited: "answers the client log beacon over its rate; dropped lines are not reported",
bad_cursor: "a list cursor the UI itself wrote; only a hand-edited request gets this",
invalid_input: "a NUL byte in a request, which no form of the UI can send",
} satisfies Record<string, string>;
@@ -0,0 +1,80 @@
// @vitest-environment jsdom
//
// A server refusal said through the records (`refusalCodes`), in both languages, by the real i18n —
// one case per record shape, so moving the dispatch into data changed what no reader sees.
import { afterEach, describe, expect, it } from "vitest";
import i18n from "../../i18n";
import { formatBytes } from "../format";
import { localizeDetail } from "./refusalText";
afterEach(async () => {
await i18n.changeLanguage("en");
});
describe.each(["en", "hu"])("localizeDetail (%s)", (lng) => {
const say = async (d: unknown) => {
await i18n.changeLanguage(lng);
return localizeDetail(d);
};
it("names an edit refused for a file that is not video or audio", async () => {
const said = await say({ code: "edit", reason: "not_media" });
expect(said).toBe(i18n.t("downloads.errors.edit.not_media"));
expect(said).not.toBe(i18n.t("downloads.errors.edit.generic"));
});
it("says a family's reason, and its own fallback for a reason it does not know", async () => {
expect(await say({ code: "playlist_move", reason: "same_playlist" })).toBe(
i18n.t("playlists.moveErrors.same_playlist"),
);
expect(await say({ code: "playlist_move", reason: "nope" })).toBe(
i18n.t("playlists.moveFailed"),
);
});
it("reads a sign-in refusal's one-line form", async () => {
expect(await say({ code: "login", reason: "suspended" })).toBe(
i18n.t("welcome.auth.blocked.suspended.short"),
);
});
it("puts a refusal's numbers into its sentence", async () => {
expect(await say({ code: "quota", reason: "max_jobs", limit: 7 })).toContain("7");
expect(await say({ code: "reset", reason: "weak_password", min: 12 })).toContain("12");
expect(await say({ code: "refused", reason: "view_name_taken", name: "Esti" })).toContain(
"Esti",
);
});
it("puts a download cap in GB and an upload cap in bytes, in the reader's number format", async () => {
expect(await say({ code: "quota", reason: "max_bytes", limit: 1.5 * 2 ** 30 })).toContain(
lng === "hu" ? "1,5" : "1.5",
);
const upload = await say({ code: "upload_too_large", limit: 5_242_880 });
expect(upload).toContain(formatBytes(5_242_880, lng));
expect(upload).not.toBe(i18n.t("downloads.errors.upload.tooLarge", { size: "" }));
});
it("says a shared-sentence code's own fallback for a reason it does not know", async () => {
expect(await say({ code: "quota", reason: "nope" })).toBe(
i18n.t("downloads.errors.quota.generic"),
);
expect(await say({ code: "collection", reason: "nope" })).toBe(
i18n.t("plex.collection.editFailed"),
);
});
it("says a shared-sentence code's fallback, and a fixed code's one sentence", async () => {
expect(await say({ code: "reset", reason: "invalid" })).toBe(
i18n.t("welcome.auth.deadLink.short"),
);
expect(await say({ code: "sync_cooldown" })).toBe(i18n.t("errors.syncCooldown"));
});
it("reads a code no screen shows, or one it does not know, as the generic failure", async () => {
expect(await say({ code: "bad_cursor" })).toBe(i18n.t("errors.generic"));
expect(await say({ code: "never_sent" })).toBe(i18n.t("errors.generic"));
expect(await say({ code: "constructor" })).toBe(i18n.t("errors.generic"));
expect(await say(null)).toBe(i18n.t("errors.generic"));
});
});
+57 -104
View File
@@ -3,15 +3,52 @@
// moment, and the notice store cannot import the request core that imports it.
import i18n from "../../i18n";
import { formatBytes, formatDecimal } from "../format";
import {
REFUSAL_BY_REASON,
REFUSAL_FAMILIES,
REFUSAL_FIXED,
type RefusalFamily,
} from "./refusalCodes";
// Localize a STRUCTURED server error ({ code, reason, ... }) into the user's language. FastAPI
// usually returns a plain string `detail`, but some routes (e.g. download quota/edit) return a
// structured object with an i18n reason key + numbers so HU/DE users don't get hardcoded English
// with a dot-decimal separator. Falls back to a generic message for unknown shapes.
// The refusal codes of the admin's account and invite actions (routes/admin.py). Their reasons share
// one namespace, `errors.admin.<reason>`, which i18n/backendIds.contract checks against the backend.
const ADMIN_REFUSALS = new Set(["activate", "resend", "invite_send"]);
type Detail = {
code?: string;
reason?: string;
limit?: number;
limit_s?: number;
min?: number;
} | null;
// Which sentence answers which code lives in `refusalCodes` — the records the backendIds contract holds
// against every code and reason the backend can send. A Map, so a code is looked up as data only.
const families = new Map<string, RefusalFamily>(Object.entries(REFUSAL_FAMILIES));
const byReason = new Map(
Object.entries(REFUSAL_BY_REASON).map(([code, f]) => [
code,
{ reasons: new Map<string, string>(Object.entries(f.reasons)), fallback: f.fallback },
]),
);
const fixed = new Map<string, string>(Object.entries(REFUSAL_FIXED));
// The numbers a refusal carries, in the reader's number format: a download cap in GB, a clip cap in
// minutes, an upload cap in bytes, a password's minimum length; a generic `refused` interpolates
// whatever values its sentence names (`{{min}}`, `{{name}}`, …).
function valuesOf(o: NonNullable<Detail>): Record<string, unknown> {
const lang = i18n.language;
if (o.code === "refused") return { ...o };
if (o.code === "clip")
return { minutes: o.limit_s == null ? undefined : formatDecimal(o.limit_s / 60, lang, 1, 0) };
if (o.code === "quota")
return o.reason === "max_bytes"
? { size: formatDecimal((o.limit || 0) / 1_073_741_824, lang, 1, 0) }
: { limit: o.limit };
if (o.code === "reset") return { min: o.limit ?? o.min };
if (o.code === "upload_too_large") return { size: formatBytes(o.limit, lang) };
return {};
}
// Localize a STRUCTURED server error ({ code, reason, ... }) into the user's language. A reason this
// build does not know reads as its code's fallback sentence rather than a raw key, and a code it does
// not know (or one no screen shows, `REFUSAL_UNSHOWN`) as the generic failure.
export function localizeDetail(d: unknown): string {
const t = i18n.t.bind(i18n);
// FastAPI request-validation errors: `detail` is an ARRAY of { loc, msg, type }. The UI sends
@@ -19,102 +56,18 @@ export function localizeDetail(d: unknown): string {
// `loc` field names are English, snake_case internal tokens — surface a clean, fully localized
// "check your input" message rather than leaking them into a translated sentence.
if (Array.isArray(d)) return t("errors.validation");
const o = d as
| { code?: string; reason?: string; limit?: number; limit_s?: number; min?: number }
| null
| undefined;
if (o?.code === "quota") {
if (o.reason === "max_bytes") {
const gb = formatDecimal((o.limit || 0) / 1_073_741_824, i18n.language, 1, 0);
return t("downloads.errors.quota.max_bytes", { size: gb });
}
if (o.reason === "max_jobs") return t("downloads.errors.quota.max_jobs", { limit: o.limit });
return t("downloads.errors.quota.generic");
}
if (o?.code === "edit")
return t(`downloads.errors.edit.${o.reason}`, t("downloads.errors.edit.generic"));
// A clip the server will not cut (downloads/clip): the range, the frame, the source. An unknown
// reason reads as the plain failure rather than a key; a cap comes with the refusal (`limit_s`).
if (o?.code === "clip")
return t(`clip.errors.${o.reason}`, {
defaultValue: t("clip.failed"),
minutes: o.limit_s == null ? undefined : formatDecimal(o.limit_s / 60, i18n.language, 1, 0),
const o = d as Detail | undefined;
if (!o?.code) return t("errors.generic");
const family = families.get(o.code);
if (family)
return t(`${family.prefix}.${o.reason}${family.suffix ?? ""}`, {
...valuesOf(o),
defaultValue: t(family.fallback),
});
if (o?.code === "cookies")
return t(`downloads.errors.cookies.${o.reason}`, t("downloads.errors.cookies.generic"));
// Password sign-in refusals. The card renders the account-shaped ones as a full panel
// off `detailData`; this is the one-line form every other consumer (and the card's own fallback)
// gets. A reason this build doesn't know falls through to the generic message rather than showing
// a raw key, which is what an i18n miss would otherwise put on screen.
if (o?.code === "login")
return t(`welcome.auth.blocked.${o.reason}.short`, t("welcome.auth.genericError"));
// The two email-link confirmations. Their failure is one a user meets by simply clicking an old
// link, so it is the last place that may answer in hardcoded English under a bilingual UI.
if (o?.code === "verify") return t("welcome.auth.verifyInvalid");
if (o?.code === "reset")
return o.reason === "weak_password"
? t("welcome.auth.weakPassword", { min: o.limit ?? o.min })
: t("welcome.auth.deadLink.short");
// Local-upload ingest: the per-file cap carries the limit in bytes; empty is a bare code.
if (o?.code === "upload_too_large")
return t("downloads.errors.upload.tooLarge", { size: formatBytes(o.limit, i18n.language) });
if (o?.code === "upload_empty") return t("downloads.errors.upload.empty");
// The EDGE ceiling's refusal (`bodylimit.TOO_LARGE_BODY`), shared by every route and carrying no
// number. Not a theoretical case: on a DEFAULT install it is what an over-cap upload gets,
// because the ceiling and the route's own cap are the same figure and the edge answers first —
// the route's numbered `upload_too_large` above is the answer only once an admin has LOWERED the
// knob (see bodylimit.ROUTE_MAX_BODY_BYTES). Without this branch that refusal fell through to
// the generic "couldn't carry out that action", the one message no retry can fix.
//
// A BACKSTOP, not the plan. The middleware answers without draining, so whether the browser gets
// to read this at all depends on the client having finished sending: measured readable up to
// 200 MiB over loopback, and measured as a bare `onerror` through a dev proxy. That is why the
// upload dialog checks the file against `upload_max_bytes` BEFORE sending — this branch is what
// happens when that check could not run.
//
// `errors.tooLarge` stays PAYLOAD-NEUTRAL for that reason. It is also what a JSON form gets
// (`debug_sessions` refuses an over-size draft with the same bare code), so wording it as advice
// about a file told the issue composer's user to pick a smaller one — of nothing. The
// file-specific sentence is `downloads.errors.upload.tooLarge`, one branch up, which has a number.
if (o?.code === "too_large") return t("errors.tooLarge");
// SEC-12: a download URL the server refuses to fetch. The reasons are deliberately distinct —
// "we could not look that host up" is a different thing to tell someone than "we will not fetch
// from that address" — so an unknown reason falls back to the generic refusal rather than a key.
if (o?.code === "url_blocked")
return t(`downloads.errors.urlBlocked.${o.reason}`, t("downloads.errors.urlBlocked.generic"));
// The one-step "move to another list". Both reasons mean the same thing to a reader —
// the list moved under you — but they are different enough to be worth naming separately, and an
// unknown reason falls back to the plain failure rather than printing a key.
if (o?.code === "playlist_move")
return t(`playlists.moveErrors.${o.reason}`, t("playlists.moveFailed"));
// A Plex collection name already taken in that library (the one collection-edit refusal).
if (o?.code === "collection")
return o.reason === "name_taken"
? t("plex.collection.nameTaken")
: t("plex.collection.editFailed");
// An action YouTube refused (routes/_common.youtube_refusal): what the reader can do about it —
// wait for the budget, reconnect, or retry. An unknown reason reads as the retryable one.
if (o?.code === "youtube")
return t(`errors.youtube.${o.reason}`, t("errors.youtube.unavailable"));
// Every other refusal (backend `app/refusal.py`): one family, the reason naming the sentence and
// the rest of the detail its values (`{{min}}`, `{{name}}`, …). A reason this build does not know
// reads as the generic failure rather than a key.
if (o?.code === "refused")
return t(`errors.refused.${o.reason}`, { ...o, defaultValue: t("errors.generic") });
// A playlist action the server refused (Watch later can't be deleted, YouTube said no, …).
if (o?.code === "playlist") return t(`playlists.errors.${o.reason}`, t("errors.generic"));
// An admin action on an account or an invite that the server refused: activating an address that
// is no longer approved, a confirmation link or invitation with nothing to go to or no mail to go
// by. The toast's title already names the button; this is the WHY, and it used to be the generic
// "try again", which no retry could fix.
if (ADMIN_REFUSALS.has(o?.code ?? "")) return t(`errors.admin.${o?.reason}`, t("errors.generic"));
// Any row that is gone — deleted in another tab or device, or never the caller's (a stranger's
// row 404s the same way). `reason` names WHAT is gone; one this build doesn't know says it
// generically rather than printing a key.
if (o?.code === "not_found")
return t(`errors.notFound.${o.reason}`, t("errors.notFound.generic"));
// SEC-14: a manual sync trigger on cooldown. Not an error the user caused so much as a pace
// limit, and the copy says so.
if (o?.code === "sync_cooldown") return t("errors.syncCooldown");
return t("errors.generic");
const shared = byReason.get(o.code);
if (shared) {
return t(shared.reasons.get(o.reason ?? "") ?? shared.fallback, valuesOf(o));
}
const one = fixed.get(o.code);
return one ? t(one, valuesOf(o)) : t("errors.generic");
}