siftlode/backend/tests/test_facebook_titles.py

"""A Facebook download's title when the signed-in read gives only a placeholder.

Measured on prod (2026-11-01): every Facebook download since 2026-10-28 is titled "Facebook" — the
signed-in read (the owner's cookie) now gives that placeholder and NO description, so the adapter had
no post text to fall back on, and the download name copied the placeholder too. A signed-out read of
the same reel (yt-dlp 2026.08.19, as on prod) still carries the post's text in its description.
"""

from sqlalchemy import select, update

from app.config import settings
from app.db import SessionLocal
from app.downloads import adapters, backfill, egress, pipeline
from app.downloads.models import DownloadJob, MediaAsset

_URL = "https://www.facebook.com/reel/1860682468422402/"
# The page yt-dlp says it read for it (`webpage_url`); a group post's link resolves the same way.
_PAGE = "https://www.facebook.com/watch/?v=1860682468422402"
# The measured signed-out read of that reel.
_PUBLIC = {
    "title": "163K reactions · 14K shares | sorry not helping #workhumor #officelife | Work dot exe",
    "description": "sorry not helping\n\n#workhumor #officelife",
    "uploader": "Work dot exe",
}
# A signed-out read carries the upload date too (measured 2026-11-14 on watch?v=1140126121708919:
# "20261113"), which the signed-in read no longer gives.
_PUBLIC_DATED = {**_PUBLIC, "upload_date": "20261113"}


def _fake_ydl(monkeypatch, outcome):
    """yt-dlp faked at its `YoutubeDL` boundary: records each read (its options and its `url`),
    answers `outcome`."""
    seen: list[dict] = []

    class FakeYDL:
        def __init__(self, opts):
            self.opts = opts

        def __enter__(self):
            return self

        def add_post_processor(self, pp, when="post_process"):
            pass  # the egress format filter; nothing here to filter

        def __exit__(self, *exc):
            return False

        def extract_info(self, url, download):
            seen.append({**self.opts, "url": url})
            return outcome

    monkeypatch.setattr("yt_dlp.YoutubeDL", FakeYDL)
    return seen


def test_a_placeholder_with_nothing_behind_it_takes_the_signed_out_pages_text(monkeypatch):
    seen = _fake_ydl(monkeypatch, _PUBLIC)
    fb = adapters.FacebookAdapter()

    info = backfill.public_page_info({"title": "Facebook", "description": None}, _URL, fb, 7)

    assert fb.title(info) == "sorry not helping"
    # One metadata-only read, and signed out: no cookie goes with it.
    assert len(seen) == 1
    assert seen[0]["skip_download"] is True
    assert not seen[0].get("cookiefile")
    assert seen[0][egress.GUARDED] is True and seen[0]["proxy"] == egress.proxy_url()


def test_the_signed_out_read_also_lends_the_upload_date(monkeypatch):
    # Measured on prod: since 2026-10-28 no Facebook download has an upload date (37 before had one,
    # 31 since none) — the signed-in read lost it with the title. The signed-out read still has it.
    _fake_ydl(monkeypatch, _PUBLIC_DATED)

    info = backfill.public_page_info(
        {"title": "Facebook", "description": None}, _URL, adapters.FacebookAdapter(), 7
    )

    assert info["upload_date"] == "20261113"


def test_a_date_the_signed_in_read_gave_stays(monkeypatch):
    _fake_ydl(monkeypatch, _PUBLIC_DATED)

    info = backfill.public_page_info(
        {"title": "Facebook", "description": None, "upload_date": "20261001"},
        _URL,
        adapters.FacebookAdapter(),
        7,
    )

    assert info["upload_date"] == "20261001"


def test_a_real_title_reads_nothing_more(monkeypatch):
    seen = _fake_ydl(monkeypatch, _PUBLIC)
    info = {"title": "Funny Dark Humor Dad Joke!", "description": None}

    assert backfill.public_page_info(info, _URL, adapters.FacebookAdapter(), 7) is info
    assert seen == []


def test_a_placeholder_with_the_posts_text_behind_it_reads_nothing_more(monkeypatch):
    seen = _fake_ydl(monkeypatch, _PUBLIC)
    info = {"title": "Video", "description": "Office jokes, part two"}

    assert backfill.public_page_info(info, _URL, adapters.FacebookAdapter(), 7) is info
    assert seen == []


def test_a_placeholder_the_signed_out_page_cannot_better_stays(monkeypatch):
    _fake_ydl(monkeypatch, {"title": "Facebook", "description": None})
    fb = adapters.FacebookAdapter()

    info = backfill.public_page_info({"title": "Facebook", "description": None}, _URL, fb, 7)

    assert fb.title(info) == "Facebook"


def test_another_source_is_never_read_again(monkeypatch):
    seen = _fake_ydl(monkeypatch, _PUBLIC)
    info = {"title": "Video", "description": None}

    assert backfill.public_page_info(info, _URL, adapters.SourceAdapter(), 7) is info
    assert seen == []


def test_the_backfill_retitles_a_placeholder_download_and_the_names_that_repeated_it(
    db, make_user, monkeypatch
):
    seen = _fake_ydl(monkeypatch, _PUBLIC)
    owner, other = make_user(), make_user()
    asset = MediaAsset(
        source_kind="url",
        source_ref=_URL,
        format_sig="fb",
        status="ready",
        rel_path="facebook/x.mp4",
        poster_path="facebook/x.jpg",
        uploader_url="https://www.facebook.com/1",
        title="Facebook",
        ref_count=2,
    )
    db.add(asset)
    db.flush()
    copied = DownloadJob(
        user_id=owner.id,
        asset_id=asset.id,
        source_kind="url",
        source_ref=_URL,
        status="done",
        queue_pos=0,
        profile_snapshot={},
        display_name="Facebook",
    )
    renamed = DownloadJob(
        user_id=other.id,
        asset_id=asset.id,
        source_kind="url",
        source_ref=_URL,
        status="done",
        queue_pos=0,
        profile_snapshot={},
        display_name="Office jokes",
    )
    db.add_all([copied, renamed])
    db.commit()

    assert backfill.run()["retitled"] == 1
    assert len(seen) == 1

    db.expire_all()
    assert db.get(MediaAsset, asset.id).title == "sorry not helping"
    assert db.get(DownloadJob, copied.id).display_name == "sorry not helping"
    assert db.get(DownloadJob, renamed.id).display_name == "Office jokes"
    # Run again: nothing left to retitle, and no page read for it.
    assert backfill.run()["retitled"] == 0
    assert len(seen) == 1


def test_a_name_given_while_the_backfill_runs_stays(db, make_user, monkeypatch):
    # The owner renames the download while the backfill reads the page: the owner's name wins, it is
    # not overwritten by what was read before it.
    _fake_ydl(monkeypatch, _PUBLIC)
    owner = make_user()
    asset = MediaAsset(
        source_kind="url",
        source_ref=_URL,
        format_sig="fb",
        status="ready",
        rel_path="facebook/x.mp4",
        uploader_url="https://www.facebook.com/1",
        title="Facebook",
        ref_count=1,
    )
    db.add(asset)
    db.flush()
    job = DownloadJob(
        user_id=owner.id,
        asset_id=asset.id,
        source_kind="url",
        source_ref=_URL,
        status="done",
        queue_pos=0,
        profile_snapshot={},
        display_name="Facebook",
    )
    db.add(job)
    db.commit()

    read_page = backfill.public_page_info

    def renamed_while_reading(*args):
        with SessionLocal() as other:
            other.execute(
                update(DownloadJob)
                .where(DownloadJob.id == job.id)
                .values(display_name="Office jokes")
            )
            other.commit()
        return read_page(*args)

    monkeypatch.setattr(backfill, "public_page_info", renamed_while_reading)
    monkeypatch.setattr(backfill.editmod, "ensure_poster", lambda a, root: None)

    assert backfill.run()["retitled"] == 1

    db.expire_all()
    assert db.get(DownloadJob, job.id).display_name == "Office jokes"


def test_the_poster_step_holds_no_row_lock(db, make_user, monkeypatch):
    # It can cut a frame for 30 s. A delete locks the asset and then the job: with the job held here
    # and the asset still to write, that is a deadlock, and a rename waits out the whole cut.
    _fake_ydl(monkeypatch, _PUBLIC)
    owner = make_user()
    asset = MediaAsset(
        source_kind="url",
        source_ref=_URL,
        format_sig="fb",
        status="ready",
        rel_path="facebook/x.mp4",
        uploader_url="https://www.facebook.com/1",
        title="Facebook",
        ref_count=1,
    )
    db.add(asset)
    db.flush()
    job = DownloadJob(
        user_id=owner.id,
        asset_id=asset.id,
        source_kind="url",
        source_ref=_URL,
        status="done",
        queue_pos=0,
        profile_snapshot={},
        display_name="Facebook",
    )
    db.add(job)
    db.commit()
    asset_id, job_id = asset.id, job.id

    def poster_finds_rows_free(a, root):
        with SessionLocal() as other:
            other.execute(
                select(MediaAsset).where(MediaAsset.id == asset_id).with_for_update(nowait=True)
            )
            other.execute(
                select(DownloadJob).where(DownloadJob.id == job_id).with_for_update(nowait=True)
            )
            other.rollback()
        return None

    monkeypatch.setattr(backfill.editmod, "ensure_poster", poster_finds_rows_free)

    assert backfill.run()["retitled"] == 1


def test_a_download_titled_facebook_is_named_from_the_signed_out_page(
    db, make_user, monkeypatch, tmp_path
):
    # The whole download path with yt-dlp and the media tools faked: the signed-in read gives only
    # the placeholder, and the file and the download's name get the post's text — and its date.
    seen = _fake_ydl(monkeypatch, _PUBLIC_DATED)
    monkeypatch.setattr(settings, "files_root", str(tmp_path))
    monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)
    monkeypatch.setattr(pipeline, "_ensure_phone_safe", lambda _j, produced, *_a: produced)
    monkeypatch.setattr(
        pipeline.probe, "resolve_media_facts", lambda produced, info: ("h264", "aac", 720, 1280, 9)
    )
    monkeypatch.setattr(pipeline.editmod, "thumbnail_to_jpg", lambda *a: None)
    monkeypatch.setattr(pipeline.editmod, "ensure_poster", lambda *a: None)
    owner = make_user()
    asset = MediaAsset(
        source_kind="url", source_ref=_URL, format_sig="fb", status="downloading", ref_count=1
    )
    db.add(asset)
    db.flush()
    job = DownloadJob(
        user_id=owner.id,
        asset_id=asset.id,
        source_kind="url",
        source_ref=_URL,
        status="running",
        queue_pos=0,
        profile_snapshot={},
    )
    db.add(job)
    db.commit()

    def signed_in_read(job_id, asset_id, url, spec, staging, adapter, **kw):
        f = staging / "1860682468422402.mp4"
        f.write_bytes(b"v" * 100)
        return {
            "id": "1860682468422402",
            "title": "Facebook",
            "uploader": "Work dot exe",
            "webpage_url": _PAGE,
            "requested_downloads": [{"filepath": str(f)}],
        }

    monkeypatch.setattr(pipeline, "_extract_with_fallback", signed_in_read)
    pipeline._download(job.id, asset.id, "url", _URL, {}, owner.id)

    # The signed-out read goes to the page yt-dlp read, not the link queued: a group post's link
    # resolves to its video, and signed out the post itself is the login wall.
    assert [r["url"] for r in seen] == [_PAGE]
    db.expire_all()
    assert db.get(DownloadJob, job.id).status == "done"
    assert db.get(MediaAsset, asset.id).title == "sorry not helping"
    assert db.get(DownloadJob, job.id).display_name == "sorry not helping"
    done = db.get(MediaAsset, asset.id)
    assert done.upload_date is not None and done.upload_date.isoformat() == "2026-11-13"
    # The date names the file and its season, not 0000-00-00 / Season_0.
    assert done.rel_path.startswith("Work_dot_exe/Season_2026/Work_dot_exe_-_2026-11-13_-_")