siftlode/backend/tests/test_source_size.py

"""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) — a listed one, by its
    NAME (`module::function`), may not rise; no other function may cross the limit. (Ranked per
    module until the E32 review: one function could grow by what another in the same module lost.)
    A rename reads as a new offender plus a stale entry: move its row.
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/routes/admin.py": 864,
    "app/routes/auth.py": 1353,
    "app/routes/downloads.py": 1237,
}

COMPLEX: dict[str, int] = {
    "app/bodylimit.py::__call__": 11,
    "app/discover/hero.py::pick": 14,
    "app/discover/pins.py::_shelf_card": 12,
    "app/downloads/edit.py::normalize_edit_spec": 16,
    "app/downloads/edit.py::run_ffmpeg": 15,
    "app/downloads/formats.py::build_ydl_opts": 11,
    "app/downloads/gc.py::_sweep_orphan_files": 15,
    "app/downloads/gc.py::run_download_gc": 11,
    "app/downloads/og.py::_jpeg_size": 12,
    "app/downloads/og.py::_watch_tags": 11,
    "app/downloads/pipeline.py::_download": 14,
    "app/downloads/pipeline.py::_process_edit": 12,
    "app/feedquery.py::feed_summary": 14,
    "app/feedquery.py::filtered_query": 28,
    "app/plex/query.py::meta_filter_conds": 13,
    "app/plex/stream.py::start_session": 14,
    "app/plex/subtitles.py::_ffmpeg_extract": 12,
    "app/plex/sync.py::sync": 13,
    "app/plex/tmdb.py::parse_tmdb": 11,
    "app/plex/watch_sync.py::run_plex_watch_sync": 12,
    "app/routes/auth.py::callback": 19,
    "app/routes/debug_sessions.py::send_issue": 12,
    "app/routes/downloads.py::resolve_source": 11,
    "app/routes/plex/browse.py::facets": 25,
    "app/routes/plex/browse.py::per_table": 13,
    "app/routes/plex/browse.py::unified_library": 18,
    "app/routes/plex/detail.py::item_detail": 13,
    "app/routes/search.py::search_youtube": 15,
    "app/sync/playlists.py::merge_playlist": 19,
    "app/sync/runner.py::run_rss_poll": 13,
    "app/sync/tagsuggest.py::compute_tag_suggestions": 14,
    "app/worker.py::_run_supervised": 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, int]:
    """`module::function` -> its complexity; a name taken twice in a module gets ` 2`, ` 3` in
    source order."""
    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, int] = {}
    for m in sorted(json.loads(out), key=lambda m: (m["filename"], m["location"]["row"])):
        value = re.search(r"\((\d+) > \d+\)", m["message"])
        name = re.search(r"`([^`]+)`", m["message"])
        assert value and name, m["message"]
        rel = Path(m["filename"]).resolve().relative_to(BACKEND).as_posix()
        key, n = f"{rel}::{name.group(1)}", 2
        while key in found:
            key, n = f"{rel}::{name.group(1)} {n}", n + 1
        found[key] = int(value.group(1))
    return found


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 = [
        f"{fn}: {v} (ceiling {COMPLEX.get(fn, MAX_COMPLEXITY)})"
        for fn, v in _complexity().items()
        if v > COMPLEX.get(fn, MAX_COMPLEXITY)
    ]
    assert worse == []


def test_lists_no_function_that_is_back_under_the_limit_or_gone():
    actual = _complexity()
    assert [fn for fn in COMPLEX if fn not in actual] == []