siftlode/backend/tests/test_adapters.py

"""Per-source adapter selection + the source-specific policy each one carries. These lock the
behaviour that used to live inline in worker.build_uploader_url / backfill._should_refetch."""

from app.config import settings
from app.titles import normalize_title
from app.downloads.adapters import (
    FacebookAdapter,
    RedditAdapter,
    SourceAdapter,
    YouTubeAdapter,
    resolve_adapter,
)


def test_resolve_by_source_kind_and_host_and_info():
    assert resolve_adapter("youtube", "abc123").key == "youtube"
    assert resolve_adapter("url", "https://www.youtube.com/watch?v=x").key == "youtube"
    assert resolve_adapter("url", "https://youtu.be/x").key == "youtube"
    assert resolve_adapter("url", "https://www.facebook.com/reel/123/").key == "facebook"
    assert resolve_adapter("url", "https://fb.watch/abc/").key == "facebook"
    assert resolve_adapter("url", "https://www.reddit.com/r/x/comments/y/z").key == "reddit"
    assert resolve_adapter("url", "https://example.com/video.mp4").key == "generic"
    # After extraction the info dict's extractor decides even for a bare/redirected url.
    assert resolve_adapter("url", None, {"extractor_key": "facebook"}).key == "facebook"
    assert resolve_adapter("url", None, {"extractor_key": "Reddit"}).key == "reddit"


def test_host_match_is_netloc_not_a_loose_substring():
    # A path that merely CONTAINS "facebook.com" must not select the Facebook adapter.
    assert resolve_adapter("url", "https://example.com/facebook.com-mirror").key == "generic"


def test_info_match_is_extractor_key_not_the_webpage_url():
    # The webpage_url path may carry a SIBLING source's name (a reddit slug about youtube); only the
    # yt-dlp extractor_key identifies the source, so this must resolve reddit, not youtube.
    info = {
        "extractor_key": "Reddit",
        "webpage_url": "https://reddit.com/r/x/comments/y/youtube_is_dead",
    }
    assert resolve_adapter("url", None, info).key == "reddit"


def test_youtube_nocookie_host_still_gets_the_youtube_adapter():
    # An embed on youtube-nocookie.com is YouTube content and needs the POT/player-client strategy,
    # even though its host does not contain the substring "youtube.com".
    assert resolve_adapter("url", "https://www.youtube-nocookie.com/embed/abc123").key == "youtube"


def test_lookalike_hosts_do_not_misresolve_but_subdomains_do():
    # Netloc-scoped match: a look-alike host that merely CONTAINS the domain string is NOT the source.
    assert resolve_adapter("url", "https://myyoutube.com/watch?v=x").key == "generic"
    assert resolve_adapter("url", "https://youtube.com.evil.example/x").key == "generic"
    assert resolve_adapter("url", "https://notreddit.com/r/x").key == "generic"
    # …but real subdomains still resolve.
    assert resolve_adapter("url", "https://m.youtube.com/watch?v=x").key == "youtube"
    assert resolve_adapter("url", "https://music.youtube.com/watch?v=x").key == "youtube"


def test_webpage_url_netloc_identifies_source_when_extractor_key_absent():
    # The early-fill path passes info with no source_ref; the webpage_url netloc still resolves the
    # source even if extractor_key is missing from the progress-hook info_dict.
    assert (
        resolve_adapter(None, None, {"webpage_url": "https://www.facebook.com/reel/123"}).key
        == "facebook"
    )
    # A sibling source's name in the webpage_url PATH must not misresolve (host, not substring).
    info = {
        "extractor_key": "Reddit",
        "webpage_url": "https://www.reddit.com/r/x/comments/y/youtube",
    }
    assert resolve_adapter(None, None, info).key == "reddit"


def test_uploader_url_youtube_uses_channel_url():
    a = YouTubeAdapter()
    assert (
        a.uploader_url({"channel_url": "https://youtube.com/@chan"}) == "https://youtube.com/@chan"
    )
    assert (
        a.uploader_url({"uploader_url": "https://youtube.com/user/u"})
        == "https://youtube.com/user/u"
    )
    assert a.uploader_url({}) is None


def test_uploader_url_facebook_falls_back_to_uploader_id():
    a = FacebookAdapter()
    assert a.uploader_url({"uploader_id": "12345"}) == "https://www.facebook.com/12345"
    # A direct channel_url still wins over the id fallback.
    assert a.uploader_url({"channel_url": "https://facebook.com/pageX", "uploader_id": "9"}) == (
        "https://facebook.com/pageX"
    )
    assert a.uploader_url({}) is None


def test_uploader_url_reddit_and_generic_have_no_channel():
    assert RedditAdapter().uploader_url({"uploader_id": "99"}) is None  # not facebook → no fallback
    assert SourceAdapter().uploader_url({}) is None


def test_generic_source_url_is_passthrough_youtube_reconstructs():
    assert SourceAdapter().source_url("https://x/y") == "https://x/y"
    assert (
        YouTubeAdapter().source_url("VIDID12345_") == "https://www.youtube.com/watch?v=VIDID12345_"
    )


def test_client_sets_and_retriable_are_per_source():
    # Generic sources make a single attempt and never retry on the YouTube per-client markers.
    assert SourceAdapter().client_sets() == [None]
    assert SourceAdapter().client_sets(authed=True) == [None]
    assert SourceAdapter().retriable("Sign in to confirm you're not a bot") is False
    assert YouTubeAdapter().retriable("Sign in to confirm you're not a bot") is True
    assert YouTubeAdapter().retriable("some unrelated network blip") is False
    # The empty-set error of a cookie-incapable client list (see _COOKIE_CAPABLE): retriable, so an
    # admin who configures one still gets the next set tried instead of a hard failure.
    assert YouTubeAdapter().retriable("ERROR: No player clients have been requested") is True
    # The media fetch refused after a successful extraction: the format URLs belong to the client
    # that minted them, so the next client set is a real second chance (it used to end the job).
    assert (
        YouTubeAdapter().retriable(
            "ERROR: unable to download video data: HTTP Error 403: Forbidden"
        )
        is True
    )
    assert (
        SourceAdapter().retriable("ERROR: unable to download video data: HTTP Error 403") is False
    )
    # Facebook: a second, Chrome-impersonating attempt only WITH the user's cookie, and only behind
    # the parse wall (without a cookie that wall is the login wall — no disguise gets past it).
    fb = FacebookAdapter()
    assert fb.client_sets() == [None]
    assert len(fb.client_sets(authed=True)) == 2
    assert fb.ydl_opts_extra(fb.client_sets(authed=True)[0]) == {}
    assert fb.retriable("ERROR: [facebook] 1: Cannot parse data; please report this issue") is True
    assert fb.retriable("ERROR: [facebook] 1: No video formats found!") is False


def test_a_signed_in_youtube_download_only_runs_cookie_capable_clients(monkeypatch):
    yt = YouTubeAdapter()
    monkeypatch.setattr(settings, "download_player_clients", "android_vr")
    monkeypatch.setattr(settings, "download_player_clients_fallback", "web_safari,web,android")
    # Signed out: the admin's ladder exactly as configured.
    assert yt.client_sets() == [["android_vr"], ["web_safari", "web", "android"]]
    # Signed in: yt-dlp REMOVES every client that takes no cookies, and a set it empties that way
    # raises "No player clients have been requested" — which would fail the download on attempt 1,
    # before the POT-backed web fallback the cookie exists for. So the adapter never asks for one:
    # the android_vr-only set is gone and `android` is dropped out of the fallback.
    assert yt.client_sets(authed=True) == [["web_safari", "web"]]


def test_a_signed_in_download_still_has_clients_when_the_admin_configured_none(monkeypatch):
    # Both knobs cookie-incapable: fall back to the POT-backed web pair rather than no attempt at all.
    monkeypatch.setattr(settings, "download_player_clients", "android_vr")
    monkeypatch.setattr(settings, "download_player_clients_fallback", "android,ios")
    assert YouTubeAdapter().client_sets(authed=True) == [["web_safari", "web"]]


def test_the_cookie_capable_clients_match_the_installed_ytdlp():
    # The list is yt-dlp's, not ours (INNERTUBE_CLIENTS[...]["SUPPORTS_COOKIES"]). Pinned here so a
    # yt-dlp bump that renames or re-flags a client turns the gate red, instead of silently sending
    # a signed-in download to a client that drops the cookie.
    from yt_dlp.extractor.youtube._base import INNERTUBE_CLIENTS

    live = {k for k, v in INNERTUBE_CLIENTS.items() if v.get("SUPPORTS_COOKIES")}
    assert live == set(YouTubeAdapter._COOKIE_CAPABLE)
    assert set(YouTubeAdapter._AUTHED_DEFAULT) <= live


def test_youtube_ydl_opts_extra_carries_player_client_and_pot():
    ea = YouTubeAdapter().ydl_opts_extra(["android_vr"])["extractor_args"]
    assert ea["youtube"]["player_client"] == ["android_vr"]
    assert "youtubepot-bgutilhttp" in ea  # the bgutil PO-token base_url is wired
    # A client-less attempt still wires POT but no player_client key.
    assert "youtube" not in YouTubeAdapter().ydl_opts_extra(None)["extractor_args"]
    assert SourceAdapter().ydl_opts_extra(["android_vr"]) == {}  # generic contributes nothing


def test_wants_metadata_backfill_matches_old_should_refetch():
    assert YouTubeAdapter().wants_metadata_backfill() is True
    assert FacebookAdapter().wants_metadata_backfill() is True
    assert RedditAdapter().wants_metadata_backfill() is False
    assert SourceAdapter().wants_metadata_backfill() is False


def test_nfo_uniqueid_type_is_the_adapter_key():
    assert YouTubeAdapter().nfo_uniqueid_type() == "youtube"
    assert SourceAdapter().nfo_uniqueid_type() == "generic"


def test_title_is_yt_dlps_for_a_source_without_placeholders():
    assert SourceAdapter().title({"title": "Video", "description": "text"}) == "Video"
    assert SourceAdapter().title({}) is None


def test_facebook_placeholder_title_takes_the_posts_text():
    # Measured: signed in, yt-dlp titles every reel "Video" (an 18+ reel read as Chrome: "Facebook"),
    # while the post's text is still in the description.
    a = FacebookAdapter()
    info = {"title": "Video", "description": "\n  Tune of The Day\nsecond line"}
    assert a.title(info) == "Tune of The Day"
    assert a.title({**info, "title": "Facebook"}) == "Tune of The Day"
    # A real title (signed out) is left for normalize_title, engagement prefix and all.
    assert a.title({**info, "title": "456K views | Tune"}) == "456K views | Tune"
    # Nothing better on the page: keep what yt-dlp gave.
    assert a.title({"title": "Facebook", "description": None}) == "Facebook"


def test_facebook_title_from_a_long_description_is_the_whole_line_after_cleaning():
    # Dev UAT: a cut with "…" came out of normalize_title as "… live at." (NFKC "..." collapsed to "."),
    # a sentence that seemed to end. The whole first line survives the cleaning intact instead.
    line = 'Tune of The Day, Argy playing his version of the classic "Resurrection" with Paul Oakenfold live at Space'
    assert normalize_title(FacebookAdapter().title({"title": "Video", "description": line})) == line