siftlode/backend/tests/test_video_length.py

"""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 and count it as `live_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: not counted as "+N live"
    ]
    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_the_live_ones(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["live_without_length"] == 2  # the archive and the one on air

    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()["live_without_length"] == 2

    channels = client.get("/api/channels")
    assert channels.status_code == 200
    (ch,) = channels.json()
    assert ch["total_duration_seconds"] == KNOWN_SUM
    assert ch["live_without_length"] == 2


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 `<module>`.
        if any(_reads_raw(n) and id(n) not in in_a_function for n in ast.walk(tree)):
            found.add((rel, "<module>"))
    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)