siftlode/backend/tests/test_formats.py

"""build_ydl_opts: EVERY download is clamped to the mp4/H.264 compat profile (the browser-
playability guarantee) and must SELECT H.264 (not merely prefer it), so shared downloads play in
every browser incl. iOS/Safari. Non-mp4 containers and non-H.264 vcodecs are coerced, not honoured.
See the iPad AV1 broken-player fix + the format narrowing."""

import contextlib
import functools
import http.server
import threading

import yt_dlp

from app.downloads import edit as editmod
from app.downloads import formats


def _opts(spec):
    return formats.build_ydl_opts(spec, "/tmp/%(id)s.%(ext)s", lambda d: None)


def test_adapter_extra_opts_are_merged_in():
    # The SourceAdapter contributes source-specific opts (e.g. YouTube's extractor_args) via `extra`.
    extra = {"extractor_args": {"youtube": {"player_client": ["web"]}}}
    opts = formats.build_ydl_opts({}, "/tmp/%(id)s.%(ext)s", lambda d: None, extra=extra)
    assert opts["extractor_args"]["youtube"]["player_client"] == ["web"]
    # No extra → no extractor_args (a generic source contributes nothing).
    assert "extractor_args" not in formats.build_ydl_opts({}, "/tmp/x", lambda d: None)


def test_mp4_default_selects_avc1_with_progressive_fallback():
    fmt = _opts({"mode": "av", "container": "mp4"})["format"]
    # avc1 video-only first, then the avc1 progressive, then an unrestricted fallback.
    assert fmt.startswith("bestvideo[vcodec^=avc1]+bestaudio[acodec^=mp4a]")
    assert "/best[vcodec^=avc1]" in fmt
    assert fmt.endswith("bestvideo+bestaudio/best")


def test_format_sig_salt_forks_a_distinct_identity():
    spec = {"mode": "av", "container": "mp4", "max_height": 720}
    base = formats.format_sig(spec)
    # A salt yields a distinct cache identity (force-redownload's private-asset fork); different
    # salts differ from each other and from the unsalted sig, same salt is stable.
    assert formats.format_sig(spec, salt="7:12") != base
    assert formats.format_sig(spec, salt="7:12") != formats.format_sig(spec, salt="7:13")
    assert formats.format_sig(spec, salt="7:12") == formats.format_sig(spec, salt="7:12")


def test_mp4_default_keeps_h264_sort_tiebreaker():
    sort = _opts({"mode": "av", "container": "mp4"})["format_sort"]
    assert "vcodec:h264" in sort and "acodec:aac" in sort


def test_video_only_mp4_selects_avc1():
    fmt = _opts({"mode": "v", "container": "mp4"})["format"]
    assert fmt.startswith("bestvideo[vcodec^=avc1]")
    assert "+bestaudio" not in fmt  # video-only never merges audio
    # A video-only download stays audio-free: every `bestvideo` (video-only) option is preferred
    # over any `best` (progressive, audio-bearing) fallback.
    assert fmt.index("bestvideo") < fmt.index("best[vcodec^=avc1]") < fmt.rindex("/best")


def test_max_height_caps_every_branch():
    fmt = _opts({"mode": "av", "container": "mp4", "max_height": 720})["format"]
    assert fmt.count("[height<=?720]") >= 3  # applied to each avc1/fallback branch


def test_explicit_vcodec_is_clamped_to_the_compat_selector():
    # Browser-playability guarantee: a non-H.264 vcodec no longer opts out — normalize() coerces
    # vcodec→None, so the avc1 selector (and mp4 merge) is used regardless of what was requested.
    opts = _opts({"mode": "av", "container": "mp4", "vcodec": "vp9"})
    assert opts["format"].startswith("bestvideo[vcodec^=avc1]")
    assert opts["merge_output_format"] == "mp4"
    assert "vcodec:vp9" not in opts["format_sort"]


def test_non_mp4_container_is_clamped_to_mp4():
    # A webm/mkv request is coerced to the mp4/H.264 compat profile (nothing plays mkv in-browser;
    # webm/VP9 fails iOS), so the avc1 selector is used and the merge target is mp4.
    opts = _opts({"mode": "av", "container": "webm"})
    assert opts["format"].startswith("bestvideo[vcodec^=avc1]")
    assert opts["merge_output_format"] == "mp4"


def test_normalize_clamps_to_browser_playable():
    n = formats.normalize({"container": "mkv", "vcodec": "av1", "audio_format": "opus"})
    assert n["container"] == "mp4"
    assert n["vcodec"] is None
    assert n["audio_format"] == "m4a"
    # A playable audio choice is preserved.
    assert formats.normalize({"audio_format": "mp3"})["audio_format"] == "mp3"


def test_non_playable_specs_dedup_to_the_compat_identity():
    # Two requests that differ only in (now-clamped) non-playable fields share ONE cache asset.
    a = formats.format_sig({"mode": "av", "max_height": 720, "container": "webm", "vcodec": "av1"})
    b = formats.format_sig({"mode": "av", "max_height": 720, "container": "mp4"})
    assert a == b


def test_audio_only_unaffected():
    assert _opts({"mode": "a", "audio_format": "m4a"})["format"] == "bestaudio/best"


# --- SEC-13 · one enqueue is one bounded download -------------------------------------------


def test_a_playlist_url_can_only_yield_one_item():
    # `noplaylist` alone only demotes a video-inside-a-playlist URL to that video; a URL that is
    # purely a playlist or channel has no such video, so without this it expands into every entry.
    opts = _opts({})
    assert opts["noplaylist"] is True
    assert opts["playlist_items"] == "1"


def test_the_remaining_quota_becomes_a_hard_size_ceiling():
    opts = formats.build_ydl_opts({}, "/tmp/x", lambda d: None, max_bytes=5_000_000)
    assert opts["max_filesize"] == 5_000_000


def test_no_budget_means_no_size_filter():
    # Unlimited users pass None; a non-positive remainder is the quota layer's refusal to make at
    # enqueue, not an impossible yt-dlp filter to express here.
    for budget in (None, 0, -1):
        assert "max_filesize" not in formats.build_ydl_opts(
            {}, "/tmp/x", lambda d: None, max_bytes=budget
        )


def test_a_source_adapter_cannot_raise_the_size_ceiling():
    # The adapter merge happens after max_filesize is set, so this pins that no adapter option can
    # widen a storage cap — the mirror of the cookiefile rule, in the opposite direction.
    opts = formats.build_ydl_opts(
        {}, "/tmp/x", lambda d: None, extra={"max_filesize": 999_999_999}, max_bytes=5_000_000
    )
    assert opts["max_filesize"] == 5_000_000


def test_a_join_cut_list_is_capped():
    # An accurate join emits two filters per segment into one ffmpeg filter_complex, so an unbounded
    # list is an unbounded command line. The surplus is dropped; the legitimate head survives.
    segs = [{"start_s": i, "end_s": i + 1} for i in range(500)]
    out = editmod.normalize_edit_spec({"segments": segs})
    assert len(out["segments"]) == editmod._MAX_SEGMENTS
    assert out["segments"][0] == {"start_s": 0.0, "end_s": 1.0}


def test_a_normal_cut_list_is_untouched():
    segs = [{"start_s": 0, "end_s": 5}, {"start_s": 10, "end_s": 12}]
    out = editmod.normalize_edit_spec({"segments": segs})
    assert len(out["segments"]) == 2


def test_audio_versions_its_cache_identity_apart_from_video(monkeypatch):
    # The .mp3/.m4a-stays-itself fix must re-derive cached remuxed audio WITHOUT re-downloading every
    # cached video: bumping one version moves only its own mode's signature.
    audio, video = {"mode": "a", "audio_format": "mp3"}, {"mode": "av"}
    before = formats.format_sig(audio), formats.format_sig(video)
    monkeypatch.setattr(formats, "_AUDIO_SIG_VERSION", formats._AUDIO_SIG_VERSION + 1)
    assert formats.format_sig(audio) != before[0]
    assert formats.format_sig(video) == before[1]
    monkeypatch.setattr(formats, "_SIG_VERSION", formats._SIG_VERSION + 1)
    assert formats.format_sig(video) != before[1]


def test_the_mp3_fix_moved_the_audio_cache_identity(monkeypatch):
    # Before the fix every mode hashed with version 4, so an audio signature still at 4 would find the
    # cached mp3-remuxed-into-mp4 assets iPad refuses (prod job 268) instead of re-deriving them.
    audio = {"mode": "a", "audio_format": "mp3"}
    current = formats.format_sig(audio)
    monkeypatch.setattr(formats, "_AUDIO_SIG_VERSION", 4)
    assert formats.format_sig(audio) != current


class _QuietFiles(http.server.SimpleHTTPRequestHandler):
    def log_message(self, *args):  # the test's output is yt-dlp's, not the server's
        pass

    def handle(self):
        # yt-dlp hangs up on a file over the cap as soon as it reads its size.
        with contextlib.suppress(ConnectionError):
            super().handle()


def test_yt_dlp_refuses_a_file_over_the_cap_without_raising_and_the_watch_sees_it(tmp_path):
    # The boundary `watching_size_cap` reads, held against the yt-dlp this image pins: over
    # `max_filesize` its HTTP downloader logs one line and RETURNS — no exception, no file — so
    # without the watch the pipeline would only find "no output file" and call it a processing fault.
    served = tmp_path / "served"
    served.mkdir()
    (served / "v.mp4").write_bytes(b"\0" * 200_000)
    srv = http.server.ThreadingHTTPServer(
        ("127.0.0.1", 0), functools.partial(_QuietFiles, directory=str(served))
    )
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    url = f"http://127.0.0.1:{srv.server_port}/v.mp4"
    try:
        for cap, refused in ((1_000, True), (10_000_000, False)):
            out = tmp_path / f"out-{cap}"
            opts = formats.build_ydl_opts(
                {}, str(out / "%(id)s.%(ext)s"), lambda d: None, max_bytes=cap
            )
            with formats.watching_size_cap() as over, yt_dlp.YoutubeDL(opts) as ydl:
                ydl.extract_info(url, download=True)
            assert over[0] is refused, cap
            files = [p for p in out.glob("*") if p.suffix == ".mp4"] if out.exists() else []
            assert bool(files) is not refused, cap
    finally:
        srv.shutdown()