kaledonya-clips/intake/tests/test_previews.py

import httpx
import pytest
from fastapi.testclient import TestClient

from app.previews import (
    LONG_SECONDS,
    Previews,
    encode_cmd,
    follow_progress,
    niced,
    target_bytes,
    video_kbps,
)
from app.server import Settings, create_app
from tests.test_server import FakeZipline as UploadZipline
from tests.test_site_api import FakeZipline, login, mkfile


def test_bitrate_fits_the_target():
    for seconds in (30, 70, 300):
        kbps = video_kbps(seconds)
        assert (kbps + 128) * 1000 / 8 * seconds <= target_bytes(seconds)
    assert video_kbps(10) == 4000  # short clips are capped, not blown up
    assert video_kbps(3600) is None  # an hour can't fit in 22 MB at any watchable quality
    assert video_kbps(0) is None


def test_long_clips_get_a_smaller_file():
    assert target_bytes(LONG_SECONDS) == 22_000_000
    assert target_bytes(LONG_SECONDS + 1) == 18_000_000


def test_encode_keeps_one_audio_track_and_720p_up_to_four_minutes():
    cmd = encode_cmd("http://zipline:3000/a.mp4", "out.mp4", 600, LONG_SECONDS)
    assert cmd[cmd.index("-map") + 1:cmd.index("-map") + 4] == ["0:v:0", "-map", "0:a:0?"]
    assert "scale=-2:'min(720,ih)'" in cmd and "+faststart" in cmd
    assert "scale=-2:'min(480,ih)'" in encode_cmd("x", "y", 600, LONG_SECONDS + 1)


def test_encode_runs_at_low_priority_when_nice_exists(monkeypatch):
    monkeypatch.setattr("app.previews.shutil.which", lambda name: "/usr/bin/nice")
    assert niced(["ffmpeg", "-i", "x"]) == ["nice", "-n", "10", "ffmpeg", "-i", "x"]
    monkeypatch.setattr("app.previews.shutil.which", lambda name: None)
    assert niced(["ffmpeg"]) == ["ffmpeg"]


def test_small_clips_get_no_copy(tmp_path, monkeypatch):
    p = Previews(tmp_path, "http://zipline:3000", over_bytes=1000)
    monkeypatch.setattr(p, "make", lambda *a: pytest.fail("should not encode"))
    p.request("small.mp4", 999)
    assert p.jobs.empty() and not p.pending


def test_failed_clip_is_not_retried(tmp_path):
    p = Previews(tmp_path, "http://zipline:3000", over_bytes=10)
    (tmp_path / f"{Previews.key('big.mp4', 50)}.failed").touch()
    p.request("big.mp4", 50)
    assert p.jobs.empty()


def test_key_changes_with_size_and_rejects_paths(tmp_path):
    assert Previews.key("a.mp4", 1) != Previews.key("a.mp4", 2)
    p = Previews(tmp_path, "http://zipline:3000", over_bytes=10)
    assert p.file("../secret") is None


@pytest.fixture
def setup(tmp_path, monkeypatch):
    zl = FakeZipline()
    zl.files["tok-alice"].append(mkfile("e9", "big.mp4", size=5000))
    asked = []
    monkeypatch.setattr(Previews, "request", lambda self, name, length: asked.append((name, length)))
    monkeypatch.setattr("app.site_api.video_size", lambda url, length: (1280, 720))
    settings = Settings(zipline_url="http://zipline:3000", site_origin="https://clips.example.com",
                        upload_url="https://upload.example.com", max_bytes=10_000, work_dir=tmp_path, preview_over=1000)
    client = TestClient(create_app(settings, transport=httpx.MockTransport(zl.handler)), base_url="https://testserver")
    return client, asked, tmp_path / "previews"


def test_big_clip_without_copy_links_the_original_and_asks_for_one(setup):
    client, asked, _ = setup
    r = client.get("/v/big.mp4")
    assert '<meta property="og:video" content="https://clips.example.com/big.mp4">' in r.text
    assert asked == [("big.mp4", 5000)]


def test_big_clip_with_copy_links_the_copy(setup):
    client, asked, folder = setup
    key = Previews.key("big.mp4", 5000)
    folder.mkdir()
    (folder / f"{key}.mp4").write_bytes(b"0123456789")
    r = client.get("/v/big.mp4")
    assert f'<meta property="og:video" content="https://clips.example.com/_site/preview/{key}.mp4">' in r.text
    assert '<meta property="og:video:width" content="1280">' in r.text
    assert asked == []
    got = client.get(f"/_site/preview/{key}.mp4", headers={"range": "bytes=2-5"})
    assert got.status_code == 206 and got.content == b"2345"
    assert got.headers["content-type"] == "video/mp4"
    assert client.get("/_site/preview/" + "0" * 24 + ".mp4").status_code == 404


def test_upload_of_a_video_asks_for_a_copy(tmp_path, monkeypatch):
    asked = []
    monkeypatch.setattr(Previews, "request", lambda self, name, length: asked.append((name, length)))
    settings = Settings(zipline_url="http://zipline:3000", site_origin="https://clips.example.com",
                        upload_url="https://upload.example.com", max_bytes=10_000, work_dir=tmp_path, preview_over=1000)
    client = TestClient(create_app(settings, transport=httpx.MockTransport(UploadZipline().handler)))
    for name, kind in (("clip.mp4", "video/mp4"), ("notes.zip", "application/zip")):
        r = client.post("/upload", headers={"authorization": "tok"}, files={"file": (name, b"x" * 1500, kind)})
        assert r.status_code == 200
    assert asked == [("clip.mp4", 1500)]


def test_rename_carries_the_copy_over(setup):
    client, _, folder = setup
    folder.mkdir()
    (folder / f"{Previews.key('big.mp4', 5000)}.mp4").write_bytes(b"copy")
    login(client)
    assert client.patch("/_site/api/files/e9", json={"name": "renamed.mp4"}).json() == {"name": "renamed.mp4"}
    assert (folder / f"{Previews.key('renamed.mp4', 5000)}.mp4").read_bytes() == b"copy"
    assert not (folder / f"{Previews.key('big.mp4', 5000)}.mp4").exists()
    r = client.get("/v/renamed.mp4")
    assert f"/_site/preview/{Previews.key('renamed.mp4', 5000)}.mp4" in r.text


def test_discord_status_tells_the_site_when_the_link_is_safe(setup):
    client, asked, folder = setup
    get = lambda name: client.get(f"/_site/api/discord/{name}").json()
    assert get("clutch.mp4") == {"state": "small", "progress": 1.0, "queued": False}  # plays as is
    assert get("big.mp4") == {"state": "making", "progress": 0.0, "queued": True}
    assert asked == [("big.mp4", 5000)]
    folder.mkdir()
    key = Previews.key("big.mp4", 5000)
    (folder / f"{key}.failed").touch()
    assert get("big.mp4")["state"] == "none"
    (folder / f"{key}.mp4").write_bytes(b"copy")
    assert get("big.mp4") == {"state": "ready", "progress": 1.0, "queued": False}
    assert get("nope.mp4")["state"] == "small"


def test_status_shows_the_encode_progress_and_the_queue(tmp_path):
    p = Previews(tmp_path, "http://zipline:3000", over_bytes=10)
    p.worker = type("Busy", (), {"is_alive": lambda self: True})()  # a running worker: request() only queues
    p.request("a.mp4", 50)
    p.request("b.mp4", 50)
    p.current, p.progress = Previews.key("a.mp4", 50), 0.42
    assert p.status("a.mp4", 50) == {"state": "making", "progress": 0.42, "queued": False}
    assert p.status("b.mp4", 50) == {"state": "making", "progress": 0.0, "queued": True}


def test_follow_progress_reads_ffmpeg_progress_lines():
    seen = []
    lines = ["frame=10", "out_time_us=5000000", "progress=continue", "out_time_us=N/A",
             "out_time_us=9000000", "out_time_us=12000000", "progress=end"]
    follow_progress(lines, 10.0, seen.append)
    assert seen == [0.5, 0.9, 1.0]


def test_encode_reports_progress():
    cmd = encode_cmd("x", "y", 600, 30)
    assert cmd[cmd.index("-progress") + 1] == "pipe:1" and "-nostats" in cmd