siftlode/backend/tests/test_maintenance_lifecycle.py

"""sync/maintenance — the job that HARD-DELETES videos (it was 19% covered).

Its lifecycle: a video missing from videos.list is flagged and hidden; after the grace period a
still-missing one is deleted (cascading to every playlist that held it) and each affected user
gets one notification; a video that answers again is unhidden. A live broadcast is never a delete
trigger, and an upcoming premiere that never went live is. YouTube is a fake that answers
videos.list from a dict.
"""

from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import select

from app.models import Notification
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
from app.sync import maintenance


class _YT:
    def __init__(self, items: dict[str, dict]):
        self.items = items
        self.asked: list[list[str]] = []

    def get_videos(self, ids):
        self.asked.append(list(ids))
        return [self.items[i] for i in ids if i in self.items]


def _item(vid: str, **snippet) -> dict:
    return {"id": vid, "snippet": {"title": vid, **snippet}}


@pytest.fixture
def channel(db):
    ch = Channel(id="UC_maint", title="Maint")
    db.add(ch)
    db.flush()
    return ch


def test_a_flagged_video_that_answers_again_is_unhidden(db, channel):
    v = Video(
        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")}), 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}


def test_a_video_still_missing_inside_its_grace_period_is_kept_hidden(db, channel):
    v = Video(
        id="v1",
        channel_id=channel.id,
        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({}), datetime.now(timezone.utc))
    assert db.get(Video, "v1") is not None
    assert out["deleted"] == 0


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 = datetime.now(timezone.utc) - timedelta(days=maintenance.DEFAULT_GRACE_DAYS + 1)
    for vid in ("v1", "v2"):
        db.add(
            Video(
                id=vid,
                channel_id=channel.id,
                title=f"T {vid}",
                unavailable_since=gone_since,
                unavailable_reason="removed",
            )
        )
    pa = Playlist(user_id=a.id, name="A")
    pb = Playlist(user_id=b.id, name="B")
    db.add_all([pa, pb])
    db.flush()
    db.add_all(
        [
            PlaylistItem(playlist_id=pa.id, video_id="v1", position=0),
            PlaylistItem(playlist_id=pa.id, video_id="v2", position=1),
            PlaylistItem(playlist_id=pb.id, video_id="v2", position=0),
        ]
    )
    db.commit()
    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
    notes = {
        n.user_id: n.data["count"]
        for n in db.execute(select(Notification).where(Notification.type == "maintenance"))
        .scalars()
        .all()
    }
    assert notes == {a.id: 2, b.id: 1}  # one per user, not one per video
    assert out == {"checked": 2, "recovered": 0, "deleted": 2, "notified": 2}


def test_rolling_revalidation_flags_the_missing_and_keeps_a_live_stream(db, channel, monkeypatch):
    monkeypatch.setattr(maintenance.sysconfig, "get_int", lambda db, key: 10)
    monkeypatch.setattr(maintenance.quota, "backfill_reserve_reached", lambda: False)
    db.add_all(
        [
            Video(id="ok", channel_id=channel.id),
            Video(id="gone", channel_id=channel.id),
            Video(id="live", channel_id=channel.id),
        ]
    )
    db.commit()
    yt = _YT({"ok": _item("ok"), "live": _item("live", liveBroadcastContent="live")})
    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
    assert db.get(Video, "live").unavailable_since is None
    assert out == {"checked": 3, "flagged": 1}
    assert all(db.get(Video, v).last_checked_at is not None for v in ("ok", "gone", "live"))


def test_a_run_asks_youtube_about_each_video_once(db, channel, monkeypatch, make_user):
    # Phase 1 recovers v1; phase 2 must not take it as "not checked this run" and fetch it again.
    monkeypatch.setattr(maintenance.sysconfig, "get_int", lambda db, key: 10)
    monkeypatch.setattr(maintenance.quota, "backfill_reserve_reached", lambda: False)
    yt = _YT({"v1": _item("v1"), "v2": _item("v2")})

    class _Client:
        def __init__(self, *_):
            pass

        def __enter__(self):
            return yt

        def __exit__(self, *_):
            return False

    monkeypatch.setattr(maintenance, "YouTubeClient", _Client)
    monkeypatch.setattr(maintenance, "get_service_user", lambda db: make_user())
    db.add_all(
        [
            Video(
                id="v1",
                channel_id=channel.id,
                unavailable_since=datetime.now(timezone.utc),
                unavailable_reason="removed",
            ),
            Video(id="v2", channel_id=channel.id),
        ]
    )
    db.commit()
    maintenance.run_maintenance(db)
    asked = [vid for batch in yt.asked for vid in batch]
    assert sorted(asked) == ["v1", "v2"]


def test_an_upcoming_premiere_counts_as_abandoned_only_when_it_never_went_live():
    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}

    assert maintenance._is_abandoned_upcoming(item(scheduledStartTime=past)) is True
    assert maintenance._is_abandoned_upcoming(item(scheduledStartTime=soon)) is False
    assert (
        maintenance._is_abandoned_upcoming(item(scheduledStartTime=past, actualStartTime=past))
        is False
    )
    assert (
        maintenance._is_abandoned_upcoming({"snippet": {"liveBroadcastContent": "none"}}) is False
    )
    # …and an abandoned premiere has no grace period: the confirming sweep deletes it.
    assert maintenance._grace_days("abandoned") == 0