"""R14 backend coverage: the channel-detail Normal/Shorts/Live counts (moved off the manager table), the update_channel -> immediate-backfill wiring (incl. the demo guard), 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 Channel, Subscription, User, 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): # 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) app.dependency_overrides[get_db] = lambda: db app.dependency_overrides[current_user] = lambda: me try: resp = TestClient(app).get(f"/api/channels/{ch.id}") finally: app.dependency_overrides.clear() 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): # S6: 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) app.dependency_overrides[get_db] = lambda: db app.dependency_overrides[current_user] = lambda: me try: resp = TestClient(app).get(f"/api/channels/{ch.id}") finally: app.dependency_overrides.clear() 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): # 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) app.dependency_overrides[get_db] = lambda: db app.dependency_overrides[current_user] = lambda: me try: resp = TestClient(app).get(f"/api/channels/{ch.id}") finally: app.dependency_overrides.clear() assert resp.status_code == 200, resp.text assert resp.json()["hidden"] is False def _setup_channel_and_sub(db, *, email, is_demo=False): 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, is_demo=is_demo) 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, user): calls: list[tuple] = [] # background.add_task runs after the response in TestClient; record what would be scheduled. monkeypatch.setattr(channels_route, "backfill_channel_now", lambda *a, **k: calls.append(a)) monkeypatch.setattr(state, "_configured_cache", True) app.dependency_overrides[get_db] = lambda: db app.dependency_overrides[current_user] = lambda: user try: resp = TestClient(app).patch("/api/channels/UC_wire", json={"deep_requested": True}) finally: app.dependency_overrides.clear() return resp, calls def test_deep_request_schedules_immediate_backfill_for_non_demo(db, monkeypatch): ch, me = _setup_channel_and_sub(db, email="wire@example.com") resp, calls = _patch_deep_requested(db, monkeypatch, me) assert resp.status_code == 200, resp.text assert resp.json()["deep_requested"] is True # scheduled exactly once, with (channel_id, user_id) — the args backfill_channel_now needs. assert calls == [(ch.id, me.id)] def test_deep_request_does_not_schedule_immediate_backfill_for_demo(db, monkeypatch): # The SHARED demo account must not be able to kick immediate full backfills (budget drain). _ch, demo = _setup_channel_and_sub(db, email="demo@example.com", is_demo=True) resp, calls = _patch_deep_requested(db, monkeypatch, demo) assert resp.status_code == 200, resp.text assert calls == [], "demo must not schedule an immediate backfill" # …and deep_requested is not persisted on the shared demo subscription either. assert resp.json()["deep_requested"] is False 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) runner.backfill_channel_now("UC_busy", 1) # returns immediately; _boom_session never runs 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) runner.backfill_channel_now("UC_new", 1) # over the cap → no-op, scheduler will handle it assert "UC_new" not in runner._backfill_in_progress 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