siftlode/backend/tests/test_body_limit.py

"""SEC-20 — the global request-body cap (`app/bodylimit.py`).

Two halves, deliberately kept apart:

* the MIDDLEWARE tests drive a toy ASGI app, so they pin the mechanism (declared length, counted
  bytes, the exemption match) with no router, no auth and no database in the way;
* the APP tests drive the real application, so they pin that the middleware is actually wired in,
  sits ahead of the auth dependencies, and refuses in the same shape `capped_body` does.

The pair that matters most is "an oversized body is refused" together with "a normal body still
gets through": the first alone would pass just as well if the cap refused *everything*.
"""

import ast
import json
import pathlib

import pytest
from fastapi.testclient import TestClient
from starlette.requests import Request

from app import bodylimit, debugsession, sysconfig
from app.config import settings
from app.models import SystemConfig
from app.bodylimit import DEFAULT_MAX_BODY_BYTES, BodyLimitMiddleware
from app.reader import takeout
from app.routes import diagnostics as diagnostics_routes

SMALL = b"x" * 16
OVER = b"x" * (DEFAULT_MAX_BODY_BYTES + 1)


async def _reading_app(scope, receive, send):
    """A toy app that reads the whole body the way a route does, then reports its length.

    It uses a real Starlette `Request`, so the faked `http.disconnect` reaches it as the real
    ClientDisconnect a real handler would see — the middleware's exception path is exercised, not
    stubbed around.
    """
    body = await Request(scope, receive).body()
    payload = json.dumps({"read": len(body)}).encode()
    await send(
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [(b"content-type", b"application/json")],
        }
    )
    await send({"type": "http.response.body", "body": payload})


def _client(**kwargs) -> TestClient:
    return TestClient(BodyLimitMiddleware(_reading_app, **kwargs))


def _chunks(total: int, size: int = 64 * 1024):
    """A body with NO Content-Length: httpx sends an iterator chunked, which is exactly the shape a
    declared-length check cannot see."""
    sent = 0
    while sent < total:
        n = min(size, total - sent)
        yield b"x" * n
        sent += n


# --- the mechanism ------------------------------------------------------------------------------


def test_a_declared_length_over_the_cap_is_refused_without_reading_the_body():
    r = _client().post("/anything", content=OVER)
    assert r.status_code == 413
    assert r.content == bodylimit.TOO_LARGE_BODY


def test_a_body_under_the_cap_is_passed_through_untouched():
    # The mutation guard for the test above: if the cap refused everything, this would fail.
    r = _client().post("/anything", content=SMALL)
    assert r.status_code == 200
    assert r.json() == {"read": len(SMALL)}


def test_a_chunked_body_is_refused_on_the_COUNTED_bytes():
    # No Content-Length at all, so the only thing that can stop this is counting as it arrives.
    r = _client().post("/anything", content=_chunks(DEFAULT_MAX_BODY_BYTES + 64 * 1024))
    assert r.status_code == 413
    assert r.content == bodylimit.TOO_LARGE_BODY


def test_a_chunked_body_under_the_cap_still_arrives_whole():
    size = DEFAULT_MAX_BODY_BYTES - 1
    r = _client().post("/anything", content=_chunks(size))
    assert r.status_code == 200
    assert r.json() == {"read": size}


def test_an_understated_content_length_does_not_buy_a_pass():
    # The declared-length check is the cheap half; a client that lies must still be counted out.
    r = _client().post("/anything", content=OVER, headers={"content-length": "10"})
    assert r.status_code == 413


def test_exactly_at_the_cap_is_allowed():
    size = DEFAULT_MAX_BODY_BYTES
    r = _client(limit=size).post("/anything", content=b"x" * size)
    assert r.status_code == 200
    assert r.json() == {"read": size}


# --- the per-route ceilings ---------------------------------------------------------------------

BIG = {("POST", "/api/things/{tid}/blob"): DEFAULT_MAX_BODY_BYTES * 4}


def test_a_route_with_a_raised_ceiling_may_send_a_body_over_the_global_cap():
    r = _client(routes=BIG).post("/api/things/abc/blob", content=OVER)
    assert r.status_code == 200
    assert r.json() == {"read": len(OVER)}


def test_a_raised_ceiling_is_still_a_ceiling():
    """The difference between this and an exemption, and the reason the list is a mapping: a route
    that needs more bytes does not thereby get unlimited bytes. Left uncounted, the upload route
    would let an ANONYMOUS caller stream without bound — FastAPI parses a declared body before the
    first dependency, so the auth dependency has not run yet when the bytes land."""
    over = b"x" * (BIG[("POST", "/api/things/{tid}/blob")] + 1)
    r = _client(routes=BIG).post("/api/things/abc/blob", content=over)
    assert r.status_code == 413
    assert r.content == bodylimit.TOO_LARGE_BODY


def test_a_raised_ceiling_counts_a_chunked_body_too():
    cap = BIG[("POST", "/api/things/{tid}/blob")]
    r = _client(routes=BIG).post("/api/things/abc/blob", content=_chunks(cap + 64 * 1024))
    assert r.status_code == 413


def test_the_raised_ceiling_is_scoped_to_its_method_and_its_path():
    client = _client(routes=BIG)
    # Same path, different method.
    assert client.put("/api/things/abc/blob", content=OVER).status_code == 413
    # A neighbour that merely starts the same way.
    assert client.post("/api/things/abc/blob/extra", content=OVER).status_code == 413


def test_a_path_parameter_cannot_swallow_a_slash():
    # `{tid}` matching greedily across `/` would hand the raised ceiling to routes without one.
    mw = BodyLimitMiddleware(_reading_app, routes={("POST", "/a/{tid}/b"): 99})
    assert mw.limit_for("POST", "/a/one/b") == 99
    assert mw.limit_for("POST", "/a/one/two/b") == DEFAULT_MAX_BODY_BYTES


def test_every_request_gets_a_finite_ceiling():
    """There is no pass-through path. A route that is not listed gets the default; one that is
    listed gets its own; nothing is simply not counted."""
    mw = BodyLimitMiddleware(_reading_app)
    assert mw.limit_for("GET", "/api/anything") == DEFAULT_MAX_BODY_BYTES
    for (method, template), cap in bodylimit.ROUTE_MAX_BODY_BYTES.items():
        assert cap > DEFAULT_MAX_BODY_BYTES  # the admission rule: raised, never lowered
        assert mw.limit_for(method, template.replace("{sid}", "abc")) == cap


def _declared_routes(root) -> set[tuple[str, str]]:
    """Every (method, path template) the app serves.

    It has to RECURSE: since FastAPI 0.141 / Starlette 1.6 an `include_router` leaves an opaque
    `_IncludedRouter` in `app.routes` and keeps the real routes under its `original_router`, instead
    of flattening them into the parent as older versions did. Reading `app.routes` alone finds 35
    entries and none of ours; walking the tree finds all 252.
    """
    found: set[tuple[str, str]] = set()
    stack = [root]
    while stack:
        node = stack.pop()
        for route in getattr(node, "routes", ()) or ():
            inner = getattr(route, "original_router", None)
            if inner is not None:
                stack.append(inner)
                continue
            for method in getattr(route, "methods", None) or ():
                found.add((method, route.path))
    return found


@pytest.mark.parametrize(("method", "template"), sorted(bodylimit.ROUTE_MAX_BODY_BYTES))
def test_every_listed_template_still_resolves_to_a_real_route(method, template):
    """The drift guard. The list names route templates that live in another file; a renamed route
    must fail the gate here, not lose its allowance silently in production."""
    from app.main import app

    assert (method, template) in _declared_routes(app)


def test_the_default_ceiling_clears_the_largest_body_a_default_route_takes():
    """The headroom the module's docblock claims, held as a RATIO rather than as a figure in prose.
    The client log sink caps itself at 512 KiB — below the default, which is why it is not listed:
    the global ceiling never fires on it and raising it could only widen what it accepts.

    `<=` alone would not hold the claim. The docblock calls the 2x margin load-bearing precisely
    because a ceiling that refuses a legitimate request is a worse bug than the one this module
    fixes, and a margin of 1x is no margin at all.
    """
    assert diagnostics_routes.BODY_MAX_BYTES * 2 <= DEFAULT_MAX_BODY_BYTES
    assert not any(
        path.startswith("/api/diagnostics") for _, path in bodylimit.ROUTE_MAX_BODY_BYTES
    )


# Every body-reading call site in the routers, as (module, function, limit expression). A route that
# reads its own body declares the size it MEANS to accept; if that is above the global default and
# its route is not in ROUTE_MAX_BODY_BYTES, this middleware refuses a request the route would have
# taken — silently, in production, on a path nobody tested. The point of the list is the third entry.
#
# BOTH helpers, not just `capped_body`: `capped_stream_to` was added later and the registry
# did not follow, so the upload route — the one this whole invariant was written for — was the one
# site the drift guard could not see. The names come from `_common.py` itself
# below, so a THIRD sibling cannot be added without this list noticing either — and a route that
# skips the helpers altogether and awaits `request.body()` lands here as `<no ceiling declared>`
# rather than vanishing.
#
# The module is the path relative to `app/routes/`, so a site under `plex/` is spelled
# `plex/admin.py` and cannot collide with the top-level `admin.py`.
CAPPED_BODY_SITES = {
    ("diagnostics.py", "_parsed_events", "capped_body", "BODY_MAX_BYTES"),
    ("debug_sessions.py", "_image_body", "capped_body", "debugsession.SCREENSHOT_MAX_BYTES"),
    ("feed.py", "_takeout_body", "capped_body", "takeout.MAX_BYTES"),
    ("files.py", "spool_upload", "capped_stream_to", "cap"),
}


def _body_readers() -> set[str]:
    """The names of the body-reading helpers `_common.py` exports — asked of the module rather than
    listed here, so adding a fourth one cannot quietly leave the registry below behind."""
    common = pathlib.Path(__file__).resolve().parents[1] / "app" / "routes" / "_common.py"
    tree = ast.parse(common.read_text("utf-8"), filename=str(common))
    return {
        node.name
        for node in tree.body
        if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef))
        and node.name.startswith("capped_")
    }


def test_the_registry_covers_every_body_reading_helper():
    """The mutation guard for the tests below: if a helper were renamed or added, the site list
    would still match itself while the new one went uncovered."""
    assert _body_readers() == {"capped_body", "capped_stream_to"}, (
        "routes/_common.py grew or renamed a body-reading helper — add its call sites to "
        "CAPPED_BODY_SITES and this expectation."
    )


# The ways a route can read its own body WITHOUT one of `_common.py`'s helpers. Every `Request`
# parameter under `app/routes/` is named `request`, so the base name is an exact filter and not a
# guess — the only other `.json()` in the package is `exc.json()` on a ValidationError.
#
# Without this set the three guards below know only the `capped_*` naming convention, which means
# they are blind to the shape `_common.py` itself is written in. A route spelling `await
# request.body()` inline declares no ceiling, hands no connection back, and would be invisible to
# all three — reading as "no new sites" rather than as a failure, which is the exact way a drift
# guard dies quietly.
RAW_BODY_READS = {"body", "stream", "form", "json"}


def _call_name(node: ast.Call) -> str | None:
    if isinstance(node.func, ast.Attribute):
        return node.func.attr
    return getattr(node.func, "id", None)


def _body_read_calls(fn: ast.AST, readers: set[str]) -> list[tuple[ast.Call, str]]:
    """Every call inside `fn` that reads the request body, as (node, name) in source order."""
    out: list[tuple[ast.Call, str]] = []
    for node in ast.walk(fn):
        if not isinstance(node, ast.Call):
            continue
        if (
            isinstance(node.func, ast.Attribute)
            and isinstance(node.func.value, ast.Name)
            and node.func.value.id == "request"
            and node.func.attr in RAW_BODY_READS
        ):
            out.append((node, f"request.{node.func.attr}"))
        elif _call_name(node) in readers:
            out.append((node, _call_name(node) or ""))
    return sorted(out, key=lambda pair: (pair[0].lineno, pair[0].col_offset))


def _body_reading_functions() -> dict[tuple[str, str], dict]:
    """Every function under `app/routes/` that reads a request body itself, keyed by
    (module, function).

    Parsed, not grepped. A regex over the call text has to guess how the arguments are spelled, and
    the shapes it guesses wrong — a bare `4 * 1024 * 1024`, a parameter not named `request` — are
    the natural ways the NEXT site gets written, so a miss reads as "no new sites" rather than as a
    failure. `ast` sees every call whatever it looks like.

    Keyed by the ENCLOSING FUNCTION, not just the module: a set keyed by module and limit
    expression collapses two call sites that spell their limit the same way, which is exactly what
    a second route copied from the first would do.

    The module half of the key is the path RELATIVE to `app/routes/`, not the basename: the scan
    recurses, and `admin.py`, `discover.py` and `playlists.py` each exist both at the top level and
    under `plex/`. Keying by basename would let the second of a colliding pair overwrite the first
    and take its `releases` flag out of the comparison with it — the same entry-collapse this
    registry was widened to close.

    `_common.py` is skipped: it is where the helpers live, so its own `request.stream()` loops are
    the implementation of the rule rather than a violation of it.
    """
    readers = _body_readers()
    routes_dir = pathlib.Path(__file__).resolve().parents[1] / "app" / "routes"
    found: dict[tuple[str, str], dict] = {}
    for py in sorted(routes_dir.rglob("*.py")):
        if py.name == "_common.py":
            continue
        module = py.relative_to(routes_dir).as_posix()
        tree = ast.parse(py.read_text("utf-8"), filename=str(py))
        for fn in ast.walk(tree):
            if not isinstance(fn, (ast.AsyncFunctionDef, ast.FunctionDef)):
                continue
            calls = _body_read_calls(fn, readers)
            if not calls:
                continue
            names = [name for _, name in calls]
            limits = []
            for node, name in zip([n for n, _ in calls], names, strict=True):
                if name.startswith("request."):
                    limits.append("<no ceiling declared>")
                    continue
                limit = next((kw.value for kw in node.keywords if kw.arg == "limit"), None)
                if limit is None:
                    pos = 2 if name == "capped_stream_to" else 1
                    limit = node.args[pos] if len(node.args) > pos else None
                limits.append(ast.unparse(limit) if limit is not None else "<none>")
            # POSITION, not presence. `any(...)` over the function accepted a release that ran
            # AFTER the await it exists to precede — the natural spelling for someone reading the
            # name as "release it when you're done" — while pinning a connection for the whole
            # transfer. The first body read is what the release has to beat.
            releases = [
                node
                for node in ast.walk(fn)
                if isinstance(node, ast.Call) and _call_name(node) == "release_read_connection"
            ]
            first_read = calls[0][0].lineno
            found[(module, fn.name)] = {
                "helpers": names,
                "limits": limits,
                "releases": any(node.lineno < first_read for node in releases),
                "releases_late": bool(releases)
                and not any(node.lineno < first_read for node in releases),
            }
    return found


def test_every_body_reading_site_is_known_and_fits_under_its_ceiling():
    found = {
        (module, fn, site["helpers"][i], site["limits"][i])
        for (module, fn), site in _body_reading_functions().items()
        for i in range(len(site["helpers"]))
    }
    assert found == CAPPED_BODY_SITES, (
        f"a body-reading call site appeared, moved or changed shape: {found ^ CAPPED_BODY_SITES}. "
        "Each one declares a body size; check it against bodylimit.DEFAULT_MAX_BODY_BYTES and, if "
        "it is larger, add its route to bodylimit.ROUTE_MAX_BODY_BYTES — then update this list."
    )
    # And the known ones really do satisfy the rule.
    assert diagnostics_routes.BODY_MAX_BYTES <= DEFAULT_MAX_BODY_BYTES
    assert debugsession.SCREENSHOT_MAX_BYTES > DEFAULT_MAX_BODY_BYTES
    assert ("POST", "/api/debug-sessions/{sid}/screenshots") in bodylimit.ROUTE_MAX_BODY_BYTES
    # The upload site's limit is the sysconfig knob, which is bounded BY the edge ceiling rather
    # than being a constant to compare against — that pairing is its own test below.
    assert ("POST", "/api/files/upload") in bodylimit.ROUTE_MAX_BODY_BYTES
    # A Takeout file is larger than the default: without its ceiling every real history is a 413.
    assert takeout.MAX_BYTES > DEFAULT_MAX_BODY_BYTES
    assert bodylimit.ROUTE_MAX_BODY_BYTES[("POST", "/api/takeout")] == takeout.MAX_BYTES


def test_release_read_connection_ends_the_transaction_it_was_called_for(db):
    """The helper itself: a SELECT has autobegun, and afterwards it has not."""
    from sqlalchemy import select

    from app.db import release_read_connection

    db.execute(select(SystemConfig).limit(1))
    assert db.in_transaction(), "nothing autobegan, so this test would pass vacuously"
    release_read_connection(db)
    assert not db.in_transaction()


def test_release_read_connection_refuses_to_discard_pending_work(db):
    """The precondition, ENFORCED rather than described.

    The session belongs to the whole request, not to the dependency calling this — so if something
    ahead of it ever stages a write, ending the transaction would throw that write away with
    nothing in the log. `service.enqueue` already learned that a whole-session rollback takes the
    caller's pending work with it, which is why it uses a SAVEPOINT. A comment cannot stop the next
    dependency from being added in the wrong place; this can.
    """
    from app.db import release_read_connection

    db.add(SystemConfig(key="probe_pending_write", value="1"))
    with pytest.raises(RuntimeError, match="pending work"):
        release_read_connection(db)
    db.rollback()  # the staged row is this test's own


def test_release_read_connection_refuses_to_discard_a_FLUSHED_write(db):
    """The half the unit-of-work check cannot see.

    `db.new`/`dirty`/`deleted` empty out the moment a helper flushes, and CLAUDE.md's canonical
    split is "the handler commits; helpers flush" — so the single most likely future violator is
    the one shape the obvious check is blind to. Measured before the fix: all three sets empty, the
    guard silent, the row gone after the rollback.
    """
    from app.db import release_read_connection

    db.add(SystemConfig(key="probe_flushed_write", value="1"))
    db.flush()
    assert not (db.new or db.dirty or db.deleted), (
        "the flush did not clear the unit-of-work, so this test would pass for the wrong reason"
    )
    with pytest.raises(RuntimeError, match="already WRITTEN"):
        release_read_connection(db)
    db.rollback()


def test_release_read_connection_refuses_to_discard_a_core_dml_write(db):
    """The other half: `db.execute(update(...))` never enters the unit-of-work at all, and
    `downloads/service.py` writes that way on purpose (a conditional `UPDATE … WHERE status=…` is
    the honest-states pattern)."""
    from sqlalchemy import insert

    from app.db import release_read_connection

    db.execute(insert(SystemConfig).values(key="probe_core_dml", value="1"))
    assert not (db.new or db.dirty or db.deleted), (
        "Core DML entered the unit-of-work, so this test would pass for the wrong reason"
    )
    with pytest.raises(RuntimeError, match="already WRITTEN"):
        release_read_connection(db)
    db.rollback()


def test_every_body_reading_site_hands_its_pooled_connection_back_first():
    """Reading the body yourself is what makes a request SLOW inside its own database session.

    The auth (or ownership) dependency that must resolve first is a SELECT, so SQLAlchemy has
    autobegun and the session is holding one of the pool's connections; awaiting the body then pins
    it for as long as the client keeps sending — one connection per in-flight upload, `idle in
    transaction`, with nothing to time it out. `db.release_read_connection` ends that transaction,
    and every one of these sites needs it.

    Held generically because the three sites arrived one at a time and only the newest was fixed:
    a fourth body-reading route gets the same requirement automatically.

    BEFORE the read, not merely somewhere in the function. A release that runs after the await
    satisfies the letter of the rule and none of its purpose — the connection was held for exactly
    the span the rule exists to protect — so `releases_late` is reported as its own, louder case.
    """
    sites = _body_reading_functions()
    late = [f"{module}:{fn}" for (module, fn), site in sites.items() if site["releases_late"]]
    missing = [
        f"{module}:{fn}"
        for (module, fn), site in sites.items()
        if not site["releases"] and not site["releases_late"]
    ]
    assert not late, (
        "a route calls db.release_read_connection AFTER it reads the body, which holds the pooled "
        f"connection for exactly the transfer it was supposed to free: {late}. Move the call above "
        "the await."
    )
    assert not missing, (
        "a route reads its own request body while still inside its read transaction, pinning a "
        f"pooled connection for the whole transfer: {missing}. Call "
        "db.release_read_connection(db) before the await — see routes/files.py:spool_upload."
    )


def test_the_upload_ceiling_and_the_admin_knob_cannot_drift_apart():
    """The upload route's in-handler cap is a `sysconfig` value an admin can change; the edge is a
    constant. If the knob could be set above the edge, an admin raising it would silently start
    getting 413s from a layer the admin page never mentions.

    Now the two count the SAME bytes — the body is the file, with no multipart envelope
    wrapped round it — so the old allowance is gone and the edge is the knob's own maximum,
    exactly. A gap reappearing means someone re-wrapped the upload in an envelope without saying so.

    Equality also means the EDGE is what answers on a default install, where no override has lowered
    the knob: the declared-length check refuses before a byte is read. That is not a gap to close
    with headroom (measured: headroom only moves the boundary by its own size, and the route's drain
    pushes the count back over it) — it is a refusal the client has to be able to read, which is why
    `errors.tooLarge` exists and why the dialog checks the file against `upload_max_bytes` first.
    """
    spec = next(s for s in sysconfig.SPECS if s.key == "download_upload_max_bytes")
    edge = bodylimit.ROUTE_MAX_BODY_BYTES[("POST", "/api/files/upload")]
    assert spec.max == edge


def _route_objects(root):
    """`_declared_routes`' sibling, keeping the route OBJECT so its FastAPI dependant can be read."""
    found: dict[tuple[str, str], object] = {}
    stack = [root]
    while stack:
        node = stack.pop()
        for route in getattr(node, "routes", ()) or ():
            inner = getattr(route, "original_router", None)
            if inner is not None:
                stack.append(inner)
                continue
            for method in getattr(route, "methods", None) or ():
                found[(method, route.path)] = route
    return found


@pytest.mark.parametrize(("method", "template"), sorted(bodylimit.ROUTE_MAX_BODY_BYTES))
def test_no_route_with_a_raised_ceiling_declares_a_body_parameter(method, template):
    """The upload fix's invariant, and the reason this is a test rather than a comment.

    A raised ceiling is only honest when the route's own cap applies to the bytes AS THEY ARRIVE.
    FastAPI reads and spools a DECLARED body (`= File(...)`, `= Form(...)`, a Pydantic model)
    BEFORE the first dependency resolves — so on such a route this middleware's ceiling is the only
    thing between an anonymous caller and the whole allowance, the auth dependency never gets a say, and
    the route's in-handler cap measures bytes that have already landed. That is exactly what a fix
    closed on the upload route, which sat in `ROUTE_MAX_BODY_BYTES` with a 500 MiB ceiling and a
    `file: UploadFile = File(...)` body for one release.

    Asked of FastAPI itself rather than parsed or grepped: the router already knows which parameters
    it will read the body for, and a shape a regex or an AST walk guesses wrong is precisely how the
    next route slips through.

    `route.body_field`, NOT `route.dependant.body_params`, and the difference is the whole test.
    `body_params` holds only the parameters of the route's OWN callable; `body_field` is built from
    `_get_flat_body_params`, which recurses into sub-dependencies — and that is also what FastAPI
    itself consults when it decides to read the body. Measured on the pinned FastAPI: a body declared
    inside a sub-dependency leaves `body_params` EMPTY while the body is still read in full before
    auth, so the first version of this test was green over a reproduction of the very defect it
    exists to forbid. The upload route's own idiom — a body read from a
    `yield` dependency — is exactly the shape that would have slipped through."""
    from app.main import app

    route = _route_objects(app).get((method, template))
    assert route is not None, f"{method} {template} is not a route any more"
    assert route.body_field is None, (
        f"{method} {template} declares a body ({route.body_field.name}) on the route or in one of "
        "its dependencies, so FastAPI reads it before the first dependency runs and the raised "
        "ceiling is all that stands between an anonymous caller and the whole allowance. Read the "
        "body yourself, from a dependency declared AFTER the auth one — see "
        "routes/files.py:spool_upload."
    )


def test_a_stored_override_outside_the_bounds_falls_back_on_READ(db):
    """`set_value` refuses an out-of-range int, but neither bound has always been what it is now —
    the upload knob had no `max` before SEC-20 and its `min` used to be 0 — so a value that was
    legal when it was written can be sitting in the database. Served as-is it is exactly the drift
    the test above forbids: the route would accept bytes the edge refuses.

    The fallback is the env default and a warning, NOT a clamp. A clamp is a silent rewrite, and on
    the min side it would be a wrong one: a stored 0 meant "no per-file cap" to the route, and
    clamped to 1 it becomes a one-byte cap that kills uploads with nothing in the log.
    """
    spec = next(s for s in sysconfig.SPECS if s.key == "download_upload_max_bytes")
    assert spec.max is not None and spec.min is not None
    default = settings.download_upload_max_bytes

    row = SystemConfig(key="download_upload_max_bytes", value=str(spec.max * 4))
    db.add(row)
    db.commit()
    assert sysconfig.get_int(db, "download_upload_max_bytes") == default

    row.value = "0"  # the legacy "no per-file cap", now below min
    db.commit()
    assert sysconfig.get_int(db, "download_upload_max_bytes") == default

    # An override that IS in range must still be honoured — otherwise a bounds predicate written
    # backwards would make every override inert and the two assertions above could not tell.
    # Deliberately not `spec.max`: that equals the env default for this key, so the assertion would
    # be satisfied by the fallback it is meant to rule out.
    row.value = str(spec.max - 1)
    db.commit()
    assert spec.max - 1 != default
    assert sysconfig.get_int(db, "download_upload_max_bytes") == spec.max - 1


def test_the_cap_sits_inside_the_response_decorators_and_outside_the_setup_gate():
    """The registration order, held in code rather than in the comment beside it. `add_middleware`
    inverts the reading order (last registered = OUTERMOST), which is exactly the kind of thing a
    prose description gets backwards — it already did once.

    The layers are named, not counted by class: all four `@app.middleware("http")` decorators —
    including the setup gate, which must be INSIDE — register as the same `BaseHTTPMiddleware`, so a
    class-shaped assertion passes for every arrangement anyone would plausibly get wrong.

    Outside the cap: only response decorators, so the 413 is dressed like any other answer. Inside
    it: everything that does work on the caller's behalf — the setup gate opens a DB session per
    request while an instance is unconfigured, which is the anonymous pre-auth work SEC-20 exists to
    remove.
    """
    from app.main import app

    def layer(m) -> str:
        dispatch = m.kwargs.get("dispatch") if m.kwargs else None
        return dispatch.__name__ if dispatch is not None else getattr(m.cls, "__name__", str(m.cls))

    names = [layer(m) for m in app.user_middleware]  # outer → inner
    i = names.index("BodyLimitMiddleware")
    assert set(names[:i]) <= {
        "log_requests",
        "security_headers",
        "static_cache_headers",
        "CORSMiddleware",
    }, f"something that is not a response decorator sits outside the cap: {names[:i]}"
    assert "log_requests" in names[:i], "the access log must stay outside, so a 413 is logged"
    assert "setup_gate" in names[i + 1 :], "the setup gate must not run for an oversized body"
    assert "SessionMiddleware" in names[i + 1 :]


def test_the_screenshot_ceiling_matches_the_cap_the_route_itself_enforces():
    assert debugsession.SCREENSHOT_MAX_BYTES > DEFAULT_MAX_BODY_BYTES
    assert (
        bodylimit.ROUTE_MAX_BODY_BYTES[("POST", "/api/debug-sessions/{sid}/screenshots")]
        == debugsession.SCREENSHOT_MAX_BYTES
    )


# --- the real application -----------------------------------------------------------------------


def test_the_global_refusal_carries_the_baseline_security_headers():
    """The 413 is a response like any other, so it has to be dressed like one. The middleware sits
    INSIDE the security-header and CORS layers for exactly this reason: a refusal a browser drops
    for a missing header is indistinguishable from a network failure, and the user would be shown a
    generic error instead of the translated `too_large`.

    A bare TestClient on purpose, not the `client` fixture: this is the only test that exercises the
    middleware ORDER, and that fixture is DB-gated — so on a host without Postgres the one assertion
    holding the move would have skipped while the file still reported green. The refusal never
    reaches a database; it is answered above the router.
    """
    from app.main import app

    r = TestClient(app).post(
        "/auth/password-login", content=OVER, headers={"content-type": "text/plain"}
    )
    assert r.status_code == 413
    assert r.headers.get("x-content-type-options") == "nosniff"
    assert "referrer-policy" in r.headers


def test_the_login_route_is_refused_before_a_single_dependency_runs(client, monkeypatch):
    """`/auth/password-login` is the reason SEC-20 exists: it is anonymous, it declares a body, and
    FastAPI reads that body in full before the first dependency. `_login_limiter.allow` is the very
    first statement of the handler, so it is the sharpest available proof that nothing ran."""
    from app.routes import auth as auth_routes

    reached = False

    def _record(*args, **kwargs):
        nonlocal reached
        reached = True
        return True

    monkeypatch.setattr(auth_routes._login_limiter, "allow", _record)

    r = client.post("/auth/password-login", content=OVER, headers={"content-type": "text/plain"})
    assert r.status_code == 413
    assert not reached, "the handler ran — the cap is inside the router instead of ahead of it"


def test_a_normal_login_body_still_reaches_the_handler(client, monkeypatch):
    """The mutation check for the test above. If the cap refused every body, the 413 test would
    pass while the app was broken; this one would not."""
    from app.routes import auth as auth_routes

    reached = False

    def _record(*args, **kwargs):
        nonlocal reached
        reached = True
        return True

    monkeypatch.setattr(auth_routes._login_limiter, "allow", _record)

    r = client.post("/auth/password-login", json={"email": "nobody@example.com", "password": "x"})
    assert r.status_code == 401
    assert reached


def test_the_global_refusal_is_byte_identical_to_the_per_route_one(db, as_user, make_user):
    """Two shapes for one fact is how the `cancelled`/`canceled` bug happened. The log sink caps
    itself with `capped_body`; its 413 and the middleware's must be the same bytes.

    Signed in on purpose: `current_user` resolves before the body dependency, so an anonymous call
    would be answered 401 and never reach the per-route cap this compares against. The body is over
    the SINK's 512 KiB but under the global 1 MiB, so it is `capped_body` that refuses it.
    """
    user = make_user()
    db.commit()
    over_the_sink = b"x" * (diagnostics_routes.BODY_MAX_BYTES + 1)
    assert len(over_the_sink) < DEFAULT_MAX_BODY_BYTES
    r = as_user(user).post(
        "/api/diagnostics/events",
        content=over_the_sink,
        headers={"content-type": "application/json"},
    )
    assert r.status_code == 413
    # RAW BYTES, not parsed JSON: the two used to differ by the two spaces `json.dumps` puts in by
    # default and `JSONResponse` does not, and a structural comparison could not see it — which is
    # the only kind of drift the docblock's promise was ever at risk from.
    assert r.content == bodylimit.TOO_LARGE_BODY