144 lines
5.4 KiB
Python
144 lines
5.4 KiB
Python
"""Members-only / gated videos: identified from discovery, labelled, and kept
|
|
out of bulk work without becoming unreachable.
|
|
|
|
yt-dlp's flat listing reports availability="subscriber_only" for members-only
|
|
videos, so they are knowable before an extraction attempt is ever spent on them.
|
|
Measured on a real channel: 120 flat entries in one request, exactly 4 flagged,
|
|
matching exactly the 4 the DB had learned about the expensive way.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from yt_scraper.store import BLOCKING_AVAILABILITY, Store, VideoRef
|
|
|
|
|
|
def _store(tmp_path) -> Store:
|
|
store = Store(tmp_path / "state.db")
|
|
store.upsert_channel("UC1", "@alpha", "Alpha", 0)
|
|
return store
|
|
|
|
|
|
def _ref(vid: str, availability: str | None = None) -> VideoRef:
|
|
return VideoRef(vid, "UC1", vid, f"https://www.youtube.com/watch?v={vid}", availability=availability)
|
|
|
|
|
|
# ---------------------------------------------------------------- classification
|
|
|
|
def test_availability_from_discovery_is_stored_and_classified(tmp_path):
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("gated", "subscriber_only"), _ref("open", "public")])
|
|
|
|
assert store.get_video("gated").availability == "subscriber_only"
|
|
assert store.get_video("gated").block_reason == "members_only"
|
|
assert store.get_video("open").block_reason is None
|
|
|
|
|
|
@pytest.mark.parametrize("availability,expected", [
|
|
("subscriber_only", "members_only"),
|
|
("premium_only", "premium_only"),
|
|
("private", "private"),
|
|
("needs_auth", "needs_auth"),
|
|
])
|
|
def test_every_blocking_availability_maps_to_a_reason(tmp_path, availability, expected):
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("v", availability)])
|
|
assert store.get_video("v").block_reason == expected
|
|
|
|
|
|
@pytest.mark.parametrize("availability", ["public", "unlisted", None])
|
|
def test_fetchable_availability_is_never_blocked(tmp_path, availability):
|
|
"""`unlisted` downloads perfectly well — mislabelling it would hide videos
|
|
the user can actually have."""
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("v", availability)])
|
|
assert store.get_video("v").block_reason is None
|
|
assert "unlisted" not in BLOCKING_AVAILABILITY
|
|
|
|
|
|
def test_block_reason_falls_back_to_the_recorded_error(tmp_path):
|
|
"""Rows burned into `error` before availability was captured must still be
|
|
identifiable without re-fetching them."""
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("old")])
|
|
store.mark_error("old", "ERROR: [youtube] x: Join this channel to get access to members-only content")
|
|
|
|
assert store.get_video("old").availability is None
|
|
assert store.get_video("old").block_reason == "members_only"
|
|
|
|
|
|
def test_rate_limit_error_is_not_a_block_reason(tmp_path):
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("t")])
|
|
store.mark_error("t", "ERROR: Video unavailable. The current session has been rate-limited by YouTube")
|
|
|
|
assert store.get_video("t").block_reason is None
|
|
|
|
|
|
def test_rediscovery_does_not_wipe_a_known_availability(tmp_path):
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("v", "subscriber_only")])
|
|
store.upsert_videos([_ref("v", None)]) # a later flat pass omitted the field
|
|
|
|
assert store.get_video("v").availability == "subscriber_only"
|
|
|
|
|
|
def test_set_availability_records_what_extraction_learned(tmp_path):
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("v")])
|
|
store.set_availability("v", "subscriber_only")
|
|
assert store.get_video("v").block_reason == "members_only"
|
|
|
|
store.set_availability("v", None) # must not clear it
|
|
assert store.get_video("v").availability == "subscriber_only"
|
|
|
|
|
|
# ---------------------------------------------------------------- queue behaviour
|
|
|
|
def test_blocked_videos_are_kept_out_of_the_pending_queue(tmp_path):
|
|
"""Bulk runs must not spend requests on videos that cannot be fetched."""
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("gated", "subscriber_only"), _ref("ok"), _ref("unlisted", "unlisted")])
|
|
|
|
ids = {v.video_id for v in store.get_pending("UC1")}
|
|
assert ids == {"ok", "unlisted"}
|
|
|
|
all_ids = {v.video_id for v in store.get_pending("UC1", include_blocked=True)}
|
|
assert all_ids == {"ok", "unlisted", "gated"}
|
|
|
|
|
|
def test_a_blocked_video_is_still_reachable_by_id(tmp_path):
|
|
"""Buying the membership must not leave the video permanently stranded."""
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("gated", "subscriber_only")])
|
|
|
|
assert store.get_video("gated") is not None, "explicit per-video processing still works"
|
|
|
|
|
|
# ---------------------------------------------------------------- filtering
|
|
|
|
def test_blocked_filter_finds_them_across_statuses(tmp_path):
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([
|
|
_ref("gated", "subscriber_only"),
|
|
_ref("legacy"),
|
|
_ref("fine"),
|
|
])
|
|
store.mark_error("legacy", "ERROR: Join this channel to get access to members-only content")
|
|
store.mark_status("fine", "no_subtitles")
|
|
|
|
rows, total = store.query_videos(blocked=True)
|
|
assert {r.video_id for r in rows} == {"gated", "legacy"}
|
|
assert total == 2
|
|
|
|
rows, total = store.query_videos(blocked=False)
|
|
assert {r.video_id for r in rows} == {"fine"}
|
|
|
|
|
|
def test_blocked_filter_absent_means_everything(tmp_path):
|
|
store = _store(tmp_path)
|
|
store.upsert_videos([_ref("gated", "subscriber_only"), _ref("fine")])
|
|
|
|
_rows, total = store.query_videos()
|
|
assert total == 2
|