siftlode/backend/tests/test_channel_details_refresh.py

"""A channel's metadata is fetched again once it is stale, not only the first time it is seen.

Banners, avatars, descriptions and counts change on YouTube, and an old banner URL stops resolving:
a channel that kept the metadata of the day it was first imported showed a dead banner forever. The
subscription import and the channel page both refresh what is older than `DETAILS_MAX_AGE`, and
leave fresh metadata alone (each refresh spends quota). DB lane.
"""

from datetime import datetime, timedelta, timezone

import pytest

from app.models import Channel, Subscription
from app.routes import channels
from app.sync import subscriptions
from app.sync.subscriptions import DETAILS_MAX_AGE, import_subscriptions

NOW = datetime.now(timezone.utc)
FRESH = NOW - timedelta(days=1)
STALE = NOW - DETAILS_MAX_AGE - timedelta(hours=1)


def _item(cid: str) -> dict:
    return {
        "id": cid,
        "snippet": {"title": f"{cid} renamed", "publishedAt": "2010-01-01T00:00:00Z"},
        "brandingSettings": {"image": {"bannerExternalUrl": f"https://yt3.example/{cid}-new"}},
    }


def _youtube(asked: list[list[str]], subscribed: list[str], gone: tuple[str, ...] = ()):
    class _Client:
        def __init__(self, db, user):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def iter_subscriptions(self):
            for cid in subscribed:
                yield {"channel_id": cid, "title": cid, "yt_subscription_id": f"sub-{cid}"}

        def get_channels(self, ids):
            asked.append(sorted(ids))
            # A terminated / deleted channel is simply absent from `channels.list`'s answer.
            return [_item(cid) for cid in ids if cid not in gone]

    return _Client


def _channel(db, cid: str, synced: datetime | None) -> None:
    db.add(
        Channel(
            id=cid,
            title=cid,
            banner_url=f"https://yt3.example/{cid}-old",
            published_at=datetime(2010, 1, 1, tzinfo=timezone.utc),
            details_synced_at=synced,
        )
    )


def test_the_import_refreshes_stale_and_missing_details_and_leaves_fresh_ones(
    db, make_user, monkeypatch
):
    user = make_user()
    _channel(db, "UCfresh", FRESH)
    _channel(db, "UCstale", STALE)
    _channel(db, "UCnever", None)
    db.flush()
    for cid in ("UCfresh", "UCstale", "UCnever"):
        db.add(Subscription(user_id=user.id, channel_id=cid))
    db.commit()
    asked: list[list[str]] = []
    monkeypatch.setattr(
        subscriptions, "YouTubeClient", _youtube(asked, ["UCfresh", "UCstale", "UCnever"])
    )

    result = import_subscriptions(db, user)

    assert asked == [["UCnever", "UCstale"]]
    assert result["channels_detailed"] == 2
    stale = db.get(Channel, "UCstale")
    assert stale is not None
    assert stale.banner_url == "https://yt3.example/UCstale-new"
    assert stale.details_synced_at is not None and stale.details_synced_at > STALE
    fresh = db.get(Channel, "UCfresh")
    assert fresh is not None and fresh.banner_url == "https://yt3.example/UCfresh-old"


@pytest.mark.parametrize(("synced", "refreshed"), [(STALE, True), (FRESH, False)])
def test_the_channel_page_refreshes_a_stale_channel_only(
    db, make_user, monkeypatch, synced, refreshed
):
    user = make_user()
    _channel(db, "UCpage", synced)
    db.commit()
    asked: list[list[str]] = []
    monkeypatch.setattr(channels, "YouTubeClient", _youtube(asked, []))

    body = channels.channel_detail("UCpage", user=user, db=db)

    assert asked == ([["UCpage"]] if refreshed else [])
    suffix = "new" if refreshed else "old"
    assert body["banner_url"] == f"https://yt3.example/UCpage-{suffix}"


def test_a_channel_youtube_no_longer_returns_is_not_asked_for_on_every_view(
    db, make_user, monkeypatch
):
    """A terminated channel is missing from `channels.list`'s answer, so nothing was stamped and the
    channel page asked for it again (1 quota unit) on every single view. The miss is stamped like a
    fetch, so the next ask waits for `DETAILS_MAX_AGE` like any other channel."""
    user = make_user()
    db.add(Channel(id="UCgone", title="UCgone"))
    db.commit()
    asked: list[list[str]] = []
    monkeypatch.setattr(channels, "YouTubeClient", _youtube(asked, [], gone=("UCgone",)))

    channels.channel_detail("UCgone", user=user, db=db)
    channels.channel_detail("UCgone", user=user, db=db)

    assert asked == [["UCgone"]]


def test_the_import_stamps_a_channel_youtube_no_longer_returns(db, make_user, monkeypatch):
    user = make_user()
    _channel(db, "UCgone", STALE)
    db.flush()
    db.add(Subscription(user_id=user.id, channel_id="UCgone"))
    db.commit()
    asked: list[list[str]] = []
    monkeypatch.setattr(
        subscriptions, "YouTubeClient", _youtube(asked, ["UCgone"], gone=("UCgone",))
    )

    import_subscriptions(db, user)
    import_subscriptions(db, user)

    assert asked == [["UCgone"]]