siftlode/backend/tests/test_download_failure_catalog.py

"""DB-backed tests for the download failure catalog (E11.5 S1) at the places a failure is WRITTEN.

`test_download_errors` pins which entry a message matches; these pin that every writer hands the
matcher what it needs — the link and whether the user's cookie went with it — and that a stored entry
never outlives the failure it describes:
- the download child's failure branch (`pipeline._download`);
- a second job meeting an already-failed asset (`worker._process_job`), which only has the text;
- Retry / Re-download, which requeue the job;
- the `0084` backfill of the rows that failed before the catalog existed;
- the downloads list, which is how the entry reaches the UI.
It also holds the one success-path check of the same child: the duration it stores (E11.5 S3).
"""

import shutil
import subprocess

import pytest
from alembic import command
from alembic.config import Config
from cryptography.fernet import Fernet
from sqlalchemy import text

from app import security
from app.config import settings
from app.db import engine
from app.downloads import credentials, formats, pipeline, service
from app.models import DownloadJob, DownloadQuota, MediaAsset, User
from app.worker import _process_job

_REEL = "https://www.facebook.com/reel/1736360027626346/?fs=e&fs=e"
# Verbatim yt-dlp 2026.07.04, prod job 256.
_FB_PARSE = (
    "ERROR: [facebook] 1736360027626346: Cannot parse data; please report this issue on  "
    "https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template."
)
_COOKIES = "# Netscape HTTP Cookie File\n.facebook.com\tTRUE\t/\tTRUE\t0\tc_user\t123\n"


@pytest.fixture
def _fernet(monkeypatch):
    # A throwaway key, so a stored credential encrypts and decrypts (mirrors test_download_credentials).
    monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))


def _user(db, email):
    u = User(email=email)
    db.add(u)
    db.flush()
    db.add(
        DownloadQuota(
            user_id=u.id, max_bytes=10**12, max_concurrent=5, max_jobs=10**6, unlimited=True
        )
    )
    return u


def _asset_and_job(db, user, *, sig, asset_status, job_status, error=None):
    a = MediaAsset(
        source_kind="url",
        source_ref=_REEL,
        format_sig=sig,
        status=asset_status,
        error=error,
        ref_count=1,
    )
    db.add(a)
    db.flush()
    j = DownloadJob(
        user_id=user.id,
        asset_id=a.id,
        source_kind="url",
        source_ref=_REEL,
        status=job_status,
        queue_pos=0,
        profile_snapshot={},
    )
    db.add(j)
    db.commit()
    return a, j


def _fail_the_child(monkeypatch, tmp_path, *, cookie: bool):
    monkeypatch.setattr(settings, "download_root", str(tmp_path))
    monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)
    cookiefile = tmp_path / "cookies.txt"
    cookiefile.write_text("# Netscape HTTP Cookie File\n")
    monkeypatch.setattr(
        pipeline.credentials,
        "resolve_cookiefile",
        lambda *a, **k: cookiefile if cookie else None,
    )

    def extract(*a, **k):
        raise RuntimeError(_FB_PARSE)

    monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)


def test_the_download_child_diagnoses_with_the_link_and_no_cookie(db, monkeypatch, tmp_path):
    u = _user(db, "fb-nocookie@example.com")
    a, j = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
    _fail_the_child(monkeypatch, tmp_path, cookie=False)

    pipeline._download(j.id, a.id, "url", _REEL, {}, u.id)

    db.expire_all()
    job = db.get(DownloadJob, j.id)
    assert (job.status, job.error_code, job.error_detail) == (
        "error",
        "login_required",
        "facebook.login_wall",
    )


def test_the_download_child_knows_the_cookie_went_with_it(db, monkeypatch, tmp_path):
    # The same text with the user's cookie attached: telling them to add cookies would be wrong.
    u = _user(db, "fb-cookie@example.com")
    a, j = _asset_and_job(db, u, sig="private", asset_status="downloading", job_status="running")
    _fail_the_child(monkeypatch, tmp_path, cookie=True)

    pipeline._download(j.id, a.id, "url", _REEL, {}, u.id)

    db.expire_all()
    job = db.get(DownloadJob, j.id)
    assert (job.error_code, job.error_detail) == ("login_required", "facebook.cookie_rejected")


def test_a_failure_the_database_would_not_record_is_not_claimed(db, monkeypatch, tmp_path):
    """The job is still `running`: the runner reports this with the write's error, and the worker
    fails the job. A `download.job.failed` here as well would be a second line, and a false one."""
    u = _user(db, "fb-dbdown@example.com")
    a, j = _asset_and_job(db, u, sig="dbdown", asset_status="downloading", job_status="running")
    _fail_the_child(monkeypatch, tmp_path, cookie=False)
    said = []
    monkeypatch.setattr(pipeline, "_log_download_failed", lambda job_id, *rest: said.append(job_id))

    def db_down(*a, **k):
        raise ConnectionError("the database went away")

    monkeypatch.setattr(pipeline.service, "finalize_if_running", db_down)

    with pytest.raises(ConnectionError):
        pipeline._download(j.id, a.id, "url", _REEL, {}, u.id)
    assert said == []


def test_a_job_meeting_a_failed_shared_asset_gets_the_same_explanation(db):
    # Before the catalog this path wrote the text and NO code: "Download failed" for the second
    # job, "Facebook sign-in required" for the first.
    u = _user(db, "fb-second@example.com")
    _, j = _asset_and_job(
        db, u, sig="shared", asset_status="error", job_status="running", error=_FB_PARSE
    )

    _process_job(j.id)

    db.expire_all()
    job = db.get(DownloadJob, j.id)
    assert (job.status, job.error_code, job.error_detail) == (
        "error",
        "login_required",
        "facebook.login_wall",
    )


def test_a_job_meeting_its_users_failed_private_asset_knows_the_cookie_was_used(db, _fernet):
    u = _user(db, "fb-private@example.com")
    credentials.upsert_credential(db, u.id, "facebook", _COOKIES)
    sig = credentials.authed_format_sig(formats.normalize({}), u.id, "facebook")
    _, j = _asset_and_job(
        db, u, sig=sig, asset_status="error", job_status="running", error=_FB_PARSE
    )

    _process_job(j.id)

    db.expire_all()
    job = db.get(DownloadJob, j.id)
    assert (job.error_code, job.error_detail) == ("login_required", "facebook.cookie_rejected")


def test_a_private_asset_the_cookie_could_not_go_with_is_explained_as_the_download_had_it(db):
    # The cookie was deleted (or can no longer be decrypted) after the jobs were queued: the download
    # child had nothing to attach and said "add your cookies". The job meeting that failed asset must say
    # the same — the asset's salt alone would have told it "your cookies were refused".
    u = _user(db, "fb-gone@example.com")
    sig = credentials.authed_format_sig(formats.normalize({}), u.id, "facebook")
    _, j = _asset_and_job(
        db, u, sig=sig, asset_status="error", job_status="running", error=_FB_PARSE
    )

    _process_job(j.id)

    db.expire_all()
    job = db.get(DownloadJob, j.id)
    assert (job.error_code, job.error_detail) == ("login_required", "facebook.login_wall")


def test_a_job_canceled_before_it_meets_a_failed_asset_stays_canceled(db):
    # The cancel landed between the claim and this write. `error` would hold the asset again after the
    # route already released it — a later delete releases twice.
    u = _user(db, "fb-canceled@example.com")
    _, j = _asset_and_job(
        db, u, sig="shared", asset_status="error", job_status="canceled", error=_FB_PARSE
    )

    _process_job(j.id)

    db.expire_all()
    job = db.get(DownloadJob, j.id)
    assert (job.status, job.error_code, job.error_detail) == ("canceled", None, None)


def test_the_downloads_list_carries_the_catalog_entry(db, as_user):
    # The whole explanation rides on this one field: without it every row falls back to the category.
    u = _user(db, "fb-list@example.com")
    _, j = _asset_and_job(
        db, u, sig="shared", asset_status="error", job_status="error", error=_FB_PARSE
    )
    j.error_code, j.error_detail = "login_required", "facebook.login_wall"
    db.commit()

    [row] = as_user(u).get("/api/downloads").json()

    assert (row["error_code"], row["error_detail"]) == ("login_required", "facebook.login_wall")


@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
def test_a_download_yt_dlp_gave_no_duration_stores_the_files_duration(db, monkeypatch, tmp_path):
    # Facebook reel 257 (E11.5): yt-dlp's info had no duration, the asset stored NULL, and the Library
    # row lost its trim action. Through the real finalize, with a real 3-second file.
    u = _user(db, "fb-duration@example.com")
    a, j = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
    monkeypatch.setattr(settings, "download_root", str(tmp_path))
    monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)

    def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
        media = staging / "1736360027626346.mp4"
        subprocess.run(
            [
                "ffmpeg",
                "-y",
                "-loglevel",
                "error",
                "-f",
                "lavfi",
                "-i",
                "testsrc=size=64x64:rate=10",
                "-f",
                "lavfi",
                "-i",
                "sine=frequency=440",
                "-t",
                "3",
                "-c:v",
                "libx264",
                "-pix_fmt",
                "yuv420p",
                "-c:a",
                "aac",
                str(media),
            ],
            check=True,
        )
        return {"id": "1736360027626346", "ext": "mp4", "title": "Reel", "filepath": str(media)}

    monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)

    pipeline._download(j.id, a.id, "url", _REEL, {}, u.id)

    db.expire_all()
    assert db.get(DownloadJob, j.id).status == "done"
    assert db.get(MediaAsset, a.id).duration_s == 3


def test_retry_and_redownload_drop_the_old_explanation(db):
    u = _user(db, "fb-requeue@example.com")
    for requeue in (service.resume, service.redownload):
        _, j = _asset_and_job(
            db, u, sig=f"s-{requeue.__name__}", asset_status="error", job_status="error"
        )
        j.error, j.error_code, j.error_detail = _FB_PARSE, "login_required", "facebook.login_wall"
        db.commit()

        requeue(db, j)

        db.refresh(j)
        assert (j.status, j.error_code, j.error_detail) == ("queued", None, None), requeue.__name__


def test_0084_backfills_only_the_facebook_parse_failures_it_measured(db):
    cfg = Config("alembic.ini")
    command.downgrade(cfg, "0083_comfort_pins")
    upgraded = False
    try:
        with engine.begin() as conn:
            uid = conn.execute(
                text("INSERT INTO users (email) VALUES ('backfill@example.com') RETURNING id")
            ).scalar_one()
            rows = {
                "fb_unknown": ("error", "unknown", _FB_PARSE),
                "fb_null": ("canceled", None, _FB_PARSE),
                "other_site": ("error", "unknown", _FB_PARSE.replace("[facebook]", "[vimeo]")),
                "fb_other_code": ("error", "network", _FB_PARSE),
            }
            ids = {}
            for name, (status, code, error) in rows.items():
                ids[name] = conn.execute(
                    text(
                        "INSERT INTO download_jobs (user_id, source_kind, source_ref, "
                        "profile_snapshot, status, error_code, error) "
                        "VALUES (:u, 'url', :ref, '{}', :s, :c, :e) RETURNING id"
                    ),
                    {"u": uid, "ref": _REEL, "s": status, "c": code, "e": error},
                ).scalar_one()
        command.upgrade(cfg, "head")
        upgraded = True
        with engine.connect() as conn:
            got = {
                name: tuple(
                    conn.execute(
                        text("SELECT error_code, error_detail FROM download_jobs WHERE id = :i"),
                        {"i": i},
                    ).one()
                )
                for name, i in ids.items()
            }
        assert got == {
            "fb_unknown": ("login_required", "facebook.login_wall"),
            "fb_null": ("login_required", "facebook.login_wall"),
            "other_site": ("unknown", None),
            "fb_other_code": ("network", None),
        }
    finally:
        if not upgraded:
            command.upgrade(cfg, "head")


def _ffmpeg_media(path, *codec_args):
    subprocess.run(
        ["ffmpeg", "-y", "-loglevel", "error", "-f", "lavfi", "-i", "sine=frequency=440"]
        + list(codec_args)
        + ["-t", "2", str(path)],
        check=True,
    )


@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
def test_a_signed_in_facebook_download_is_titled_with_its_posts_text(db, monkeypatch, tmp_path):
    # Prod jobs 265–268 (2026-09-14): with a cookie yt-dlp titles every reel "Video", so the row, the
    # file name and the share preview all said "Video". The post's text is in the description.
    u = _user(db, "fb-title@example.com")
    a, j = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
    monkeypatch.setattr(settings, "download_root", str(tmp_path))
    monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)

    def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
        media = staging / "2131884261040652.m4a"
        _ffmpeg_media(media, "-c:a", "aac")
        return {
            "id": "2131884261040652",
            "ext": "m4a",
            "title": "Video",
            "description": "Tune of The Day, Argy playing his version",
            "extractor_key": "Facebook",
            "filepath": str(media),
        }

    monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)

    pipeline._download(j.id, a.id, "url", _REEL, {"mode": "a"}, u.id)

    db.expire_all()
    asset, job = db.get(MediaAsset, a.id), db.get(DownloadJob, j.id)
    assert asset.title == job.display_name == "Tune of The Day, Argy playing his version"
    assert "Tune_of_The_Day" in asset.rel_path


def test_the_early_title_takes_the_posts_text_too(db):
    u = _user(db, "fb-early-title@example.com")
    a, _ = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
    a.title = None
    db.commit()

    pipeline._fill_asset_meta_early(
        a.id,
        {
            "id": "1",
            "title": "Video",
            "description": "Almás palacsinta",
            "extractor_key": "Facebook",
        },
    )

    db.expire_all()
    assert db.get(MediaAsset, a.id).title == "Almás palacsinta"


@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
def test_an_mp3_download_is_stored_as_an_mp3(db, monkeypatch, tmp_path):
    # Prod job 268: the "Audio (MP3)" profile's extracted .mp3 was remuxed into an .mp4 and served as
    # video/mp4, which iPad's WebKit will not play. It stays an .mp3 (served as audio/mpeg).
    u = _user(db, "fb-mp3@example.com")
    a, j = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
    monkeypatch.setattr(settings, "download_root", str(tmp_path))
    monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)

    def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
        media = staging / "2131884261040652.mp3"
        _ffmpeg_media(media, "-c:a", "libmp3lame")
        return {"id": "2131884261040652", "ext": "mp3", "title": "Reel", "filepath": str(media)}

    monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)

    pipeline._download(
        j.id, a.id, "url", _REEL, formats.normalize({"mode": "a", "audio_format": "mp3"}), u.id
    )

    db.expire_all()
    asset = db.get(MediaAsset, a.id)
    assert (asset.container, asset.acodec, asset.media_kind) == ("mp3", "mp3", "audio")
    assert asset.rel_path.endswith(".mp3")


@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
@pytest.mark.parametrize("mode", ["av", "v"])
def test_a_video_download_is_still_made_browser_playable(db, monkeypatch, tmp_path, mode):
    # The mp3 fix skips the browser-playable pass for AUDIO only: a video that arrives in a container
    # browsers won't play (an .mkv) is still remuxed into an .mp4.
    u = _user(db, f"fb-mkv-{mode}@example.com")
    a, j = _asset_and_job(
        db, u, sig=f"mkv-{mode}", asset_status="downloading", job_status="running"
    )
    monkeypatch.setattr(settings, "download_root", str(tmp_path))
    monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)

    def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
        media = staging / "1736360027626346.mkv"
        audio = ["-f", "lavfi", "-i", "sine=frequency=440", "-c:a", "aac"] if mode == "av" else []
        subprocess.run(
            [
                "ffmpeg",
                "-y",
                "-loglevel",
                "error",
                "-f",
                "lavfi",
                "-i",
                "testsrc=size=64x64:rate=10",
            ]
            + audio
            + ["-t", "1", "-c:v", "libx264", "-pix_fmt", "yuv420p", str(media)],
            check=True,
        )
        return {"id": "1736360027626346", "ext": "mkv", "title": "Reel", "filepath": str(media)}

    monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)

    pipeline._download(j.id, a.id, "url", _REEL, formats.normalize({"mode": mode}), u.id)

    db.expire_all()
    asset = db.get(MediaAsset, a.id)
    assert (asset.container, asset.vcodec) == ("mp4", "h264")
    assert asset.rel_path.endswith(".mp4")