"""The size / complexity RATCHET for the backend (E32, owner 2026-10-03 — "class size" from the quality checklist, measured rather than argued). The frontend's twin is src/lib/sourceSize.contract.test.ts. Two limits, each with today's offenders as a CEILING: * a module over MAX_LINES lines — a listed one may not grow past its line count; any other module may not cross the limit; * a function over ruff's C901 limit (cyclomatic complexity, MAX_COMPLEXITY) — per module, the offenders' values highest-first may not rise, and no new offender may appear. Shrinking is free; a module or function that drops UNDER the limit must be taken off its list (red until it is), so the lists only get shorter. Raising a ceiling is a decision for the owner, with the reason in docs/gates.md. C901 is measured here rather than switched on in ruff.toml because ruff has no per-function baseline: on, it would need a `noqa` on each of today's offenders, and a `noqa` lets the function grow without limit. This keeps the ceiling on the number. """ import json import re import subprocess import sys from pathlib import Path BACKEND = Path(__file__).resolve().parent.parent MAX_LINES = 800 MAX_COMPLEXITY = 10 # ruff's C901 default (mccabe max-complexity) BIG_MODULES: dict[str, int] = { "app/downloads/edit.py": 833, "app/downloads/pipeline.py": 870, "app/logs.py": 839, "app/models.py": 1607, "app/plex/stream.py": 1128, "app/plex/subtitles.py": 1092, "app/plex/sync.py": 841, "app/routes/admin.py": 864, "app/routes/auth.py": 1353, "app/routes/downloads.py": 1237, "app/sync/playlists.py": 843, } COMPLEX: dict[str, list[int]] = { "app/bodylimit.py": [11], "app/discover/hero.py": [14], "app/discover/pins.py": [12], "app/downloads/edit.py": [16, 15], "app/downloads/formats.py": [11], "app/downloads/gc.py": [15, 11], "app/downloads/og.py": [12, 11], "app/downloads/pipeline.py": [14, 12], "app/feedquery.py": [28, 14], "app/plex/query.py": [13], "app/plex/stream.py": [14], "app/plex/subtitles.py": [12], "app/plex/sync.py": [13], "app/plex/tmdb.py": [11], "app/plex/watch_sync.py": [12], "app/routes/auth.py": [19], "app/routes/debug_sessions.py": [12], "app/routes/downloads.py": [11], "app/routes/plex/browse.py": [25, 18, 13], "app/routes/plex/detail.py": [13], "app/routes/search.py": [15], "app/sync/playlists.py": [19], "app/sync/runner.py": [13], "app/sync/tagsuggest.py": [14], "app/worker.py": [18], } def _line_counts() -> dict[str, int]: return { p.relative_to(BACKEND).as_posix(): len(p.read_text(encoding="utf-8").split("\n")) for p in sorted((BACKEND / "app").rglob("*.py")) } def _complexity() -> dict[str, list[int]]: out = subprocess.run( [ sys.executable, "-m", "ruff", "check", "--config", str(BACKEND / "ruff.toml"), "--select", "C901", "--output-format", "json", "--exit-zero", str(BACKEND / "app"), ], capture_output=True, text=True, check=True, ).stdout found: dict[str, list[int]] = {} for m in json.loads(out): value = re.search(r"\((\d+) > \d+\)", m["message"]) assert value, m["message"] rel = Path(m["filename"]).resolve().relative_to(BACKEND).as_posix() found.setdefault(rel, []).append(int(value.group(1))) return {f: sorted(v, reverse=True) for f, v in found.items()} def test_no_module_crosses_the_line_limit_and_a_listed_one_does_not_grow(): over = [ f"{f}: {n} lines (ceiling {BIG_MODULES.get(f, MAX_LINES)})" for f, n in _line_counts().items() if n > BIG_MODULES.get(f, MAX_LINES) ] assert over == [] def test_lists_no_module_that_is_back_under_the_limit(): lines = _line_counts() assert [f for f in BIG_MODULES if lines.get(f, 0) <= MAX_LINES] == [] def test_no_function_crosses_the_complexity_limit_and_a_listed_one_does_not_rise(): worse = [] for f, values in _complexity().items(): ceiling = COMPLEX.get(f, []) for i, v in enumerate(values): cap = ceiling[i] if i < len(ceiling) else MAX_COMPLEXITY if v > cap: worse.append(f"{f}: #{i + 1} is {v} (ceiling {cap})") assert worse == [] def test_lists_no_offender_that_is_back_under_the_limit(): actual = _complexity() stale = [ f"{f}: listed {len(v)}, now {len(actual.get(f, []))}" for f, v in COMPLEX.items() if len(actual.get(f, [])) < len(v) ] assert stale == []