siftlode/backend/tests/test_channel_backfill.py

"""Backend coverage: the channel-detail Normal/Shorts/Live counts (moved off the manager table),
the update_channel -> immediate-backfill wiring, and the interactive
"get full history" guard that keeps it from racing the scheduler's cursor or draining the budget."""

from datetime import datetime, timezone

import pytest
from fastapi.testclient import TestClient

import app.routes.channels as channels_route
import app.state as state
from app.auth import current_user
from app.db import get_db
from app.main import app
from app.models import User
from app.catalog.models import Channel, Subscription, Video
from app.sync import runner, videos

_PUB = datetime(2020, 1, 1, tzinfo=timezone.utc)


def test_channel_detail_returns_normal_shorts_live_counts(db, monkeypatch, override_dep):
    # published_at set so channel_detail skips the YouTube-enrich path (no quota / network in tests).
    ch = Channel(id="UC_nsl", title="N/S/L Channel", published_at=_PUB)
    db.add(ch)
    me = User(email="nsl@example.com")
    db.add(me)
    db.flush()
    # DISTINCT counts (2 shorts, 1 live) so a short<->live filter/field swap can't pass:
    # stored 5, short 2, live 1, normal = 5 - 2 - 1 = 2.
    db.add_all(
        [
            Video(id="v_norm1", channel_id=ch.id),
            Video(id="v_norm2", channel_id=ch.id),
            Video(id="v_short1", channel_id=ch.id, is_short=True),
            Video(id="v_short2", channel_id=ch.id, is_short=True),
            Video(id="v_live1", channel_id=ch.id, live_status="live"),
        ]
    )
    db.commit()

    # The request gate 503s an unconfigured instance; flip the process-level flag (auto-restored).
    monkeypatch.setattr(state, "_configured_cache", True)
    override_dep(get_db, db.info["request"])
    override_dep(current_user, lambda: me)
    resp = TestClient(app).get(f"/api/channels/{ch.id}")

    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["stored_videos"] == 5
    assert body["count_short"] == 2
    assert body["count_live"] == 1
    assert body["count_normal"] == 2


@pytest.mark.parametrize("hidden", [True, False])
def test_channel_detail_surfaces_subscription_hidden_flag(db, monkeypatch, hidden, override_dep):
    # The detail GET must expose the per-subscription `hidden` flag so the channel page can offer
    # the same Unhide toggle the manager has (the flag was previously omitted from the detail dict).
    ch = Channel(id="UC_hid", title="Hidden?", published_at=_PUB)
    db.add(ch)
    me = User(email="hid@example.com")
    db.add(me)
    db.flush()
    db.add(Subscription(user_id=me.id, channel_id=ch.id, hidden=hidden))
    db.commit()

    monkeypatch.setattr(state, "_configured_cache", True)
    override_dep(get_db, db.info["request"])
    override_dep(current_user, lambda: me)
    resp = TestClient(app).get(f"/api/channels/{ch.id}")

    assert resp.status_code == 200, resp.text
    assert resp.json()["hidden"] is hidden


def test_channel_detail_hidden_defaults_false_when_not_subscribed(db, monkeypatch, override_dep):
    # No subscription row → the flag can't be True; it must serialize as False, not raise.
    ch = Channel(id="UC_hid_ns", title="Unsub", published_at=_PUB)
    me = User(email="hidns@example.com")
    db.add_all([ch, me])
    db.commit()

    monkeypatch.setattr(state, "_configured_cache", True)
    override_dep(get_db, db.info["request"])
    override_dep(current_user, lambda: me)
    resp = TestClient(app).get(f"/api/channels/{ch.id}")

    assert resp.status_code == 200, resp.text
    assert resp.json()["hidden"] is False


def _setup_channel_and_sub(db, *, email):
    ch = Channel(
        id="UC_wire",
        title="Wire",
        published_at=_PUB,
        recent_synced_at=_PUB,  # so update_channel skips the in-request recent backfill (no network)
        uploads_playlist_id="UU_wire",
    )
    db.add(ch)
    user = User(email=email)
    db.add(user)
    db.flush()
    db.add(Subscription(user_id=user.id, channel_id=ch.id, deep_requested=False))
    db.commit()
    return ch, user


def _patch_deep_requested(db, monkeypatch, override_dep, user):
    submitted: list[tuple] = []
    calls: list[tuple] = []
    # Record the run the route submits, and what its work would call, instead of running it.
    monkeypatch.setattr(
        channels_route.scheduler,
        "run_ad_hoc",
        lambda job, fn, **kw: submitted.append((job, kw)) or fn(),
    )
    monkeypatch.setattr(channels_route, "backfill_channel_now", lambda *a, **k: calls.append(a))
    monkeypatch.setattr(state, "_configured_cache", True)
    override_dep(get_db, db.info["request"])
    override_dep(current_user, lambda: user)
    resp = TestClient(app).patch("/api/channels/UC_wire", json={"deep_requested": True})
    return resp, submitted, calls


def test_deep_request_starts_the_backfill_as_the_users_run_of_that_channel(
    db, monkeypatch, override_dep
):
    """It was a BackgroundTask: what it did, spent or failed with reached only the log."""
    ch, me = _setup_channel_and_sub(db, email="wire@example.com")
    resp, submitted, calls = _patch_deep_requested(db, monkeypatch, override_dep, me)
    assert resp.status_code == 200, resp.text
    assert resp.json()["deep_requested"] is True
    assert submitted == [("channel_backfill", {"actor_id": me.id, "target": ch.id})]
    # its work runs exactly once, with (channel_id, user_id) — the args backfill_channel_now needs.
    assert calls == [(ch.id, me.id)]


def test_turning_full_history_on_for_a_finished_channel_records_no_run(
    db, monkeypatch, override_dep
):
    """Its whole history is in already: a run would only say "nothing to do", and switching the flag
    off and on again would otherwise write one such row each time, kept for months."""
    ch, me = _setup_channel_and_sub(db, email="done@example.com")
    ch.backfill_done = True
    db.commit()
    resp, submitted, calls = _patch_deep_requested(db, monkeypatch, override_dep, me)
    assert resp.status_code == 200, resp.text
    assert (submitted, calls) == ([], [])


class TestImmediateBackfillGuard:
    """backfill_channel_now coordinates with the paced scheduler through _backfill_in_progress (per-
    channel mutual exclusion, both directions) + _immediate_running (a cap on interactive runs so a
    burst can't drain the shared budget)."""

    def setup_method(self):
        runner._backfill_in_progress.clear()
        runner._immediate_running = 0

    def teardown_method(self):
        runner._backfill_in_progress.clear()
        runner._immediate_running = 0

    @staticmethod
    def _boom_session():
        # Reached only AFTER the guard admits the channel; if the guard short-circuits (as it must
        # in these tests) this never runs, so raising proves the short-circuit.
        raise AssertionError("guard must short-circuit before opening a session")

    def test_a_second_call_for_an_in_flight_channel_is_a_noop(self, monkeypatch):
        runner._backfill_in_progress.add("UC_busy")
        monkeypatch.setattr(runner, "SessionLocal", self._boom_session)
        # returns at once — _boom_session never runs — and the run says why it did nothing.
        assert runner.backfill_channel_now("UC_busy", 1) == {"skipped": "channel busy"}
        assert runner._backfill_in_progress == {"UC_busy"}  # the in-flight entry is untouched

    def test_over_the_concurrency_cap_falls_back_to_the_scheduler(self, monkeypatch):
        runner._immediate_running = runner._MAX_CONCURRENT_IMMEDIATE
        monkeypatch.setattr(runner, "SessionLocal", self._boom_session)
        # over the cap → no-op, the scheduler will handle it
        assert runner.backfill_channel_now("UC_new", 1) == {"skipped": "at capacity"}
        assert "UC_new" not in runner._backfill_in_progress

    @staticmethod
    def _one_channel(db, monkeypatch, cid):
        ch = Channel(id=cid, title=cid, uploads_playlist_id="UU" + cid, backfill_done=False)
        me = User(email=f"{cid}@example.com")
        db.add_all([ch, me])
        db.commit()
        monkeypatch.setattr(runner, "get_service_user", lambda _db: me)
        monkeypatch.setattr(runner.quota, "backfill_reserve_reached", lambda: False)
        monkeypatch.setattr(runner, "YouTubeClient", lambda *a, **k: _NullYt())
        return ch, me

    def test_a_failure_reaches_its_run_and_still_frees_the_channel(self, db, monkeypatch):
        """The failure was logged and swallowed, so the run that now wraps it would read "ok"."""
        ch, me = self._one_channel(db, monkeypatch, "UC_fail")

        def boom(*_a):
            raise RuntimeError("playlistItems 500")

        monkeypatch.setattr(runner, "backfill_channel_deep", boom)
        with pytest.raises(RuntimeError, match="playlistItems 500"):
            runner.backfill_channel_now(ch.id, me.id)
        assert runner._backfill_in_progress == set() and runner._immediate_running == 0

    def test_a_finished_backfill_says_what_it_added(self, db, monkeypatch):
        ch, me = self._one_channel(db, monkeypatch, "UC_ok")
        monkeypatch.setattr(runner, "backfill_channel_deep", lambda *_a: 7)
        monkeypatch.setattr(runner, "reconcile_full_history", lambda _db: 1)
        assert runner.backfill_channel_now(ch.id, me.id) == {
            "videos_added": 7,
            "reconciled_full": 1,
        }

    def test_scheduler_deep_pass_skips_an_in_flight_channel(self, db, monkeypatch):
        # A channel the scheduler's query WOULD pick (deep-requested, recent-synced, not done)…
        ch = Channel(
            id="UC_inflight",
            title="In-flight",
            recent_synced_at=_PUB,
            uploads_playlist_id="UU_inflight",
            backfill_done=False,
        )
        db.add(ch)
        me = User(email="svc@example.com")
        db.add(me)
        db.flush()
        db.add(Subscription(user_id=me.id, channel_id=ch.id, deep_requested=True))
        db.commit()

        paged: list[str] = []
        monkeypatch.setattr(runner, "get_service_user", lambda _db: me)
        monkeypatch.setattr(runner.quota, "remaining_today", lambda _db: 10_000)
        monkeypatch.setattr(runner, "YouTubeClient", lambda *a, **k: _NullYt())
        monkeypatch.setattr(
            runner,
            "backfill_channel_deep",
            lambda _db, _yt, channel, _mp: paged.append(channel.id) or 0,
        )
        monkeypatch.setattr(runner, "reconcile_full_history", lambda _db: 0)

        # …is skipped by run_deep_backfill while an interactive backfill owns it (no cursor race).
        runner._backfill_in_progress.add(ch.id)
        runner.run_deep_backfill(db)
        assert paged == [], (
            "the scheduler must not page a channel an immediate backfill is handling"
        )

        # …and IS paged once the interactive backfill releases it.
        runner._backfill_in_progress.discard(ch.id)
        runner.run_deep_backfill(db)
        assert paged == [ch.id]


class _NullYt:
    def __enter__(self):
        return self

    def __exit__(self, *a):
        return False


def _playlist_page(ids, next_token=None):
    """A playlistItems.list response shaped for stub_from_playlist_item."""
    page = {
        "items": [
            {
                "contentDetails": {"videoId": vid, "videoPublishedAt": "2020-01-01T00:00:00Z"},
                "snippet": {"title": vid},
            }
            for vid in ids
        ]
    }
    if next_token is not None:
        page["nextPageToken"] = next_token
    return page


class _PagedYt:
    """Serves a fixed page-token -> page map and records the tokens it was asked for, so a test can
    assert exactly how many pages a deep-backfill run fetched (and where it resumed from)."""

    def __init__(self, pages):
        self.pages = pages
        self.calls: list[str | None] = []

    def get_playlist_items(self, playlist_id, page_token=None):
        self.calls.append(page_token)
        return self.pages[page_token]


class _RaisingYt:
    """Serves pages like _PagedYt but raises when asked for `raise_on_token`, to simulate a
    network/API failure that lands BETWEEN pages of a deep backfill."""

    def __init__(self, pages, raise_on_token):
        self.pages = pages
        self.raise_on_token = raise_on_token
        self.calls: list[str | None] = []

    def get_playlist_items(self, playlist_id, page_token=None):
        self.calls.append(page_token)
        if page_token == self.raise_on_token:
            raise RuntimeError("simulated mid-backfill failure")
        return self.pages[page_token]


class TestDeepBackfillPaging:
    """backfill_channel_deep must commit the advanced cursor after every page (durable resume, no
    re-charging quota for pages already stored) and yield the moment the shared budget hits the
    backfill reserve — the two properties that make a high immediate max_pages safe."""

    @staticmethod
    def _channel(db):
        ch = Channel(
            id="UC_deep",
            title="Deep",
            recent_synced_at=_PUB,
            uploads_playlist_id="UU_deep",
            backfill_done=False,
            backfill_cursor=None,
        )
        db.add(ch)
        db.commit()
        return ch

    def test_cursor_is_committed_each_page_and_a_later_run_resumes(self, db, monkeypatch):
        monkeypatch.setattr(videos.quota, "backfill_reserve_reached", lambda: False)  # budget ok
        ch = self._channel(db)
        yt = _PagedYt(
            {
                None: _playlist_page(["a1", "a2"], next_token="t1"),
                "t1": _playlist_page(["b1", "b2"], next_token="t2"),
                "t2": _playlist_page(["c1"], next_token=None),  # last page: no nextPageToken
            }
        )

        # First run is bounded to 2 pages: it must stop mid-catalogue with the cursor durably parked
        # on the next unfetched page and backfill_done still False.
        added = videos.backfill_channel_deep(db, yt, ch, max_pages=2)
        assert added == 4
        assert yt.calls == [None, "t1"]
        assert ch.backfill_cursor == "t2"
        assert ch.backfill_done is False
        # Durably persisted (committed per page), not just set on the in-memory object.
        db.expire(ch)
        assert ch.backfill_cursor == "t2"
        assert db.query(Video).filter(Video.channel_id == ch.id).count() == 4

        # A later run resumes FROM the stored cursor (not from the start — no wasted re-fetch) and
        # finishes: reaching the token-less last page flips backfill_done and clears the cursor.
        added2 = videos.backfill_channel_deep(db, yt, ch, max_pages=5)
        assert added2 == 1
        assert yt.calls == [None, "t1", "t2"]  # resumed at "t2", did not re-fetch None/"t1"
        assert ch.backfill_done is True
        assert ch.backfill_cursor is None

    def test_yields_when_the_shared_budget_hits_the_reserve(self, db, monkeypatch):
        # Budget is fine for the first page, then hits the reserve — the guard is checked BEFORE each
        # page, so the run fetches exactly one page and then yields, leaving the tail for the paced
        # scheduler (cursor parked, backfill_done False).
        reached = iter([False, True])
        monkeypatch.setattr(videos.quota, "backfill_reserve_reached", lambda: next(reached))
        ch = self._channel(db)
        yt = _PagedYt(
            {
                None: _playlist_page(["a1", "a2"], next_token="t1"),
                "t1": _playlist_page(["b1"], next_token="t2"),
            }
        )

        added = videos.backfill_channel_deep(db, yt, ch, max_pages=50)
        assert added == 2
        assert yt.calls == [None]  # stopped before fetching the second page
        assert ch.backfill_cursor == "t1"
        assert ch.backfill_done is False

    def test_a_mid_run_failure_leaves_the_prior_page_durably_committed(self, db, monkeypatch):
        # The whole point of committing per page (vs one commit after the loop): if paging blows up
        # mid-run, the pages already ingested stay durable, so the next run resumes from the stored
        # cursor instead of re-fetching — and re-charging quota for — them. A single end-of-loop
        # commit would lose the advanced cursor on the crash (it never reaches the post-loop commit),
        # so this asserts the property the per-page commit actually buys.
        monkeypatch.setattr(videos.quota, "backfill_reserve_reached", lambda: False)  # budget ok
        ch = self._channel(db)
        yt = _RaisingYt(
            {None: _playlist_page(["a1", "a2"], next_token="t1")},
            raise_on_token="t1",  # page 1 succeeds; fetching page 2 (token 't1') fails
        )

        with pytest.raises(RuntimeError):
            videos.backfill_channel_deep(db, yt, ch, max_pages=5)

        # Page 1's cursor AND rows were committed before the page-2 crash — re-read from the DB
        # (expire) to prove it's persisted, not just set on the in-memory object.
        db.expire(ch)
        assert ch.backfill_cursor == "t1"
        assert ch.backfill_done is False
        assert db.query(Video).filter(Video.channel_id == ch.id).count() == 2