"""A YouTube length past a day is unknown, everywhere a length is used (`catalog.models.LENGTH_MAX_S`). Measured on prod (2026-11-06): two 24/7 streams' archives report 122,962,914 s and 34,762,459 s, and the owner's Watch later total read "44,073 hr 55 min". Such a video now has no length on any API answer, the totals leave it out (a playlist's counts it as `without_length`), and the duration filter and sort pass it over like a video whose length is not fetched yet. An AST check holds who may still read the raw column. DB lane (the AST check needs none). """ import ast from datetime import datetime, timezone from pathlib import Path from sqlalchemy.dialects import postgresql from app.catalog.models import LENGTH_MAX_S, Channel, Subscription, Video, known_length from app.reader.models import Playlist, PlaylistItem YEARS = 122_962_914 # the measured archive HOUR = 3600 def _seed(db, user): db.add(Channel(id="UClen", title="Len")) db.flush() db.add(Subscription(user_id=user.id, channel_id="UClen", priority=0, hidden=False)) rows = [ ("short", 600, "none"), ("long", 2 * HOUR, "none"), ("day", LENGTH_MAX_S, "was_live"), # exactly a day still is a length ("archive", YEARS, "was_live"), ("onair", None, "live"), ("unfetched", None, "none"), # not live, but left out of the sum: counted all the same ] for i, (vid, seconds, live) in enumerate(rows): db.add( Video( id=vid, channel_id="UClen", published_at=datetime(2024, 1, i + 1, tzinfo=timezone.utc), title=vid, duration_seconds=seconds, live_status=live, is_short=False, via_search=False, via_explore=False, ) ) pl = Playlist(user_id=user.id, name="Later") db.add(pl) db.flush() for i, (vid, _s, _l) in enumerate(rows): db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=i)) db.commit() return pl KNOWN_SUM = 600 + 2 * HOUR + LENGTH_MAX_S def test_the_rule(): assert known_length(None) is None assert known_length(LENGTH_MAX_S) == LENGTH_MAX_S assert known_length(LENGTH_MAX_S + 1) is None assert known_length(YEARS) is None def test_the_sql_length_carries_its_limit_as_a_literal(): # The duration sort indexes (migration 0117) are built on this expression with 86400 in it. A # bound parameter matches no index once the driver reuses a prepared, generic plan, and the # sort-index test cannot see that: it compiles with literal binds. compiled = Video.length_seconds.expression.compile(dialect=postgresql.dialect()) assert compiled.params == {} assert f"> {LENGTH_MAX_S}" in str(compiled) def test_a_years_long_archive_has_no_length_in_the_feed_or_a_playlist(db, make_user, as_user): user = make_user() pl = _seed(db, user) client = as_user(user) feed = client.get("/api/feed?show=all&include_live=true&limit=50") assert feed.status_code == 200 by_id = {v["id"]: v["duration_seconds"] for v in feed.json()["items"]} assert by_id["archive"] is None assert by_id["day"] == LENGTH_MAX_S detail = client.get(f"/api/playlists/{pl.id}") assert detail.status_code == 200 items = {v["id"]: v["duration_seconds"] for v in detail.json()["items"]} assert items["archive"] is None one = client.get("/api/videos/archive") assert one.status_code == 200 assert one.json()["duration_seconds"] is None def test_an_off_catalog_archive_has_no_length_either(db, make_user, as_user, monkeypatch): # A video not in the catalog is looked up on YouTube (videos.list); its length is judged the same. from app.routes import feed as feed_routes class Yt: def __init__(self, *_): pass def __enter__(self): return self def __exit__(self, *exc): return False def get_videos(self, ids): hours = {"offyears": YEARS // HOUR, "offshort": 1}[ids[0]] return [ {"snippet": {}, "statistics": {}, "contentDetails": {"duration": f"PT{hours}H"}} ] monkeypatch.setattr(feed_routes, "YouTubeClient", Yt) client = as_user(make_user()) assert client.get("/api/videos/offyears").json()["duration_seconds"] is None assert client.get("/api/videos/offshort").json()["duration_seconds"] == HOUR def test_the_totals_leave_it_out_and_count_what_they_left_out(db, make_user, as_user): user = make_user() pl = _seed(db, user) client = as_user(user) listed = client.get("/api/playlists") assert listed.status_code == 200 (summary,) = [p for p in listed.json() if p["id"] == pl.id] assert summary["total_duration_seconds"] == KNOWN_SUM assert summary["without_length"] == 3 # the archive, the one on air, the unfetched one renamed = client.patch(f"/api/playlists/{pl.id}", json={"name": "Later 2"}) assert renamed.status_code == 200 assert renamed.json()["total_duration_seconds"] == KNOWN_SUM assert renamed.json()["without_length"] == 3 channels = client.get("/api/channels") assert channels.status_code == 200 (ch,) = channels.json() assert ch["total_duration_seconds"] == KNOWN_SUM # The channel table shows the sum alone (the count did not fit its column), so it is not sent. assert "without_length" not in ch and "live_without_length" not in ch def test_the_duration_filter_and_sort_pass_it_over(db, make_user, as_user): user = make_user() _seed(db, user) client = as_user(user) base = "/api/feed?show=all&include_live=true&limit=50" longer = client.get(f"{base}&min_duration={HOUR}") assert longer.status_code == 200 assert {v["id"] for v in longer.json()["items"]} == {"long", "day"} longest = client.get(f"{base}&sort=duration_desc") assert longest.status_code == 200 order = [v["id"] for v in longest.json()["items"]] assert order[:3] == ["day", "long", "short"] assert set(order[3:]) == {"archive", "onair", "unfetched"} # Who may read `duration_seconds` itself, by (file, function): what each one asks is not "how long". RAW_READERS = { # The column is defined and turned into a length here. ("catalog/models.py", "length_seconds"), ("catalog/models.py", "_length_seconds_expression"), # Ingest writes it; the live refresh asks whether YouTube has answered for an archive yet; the # Shorts classification needs only "longer than a Short", which a years-long value proves. ("sync/videos.py", "apply_video_details"), ("sync/videos.py", "refresh_live"), ("sync/videos.py", "run_shorts_classification"), ("catalog/search.py", "classify_shorts"), ("catalog/queries.py", "work_queue"), # Not a video's column: the player's own report, and the watch history's looked-up lengths, which # count a watched video as more than a glance (a watched archive is that too). ("routes/_common.py", "kept_position"), ("routes/plex/playback.py", "item_progress"), ("reader/queries.py", "_counted_channels"), ("reader/queries.py", "watched_channels"), ("reader/queries.py", "watched_channel_videos"), } def _raw_reads() -> set[tuple[str, str]]: root = Path(__file__).resolve().parents[1] / "app" found: set[tuple[str, str]] = set() for path in root.rglob("*.py"): if "alembic" in path.parts: continue rel = path.relative_to(root).as_posix() tree = ast.parse(path.read_text(encoding="utf-8")) in_a_function: set[int] = set() for fn in ast.walk(tree): if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): continue for node in ast.walk(fn): if _reads_raw(node): found.add((rel, fn.name)) in_a_function.add(id(node)) # A read outside every def (module or class level, a lambda there) is named ``. if any(_reads_raw(n) and id(n) not in in_a_function for n in ast.walk(tree)): found.add((rel, "")) return found def _reads_raw(node: ast.AST) -> bool: return isinstance(node, ast.Attribute) and node.attr == "duration_seconds" def test_only_the_listed_readers_take_the_raw_column(): found = _raw_reads() stray = sorted(found - RAW_READERS) assert not stray, f"read `length_seconds` (a length) instead of the raw column: {stray}" # A listed reader that stopped reading it leaves the list. assert not RAW_READERS - found, sorted(RAW_READERS - found)