refactor(backend): select() and an inline now everywhere, held by a test

CLAUDE.md's canonical table names SQLAlchemy 2.0 select() and an inline
datetime.now(timezone.utc), and both had drifted: five db.query(...) calls
remained in routes/auth.py and routes/me.py (seven before the demo went),
and four private now helpers stood in sync/videos.py, utils.py (imported
as _now by explore and maintenance), scheduler.py (_now_iso) and the e2e
seed. All are the canonical form now.

Ruff cannot ban a method call on an instance or a function by name, so
tests/test_canonical_patterns.py reads every app module's AST and fails on
a .query(...) call or a now/_now/now_utc helper — seen red on a planted file.
This commit is contained in:
2026-10-05 00:19:39 +02:00
parent ab76166878
commit 24c47d12bd
10 changed files with 111 additions and 59 deletions
+52
View File
@@ -0,0 +1,52 @@
"""Two rows of CLAUDE.md's canonical-patterns table, held in code (E35 S4).
Both had drifted while the table said otherwise: seven `db.query(...)` calls stood beside the
`select()` the table names, and four modules carried a private `_now` / `now_utc` helper (nine uses)
where the table says inline `datetime.now(timezone.utc)`. A rule only prose holds drifts; ruff cannot
ban a method call on an instance or a function's name, so this reads the AST of every app module.
Pure-logic: no DB.
"""
import ast
import pathlib
import re
import app
APP = pathlib.Path(app.__file__).resolve().parent
NOW_HELPER = re.compile(r"_?now(_\w+)?|now_utc|utcnow")
def _modules() -> list[tuple[str, ast.Module]]:
return [
(p.relative_to(APP.parent).as_posix(), ast.parse(p.read_text(encoding="utf-8")))
for p in sorted(APP.rglob("*.py"))
]
def test_queries_use_select_not_the_legacy_query_api():
found = [
f"{path}:{node.lineno}"
for path, tree in _modules()
for node in ast.walk(tree)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "query"
]
assert found == [], "use db.execute(select(...)) — SQLAlchemy 2.0 — not db.query(...)"
def test_the_current_time_is_inline_not_a_helper():
found = [
f"{path}:{node.lineno} {node.name}"
for path, tree in _modules()
for node in ast.walk(tree)
if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef)
and NOW_HELPER.fullmatch(node.name)
]
assert found == [], "write datetime.now(timezone.utc) inline, not a now()/_now() helper"
def test_the_scan_sees_the_app():
# a scan that reads nothing is green over nothing
assert len(_modules()) > 50
+15 -12
View File
@@ -7,14 +7,13 @@ trigger, and an upcoming premiere that never went live is. YouTube is a fake tha
videos.list from a dict.
"""
from datetime import timedelta
from datetime import datetime, timedelta, timezone
import pytest
from sqlalchemy import select
from app.models import Channel, Notification, Playlist, PlaylistItem, Video
from app.sync import maintenance
from app.utils import now_utc
class _YT:
@@ -41,11 +40,14 @@ def channel(db):
def test_a_flagged_video_that_answers_again_is_unhidden(db, channel):
v = Video(
id="v1", channel_id=channel.id, unavailable_since=now_utc(), unavailable_reason="removed"
id="v1",
channel_id=channel.id,
unavailable_since=datetime.now(timezone.utc),
unavailable_reason="removed",
)
db.add(v)
db.commit()
out = maintenance._recheck_flagged(db, _YT({"v1": _item("v1")}), now_utc())
out = maintenance._recheck_flagged(db, _YT({"v1": _item("v1")}), datetime.now(timezone.utc))
db.refresh(v)
assert v.unavailable_since is None and v.unavailable_reason is None
assert out == {"checked": 1, "recovered": 1, "deleted": 0, "notified": 0}
@@ -55,12 +57,13 @@ def test_a_video_still_missing_inside_its_grace_period_is_kept_hidden(db, channe
v = Video(
id="v1",
channel_id=channel.id,
unavailable_since=now_utc() - timedelta(days=maintenance.DEFAULT_GRACE_DAYS - 1),
unavailable_since=datetime.now(timezone.utc)
- timedelta(days=maintenance.DEFAULT_GRACE_DAYS - 1),
unavailable_reason="removed",
)
db.add(v)
db.commit()
out = maintenance._recheck_flagged(db, _YT({}), now_utc())
out = maintenance._recheck_flagged(db, _YT({}), datetime.now(timezone.utc))
assert db.get(Video, "v1") is not None
assert out["deleted"] == 0
@@ -69,7 +72,7 @@ def test_past_the_grace_period_it_is_deleted_and_each_user_who_kept_it_told_once
db, channel, make_user
):
a, b = make_user(), make_user()
gone_since = now_utc() - timedelta(days=maintenance.DEFAULT_GRACE_DAYS + 1)
gone_since = datetime.now(timezone.utc) - timedelta(days=maintenance.DEFAULT_GRACE_DAYS + 1)
for vid in ("v1", "v2"):
db.add(
Video(
@@ -92,7 +95,7 @@ def test_past_the_grace_period_it_is_deleted_and_each_user_who_kept_it_told_once
]
)
db.commit()
out = maintenance._recheck_flagged(db, _YT({}), now_utc())
out = maintenance._recheck_flagged(db, _YT({}), datetime.now(timezone.utc))
db.expire_all()
assert db.get(Video, "v1") is None and db.get(Video, "v2") is None
assert db.execute(select(PlaylistItem)).scalars().all() == [] # the cascade
@@ -118,7 +121,7 @@ def test_rolling_revalidation_flags_the_missing_and_keeps_a_live_stream(db, chan
)
db.commit()
yt = _YT({"ok": _item("ok"), "live": _item("live", liveBroadcastContent="live")})
out = maintenance._revalidate_rolling(db, yt, now_utc())
out = maintenance._revalidate_rolling(db, yt, datetime.now(timezone.utc))
db.expire_all()
assert db.get(Video, "gone").unavailable_reason == "removed"
assert db.get(Video, "ok").unavailable_since is None
@@ -150,7 +153,7 @@ def test_a_run_asks_youtube_about_each_video_once(db, channel, monkeypatch, make
Video(
id="v1",
channel_id=channel.id,
unavailable_since=now_utc(),
unavailable_since=datetime.now(timezone.utc),
unavailable_reason="removed",
),
Video(id="v2", channel_id=channel.id),
@@ -163,8 +166,8 @@ def test_a_run_asks_youtube_about_each_video_once(db, channel, monkeypatch, make
def test_an_upcoming_premiere_counts_as_abandoned_only_when_it_never_went_live():
past = (now_utc() - timedelta(days=3)).isoformat()
soon = (now_utc() - timedelta(days=1)).isoformat()
past = (datetime.now(timezone.utc) - timedelta(days=3)).isoformat()
soon = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
def item(**live):
return {"snippet": {"liveBroadcastContent": "upcoming"}, "liveStreamingDetails": live}