siftlode/backend/tests/test_playable.py

"""The phone-safe rule (`downloads.playable`): what a library file must be so a share link plays
on any phone or browser, and the one ffmpeg command that makes it so.

The plan table is the measured device verdict, class by class: what failed on a real iPhone + iPad
(AV1, Matroska, HEVC, MP3 inside an .mp4) must change, and so must what browsers other than Safari
cannot be trusted with (10-bit / 4:4:4 H.264, VP9, Opus). The ffmpeg cases run the real encoder and
probe its output, because the rule is about the bytes, not the flags.
"""

import shutil
import subprocess
from pathlib import Path

import pytest

from app.downloads import playable, probe

# conftest turns a missing ffmpeg into a hard error in the gate (SIFTLODE_TEST_REQUIRE_DB), so this
# only ever skips on an explicit pure-logic run.
needs_ffmpeg = pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")


def _data(video=None, audio=None, fmt="mov,mp4,m4a,3gp,3g2,mj2", cover=False):
    streams = []
    if cover:
        streams.append(
            {"codec_type": "video", "codec_name": "mjpeg", "disposition": {"attached_pic": 1}}
        )
    if video:
        codec, pix = video
        streams.append({"codec_type": "video", "codec_name": codec, "pix_fmt": pix})
    if audio:
        streams.append({"codec_type": "audio", "codec_name": audio})
    return {"streams": streams, "format": {"format_name": fmt}}


# (probe, extension) -> (ext, reencode_video, reencode_audio, needs_change)
CASES = {
    "h264 8-bit + aac in mp4": (_data(("h264", "yuv420p"), "aac"), "mp4", ("mp4", 0, 0, 0)),
    "full-range h264 is still 8-bit 4:2:0": (
        _data(("h264", "yuvj420p"), "aac"),
        "mp4",
        ("mp4", 0, 0, 0),
    ),
    "10-bit h264": (_data(("h264", "yuv420p10le"), "aac"), "mp4", ("mp4", 1, 0, 1)),
    "4:4:4 h264": (_data(("h264", "yuv444p"), "aac"), "mp4", ("mp4", 1, 0, 1)),
    "hevc (an iPhone's .mov)": (_data(("hevc", "yuv420p10le"), "aac"), "mov", ("mp4", 1, 0, 1)),
    "vp9 + opus webm": (_data(("vp9", "yuv420p"), "opus"), "webm", ("mp4", 1, 1, 1)),
    "av1": (_data(("av1", "yuv420p"), "aac"), "mp4", ("mp4", 1, 0, 1)),
    "h264 + aac in mkv: remux only": (_data(("h264", "yuv420p"), "aac"), "mkv", ("mp4", 0, 0, 1)),
    "h264 + opus in mp4": (_data(("h264", "yuv420p"), "opus"), "mp4", ("mp4", 0, 1, 1)),
    "mp3 beside video in mp4": (_data(("h264", "yuv420p"), "mp3"), "mp4", ("mp4", 0, 1, 1)),
    "a silent h264 video": (_data(("h264", "yuv420p")), "mp4", ("mp4", 0, 0, 0)),
    "mp3": (_data(audio="mp3", fmt="mp3"), "mp3", ("mp3", 0, 0, 0)),
    "mp3 with cover art": (_data(audio="mp3", fmt="mp3", cover=True), "mp3", ("mp3", 0, 0, 0)),
    "mp3 inside an mp4: remux to .mp3": (_data(audio="mp3"), "mp4", ("mp3", 0, 0, 1)),
    "aac in m4a": (_data(audio="aac"), "m4a", ("m4a", 0, 0, 0)),
    "aac audio-only in mp4: remux to .m4a": (_data(audio="aac"), "mp4", ("m4a", 0, 0, 1)),
    "opus in ogg": (_data(audio="opus", fmt="ogg"), "ogg", ("m4a", 0, 1, 1)),
}


@pytest.mark.parametrize("name", CASES)
def test_the_plan_for_each_class(name):
    data, ext, (want_ext, rv, ra, needs) = CASES[name]
    p = playable.plan(data, ext)
    assert p is not None
    assert (p.ext, p.reencode_video, p.reencode_audio, p.needs_change) == (
        want_ext,
        bool(rv),
        bool(ra),
        bool(needs),
    )


@pytest.mark.parametrize(
    "data",
    [
        None,  # ffprobe could not read it
        {"streams": [], "format": {"format_name": "pdf"}},
        _data(("png", "rgb24"), fmt="png_pipe"),  # a still image is a download-only file
    ],
)
def test_a_file_that_is_not_media_has_no_plan(data):
    assert playable.plan(data, "bin") is None


def test_the_extension_is_read_with_or_without_its_dot():
    data = _data(("h264", "yuv420p"), "aac")
    assert playable.plan(data, ".MP4") == playable.plan(data, "mp4")
    assert not playable.plan(data, ".MP4").needs_change


def test_a_video_re_encode_pins_8_bit_420_and_maps_only_the_judged_streams():
    p = playable.Plan("mp4", True, True, False, True)
    cmd = playable.build_cmd(Path("/in.mov"), Path("/out.mp4"), p)
    assert cmd[cmd.index("-pix_fmt") + 1] == "yuv420p"
    assert cmd[cmd.index("-c:v") + 1] == "libx264"
    assert cmd[cmd.index("-c:a") + 1] == "copy"
    maps = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-map"]
    assert maps == ["0:V:0", "0:a:0?"]  # never cover art, never a subtitle the muxer refuses
    assert cmd[-3:] == ["-movflags", "+faststart", "/out.mp4"]


def test_an_audio_remux_copies_the_audio_alone():
    p = playable.Plan("mp3", False, False, False, True)
    cmd = playable.build_cmd(Path("/in.mp4"), Path("/out.mp3"), p)
    assert [cmd[i + 1] for i, a in enumerate(cmd) if a == "-map"] == ["0:a:0"]
    assert "-c:v" not in cmd and "-pix_fmt" not in cmd
    assert cmd[cmd.index("-c:a") + 1] == "copy"
    assert "-movflags" not in cmd  # an .mp3 has no moov atom to move


# --- the real encoder ------------------------------------------------------------------------


def _ffmpeg(*args: str) -> None:
    subprocess.run(["ffmpeg", "-y", "-loglevel", "error", *args], check=True)


def _convert(src: Path, tmp_path: Path) -> Path:
    p = playable.plan(probe.probe_json(src), src.suffix)
    assert p is not None and p.needs_change
    dest = tmp_path / f"out.{p.ext}"
    subprocess.run(playable.build_cmd(src, dest, p), check=True, capture_output=True)
    return dest


@needs_ffmpeg
def test_a_10_bit_matroska_with_opus_comes_out_phone_safe(tmp_path):
    src = tmp_path / "in.mkv"
    _ffmpeg(
        "-f", "lavfi", "-i", "testsrc2=size=160x120:rate=10:duration=1",
        "-f", "lavfi", "-i", "sine=duration=1",
        "-c:v", "libx264", "-pix_fmt", "yuv420p10le", "-c:a", "libopus",
        str(src),
    )  # fmt: skip
    out = _convert(src, tmp_path)
    data = probe.probe_json(out)
    video, audio = probe.primary_streams(data)
    assert (video["codec_name"], video["pix_fmt"], video["profile"]) == ("h264", "yuv420p", "High")
    assert audio["codec_name"] == "aac"
    assert out.suffix == ".mp4"
    assert not playable.plan(data, out.suffix).needs_change  # the rule accepts its own output


@needs_ffmpeg
def test_mp3_inside_an_mp4_comes_out_as_an_mp3(tmp_path):
    # The one unsafe file prod held: an audio download remuxed into .mp4, which iPhone and iPad
    # both refused to play.
    src = tmp_path / "in.mp4"
    _ffmpeg("-f", "lavfi", "-i", "sine=duration=1", "-c:a", "libmp3lame", "-f", "mp4", str(src))
    out = _convert(src, tmp_path)
    data = probe.probe_json(out)
    assert out.suffix == ".mp3"
    assert probe.primary_streams(data)[1]["codec_name"] == "mp3"
    assert not playable.plan(data, out.suffix).needs_change