build(backend): close the lint gaps the E32 audit measured

ruff gains RUF100, PLW1510 and flake8-bugbear, every hit fixed rather than
ignored. B008 stays on with FastAPI's Depends/Query declared immutable.
basedpyright moves to standard mode, pytest to --strict-markers, and a hand
run of pytest now requires the database unless SIFTLODE_TEST_REQUIRE_DB=0.

PLW1510 found a real bug: the three ffprobe probes in plex/stream.py never
looked at the exit status, so a probe that failed was read as the file's
truth (an HDR source sent down the HW-safe path, a multi-audio file collapsed
to one track, a subtitle offset taken from garbage) and the warning each
except branch promises never fired. They now run with check=True.
This commit is contained in:
2026-10-03 16:04:06 +02:00
parent cdfe7a8986
commit 3c16012524
37 changed files with 168 additions and 99 deletions
+12 -9
View File
@@ -25,7 +25,7 @@ from fastapi.testclient import TestClient
from sqlalchemy import create_engine, event, text
from starlette.requests import HTTPConnection
import app.models # noqa: F401 — register every model on Base.metadata
import app.models # register every model on Base.metadata
import app.state as app_state
from app import security
from app.auth import current_user
@@ -37,9 +37,11 @@ from app.models import User
# Sentinel for "this dependency had no override before the fixture set one" (see override_dep).
_MISSING = object()
# The compose test lane sets this so a would-be silent skip (no DB, or no ffmpeg) becomes a hard error
# — the gate must not print green while the DB-backed / ffmpeg tests were quietly not run.
_REQUIRE_FULL = os.getenv("SIFTLODE_TEST_REQUIRE_DB") == "1"
# A would-be silent skip (no DB, or no ffmpeg) is a hard error unless the run says otherwise — the
# suite must not print green while the DB-backed / ffmpeg tests were quietly not run. ON by default
# since E32: as an opt-in only the compose lane and CI set, a hand-run `pytest` skipped ~60 DB tests
# and still passed. `SIFTLODE_TEST_REQUIRE_DB=0` is the explicit "pure-logic only" run.
_REQUIRE_FULL = os.getenv("SIFTLODE_TEST_REQUIRE_DB", "1") != "0"
def _db_reachable() -> bool:
@@ -86,19 +88,20 @@ def _truncate_all():
def pytest_configure(config):
"""Fail the whole run — loudly, once — when SIFTLODE_TEST_REQUIRE_DB=1 but the environment can't
actually run the full suite. Without this, an unreachable DATABASE_URL or a missing ffmpeg turns
"""Fail the whole run — loudly, once — when the full suite is required (the default) but the
environment can't actually run it. Without this, an unreachable DATABASE_URL or a missing ffmpeg turns
~60 DB tests + the poster tests into silent skips and the gate still exits green."""
if not _REQUIRE_FULL:
return
if not _DB_REACHABLE:
raise pytest.UsageError(
"SIFTLODE_TEST_REQUIRE_DB=1 but no Postgres is reachable at the test DATABASE_URL — the "
"DB-backed lane would silently skip. Fix the DB wiring or unset the flag."
"no Postgres is reachable at the test DATABASE_URL — the DB-backed lane would silently "
"skip. Fix the DB wiring, or run the pure-logic tests only with SIFTLODE_TEST_REQUIRE_DB=0."
)
if shutil.which("ffmpeg") is None:
raise pytest.UsageError(
"SIFTLODE_TEST_REQUIRE_DB=1 but ffmpeg is missing — the thumbnail/poster tests would skip."
"ffmpeg is missing — the thumbnail/poster tests would skip. Install it, or run the "
"pure-logic tests only with SIFTLODE_TEST_REQUIRE_DB=0."
)
+1 -1
View File
@@ -321,7 +321,7 @@ def _body_reading_functions() -> dict[tuple[str, str], dict]:
continue
names = [name for _, name in calls]
limits = []
for node, name in zip([n for n, _ in calls], names):
for node, name in zip([n for n, _ in calls], names, strict=True):
if name.startswith("request."):
limits.append("<no ceiling declared>")
continue
+1 -1
View File
@@ -259,7 +259,7 @@ def test_the_writer_drains_what_was_queued_when_it_is_stopped(db):
log.addHandler(sink)
try:
for i in range(logsink.BATCH_MAX + 5):
# noqa G004: the f-string is the POINT — a distinct rendered message per line is
# G004: the f-string is the POINT — a distinct rendered message per line is
# what keeps the flood limiter (which groups by template) out of this case.
log.warning(f"line {i}") # noqa: G004
finally:
+41
View File
@@ -8,6 +8,7 @@ Since the sweep stopped being awaited at startup it also runs CONCURRENTLY with
second half of this file pins what keeps it off a LIVE session's segments."""
import os
import sys
from pathlib import Path
import pytest
@@ -830,6 +831,46 @@ class TestProbeVideoKind:
assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE
class TestProbeExitStatus:
"""A probe whose ffprobe EXITS NON-ZERO is a failed probe, not an answer (E32). Each probe used to
read stdout without looking at the exit status, so a refusing ffprobe silently picked the fallback
— HW-safe, single audio, the requested offset — and skipped the warning its except branch promises.
Drives the REAL subprocess.run with the probe's own kwargs against a stand-in command that prints a
plausible answer and then fails, so the test fails whenever the exit status is ignored."""
@staticmethod
def _failing_ffprobe(monkeypatch, stdout: str):
real_run = stream.subprocess.run
script = f"import sys; sys.stdout.write({stdout!r}); sys.exit(1)"
monkeypatch.setattr(
stream.subprocess,
"run",
lambda _cmd, **kw: real_run([sys.executable, "-c", script], **kw),
)
@staticmethod
def _events(caplog) -> list[str]:
return [getattr(r, "event", "") for r in caplog.records]
def test_video_kind(self, monkeypatch, caplog):
self._failing_ffprobe(monkeypatch, "pix_fmt=yuv420p10le\ncolor_transfer=smpte2084\n")
with caplog.at_level("INFO"):
assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE
assert "plex.stream.probe_kind_failed" in self._events(caplog)
def test_audio_langs(self, monkeypatch, caplog):
self._failing_ffprobe(monkeypatch, "hun\neng\n")
with caplog.at_level("INFO"):
assert stream._probe_audio_langs(Path("/x")) == []
assert "plex.stream.probe_audio_failed" in self._events(caplog)
def test_first_pts(self, monkeypatch, caplog):
self._failing_ffprobe(monkeypatch, "12.5\n")
with caplog.at_level("INFO"):
assert stream._probe_first_pts(Path("/x")) is None
assert "plex.stream.first_pts_probe_failed" in self._events(caplog)
class TestOpenclPresent:
"""The OpenCL-runtime probe (_opencl_present) — the precondition for the HW HDR tonemap graph. Drives
the REAL glob against a temp ICD dir instead of monkeypatching the function, so a regression to the