"""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.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_youtube_only(): # Generic sources make a single attempt and never retry on the YouTube per-client markers. assert SourceAdapter().client_sets() == [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 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"