siftlode/backend/tests/test_facebook_impersonation.py

"""The Facebook download path with the user's cookie.

Measured on dev (yt-dlp 2026.07.04, the owner's cookie): reel 806379462330494 fails with the cookie
alone and extracts 2560p with `--impersonate chrome`; the reels that already download with the cookie
keep their plain first attempt. These pin the attempt order through the real download loop
(`pipeline._extract_with_fallback`, with yt-dlp faked at the `YoutubeDL` boundary), and that the
LOCKED curl_cffi can actually impersonate — the pair only works inside yt-dlp's own version range, and
outside it every target silently reports "unavailable".

Measured on dev and in the owner's Chrome: given a group post's link, yt-dlp finds no formats or
downloads another video from the page's feed. The post's own video is the single Video node in the
page scripts that carry the post id (prod posts 66, 145, 140, 143, 221; text posts have none while the
page holds ~10 other videos), and its watch link downloads.
"""

import pytest
import yt_dlp
from yt_dlp.networking.exceptions import TransportError
from yt_dlp.networking.impersonate import ImpersonateTarget

from app.downloads import adapters, pipeline
from app.downloads import errors as dlerrors

_URL = "https://www.facebook.com/reel/806379462330494/?fs=e&fs=e"
_POST = "1755988225525547"
_GROUP_LINK = f"https://www.facebook.com/groups/983554479435596/permalink/{_POST}/?"
_POST_VIDEO = "1408015447857753"
_FEED_VIDEO = "1056963063816966"  # what yt-dlp downloaded from this post's link: a feed video
_CHROME = ImpersonateTarget.from_str("chrome")


def _video_node(video_id: str) -> str:
    # The measured shape of a post's attachment in the page JSON.
    return '{"media":{"__typename":"Video","id":"%s","__isNode":"Video"},"styles":{}}' % video_id


def _page(*post_videos: str, post: str = _POST) -> str:
    attachments = ",".join(_video_node(v) for v in post_videos)
    return (
        '<html><script type="application/json" data-sjs>'
        f'{{"post_id":"{post}","creation_time":1785084554,"attachments":[{attachments}]}}</script>'
        '<script type="application/json" data-sjs>'
        f'{{"feed":[{_video_node(_FEED_VIDEO)}]}}</script></html>'
    )


_WALL = (
    "ERROR: [facebook] 806379462330494: Cannot parse data; please report this issue on  "
    "https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template."
)
_NO_FORMATS = "ERROR: [facebook] 2541212983053999: No video formats found!"


def test_the_locked_curl_cffi_can_impersonate_chrome():
    # YoutubeDL refuses to start when the target is unavailable — the same check a download makes.
    with yt_dlp.YoutubeDL({"impersonate": ImpersonateTarget.from_str("chrome"), "quiet": True}):
        pass


@pytest.fixture
def ydl(monkeypatch):
    """Fake `yt_dlp.YoutubeDL`: records each attempt's options and plays back `outcomes` in order;
    `web["page"]` is what a page read returns (`web["fail"]`, when set, is raised instead), and `web`
    records the pages opened, the byte cap each read asked for, and the links extracted."""
    seen: list[dict] = []
    outcomes: list = []
    web: dict = {"page": "", "fail": None, "opened": [], "reads": [], "extracted": []}

    class FakeYDL:
        def __init__(self, opts):
            seen.append(opts)

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def urlopen(self, url):
            web["opened"].append(url)
            if web["fail"] is not None:
                raise web["fail"]
            body = web["page"].encode()

            class Response:
                def read(self, amt=None):
                    web["reads"].append(amt)
                    return body

            return Response()

        def extract_info(self, url, download):
            web["extracted"].append(url)
            outcome = outcomes.pop(0)
            if isinstance(outcome, str):
                raise yt_dlp.utils.DownloadError(outcome)
            return outcome

    monkeypatch.setattr(pipeline.yt_dlp, "YoutubeDL", FakeYDL)
    monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
    monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **fields: None)
    return seen, outcomes, web


def _run(tmp_path, cookiefile, url=_URL):
    return pipeline._extract_with_fallback(
        1, 1, url, {}, tmp_path, adapters.FacebookAdapter(), cookiefile=cookiefile
    )


def test_a_cookie_download_at_the_wall_is_retried_as_chrome(ydl, tmp_path):
    seen, outcomes, web = ydl
    outcomes += [_WALL, {"id": "806379462330494"}]

    assert _run(tmp_path, "/cookies.txt") == {"id": "806379462330494"}

    assert [o.get("impersonate") for o in seen] == [None, _CHROME]
    assert all(o["cookiefile"] == "/cookies.txt" for o in seen)  # the retry stays signed in
    assert web["opened"] == []  # a reel is not a group post: no page read


def test_without_a_cookie_the_wall_is_the_login_wall_and_final(ydl, tmp_path):
    seen, outcomes, _ = ydl
    outcomes += [_WALL]

    with pytest.raises(yt_dlp.utils.DownloadError):
        _run(tmp_path, None)

    assert len(seen) == 1 and "impersonate" not in seen[0]


def test_other_facebook_failures_are_not_retried_as_chrome(ydl, tmp_path):
    seen, outcomes, _ = ydl
    outcomes += [_NO_FORMATS]

    with pytest.raises(yt_dlp.utils.DownloadError):
        _run(tmp_path, "/cookies.txt")

    assert len(seen) == 1 and "impersonate" not in seen[0]


# --- a group post downloads ITS video, or fails saying it has none --------------------------------


def test_a_cookie_group_post_downloads_the_posts_own_video(ydl, tmp_path):
    seen, outcomes, web = ydl
    web["page"] = _page(_POST_VIDEO)
    outcomes += [_WALL, {"id": _POST_VIDEO}]  # the watch link needs Chrome too (measured)

    assert _run(tmp_path, "/cookies.txt", _GROUP_LINK) == {"id": _POST_VIDEO}

    assert web["opened"] == [_GROUP_LINK]
    # Never an unbounded read.
    assert web["reads"] == [adapters.FacebookAdapter._POST_PAGE_MAX_BYTES]
    assert seen[0]["cookiefile"] == "/cookies.txt"  # the post page is read signed in
    watch = f"https://www.facebook.com/watch/?v={_POST_VIDEO}"
    # Never the post link: that is where the feed video comes from.
    assert web["extracted"] == [watch, watch]
    assert [o.get("impersonate") for o in seen[1:]] == [None, _CHROME]


def test_a_group_post_without_a_single_video_fails_saying_so(ydl, tmp_path):
    _, _, web = ydl
    web["page"] = _page()  # a text post: the only video on the page is the feed's

    with pytest.raises(RuntimeError) as exc:
        _run(tmp_path, "/cookies.txt", _GROUP_LINK)

    assert web["extracted"] == []  # yt-dlp is not left to guess from the post link
    assert dlerrors.diagnose(str(exc.value), url=_GROUP_LINK, cookie_attached=True) == (
        "format_unavailable",
        "facebook.group_post_video",
    )


def test_a_group_post_page_without_the_post_is_the_refused_cookie(ydl, tmp_path):
    # Measured with a refused cookie: Facebook answers the post link with its login page —
    # a login form, no post id anywhere. Saying "this post has no video" would blame the post.
    _, _, web = ydl
    web["page"] = '<html><title>ChatGPT magyarul</title><form id="login_form"></form></html>'

    with pytest.raises(RuntimeError) as exc:
        _run(tmp_path, "/cookies.txt", _GROUP_LINK)

    assert web["extracted"] == []
    assert dlerrors.diagnose(str(exc.value), url=_GROUP_LINK, cookie_attached=True) == (
        "login_required",
        "facebook.cookie_rejected",
    )


def test_a_failed_post_page_read_fails_the_download_as_itself(ydl, tmp_path):
    # A read that failed says nothing about the post: it must not turn into "no video in this post".
    _, _, web = ydl
    web["fail"] = TransportError("Connection reset by peer")

    with pytest.raises(TransportError) as exc:
        _run(tmp_path, "/cookies.txt", _GROUP_LINK)

    assert web["extracted"] == []
    assert dlerrors.diagnose(str(exc.value), url=_GROUP_LINK, cookie_attached=True) == (
        "network",
        None,
    )


def test_without_a_cookie_the_group_post_link_goes_to_ytdlp_unchanged(ydl, tmp_path):
    _, outcomes, web = ydl
    outcomes += [_NO_FORMATS]

    with pytest.raises(yt_dlp.utils.DownloadError):
        _run(tmp_path, None, _GROUP_LINK)

    assert web["opened"] == []  # signed out the page is the login wall: nothing to read
    assert web["extracted"] == [_GROUP_LINK]


def test_the_group_post_link_forms_are_recognised():
    for url in (
        _GROUP_LINK,
        f"https://www.facebook.com/groups/983554479435596/posts/{_POST}/",
        f"https://m.facebook.com/groups/983554479435596/permalink/{_POST}/?",
        f"https://www.facebook.com/groups/chatgptmagyarul/posts/{_POST}",
    ):
        assert adapters.facebook_group_post_id(url) == _POST, url
    for url in (
        _URL,
        f"https://www.facebook.com/watch/?v={_POST_VIDEO}",
        f"https://facebook.com.evil/groups/x/permalink/{_POST}/",
        "https://www.facebook.com/groups/983554479435596/",
    ):
        assert adapters.facebook_group_post_id(url) is None, url
    assert adapters.SourceAdapter().resolve_url(_GROUP_LINK, {"cookiefile": "/c"}) == _GROUP_LINK


def test_only_the_single_video_of_the_posts_own_blocks_counts():
    assert adapters.facebook_post_video_id(_page(_POST_VIDEO), _POST) == _POST_VIDEO
    # No video in the post, although the page has one elsewhere.
    assert adapters.facebook_post_video_id(_page(), _POST) is None
    # Two videos: guessing between them would be the wrong-video bug again.
    assert adapters.facebook_post_video_id(_page(_POST_VIDEO, "2063302677911049"), _POST) is None
    # A longer number that merely STARTS with the post id is another object.
    assert adapters.facebook_post_video_id(_page(_POST_VIDEO, post=_POST + "0"), _POST) is None