peter/siftlode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
d466ae915bfc636b699cf93f457685aa4d353711
siftlode/backend/tests/test_clip.py
T
peter d466ae915b fix(downloads): a YouTube clip outside the catalog takes its title from the fetch 
It was queued as "Clip (clip)" and stayed so: nothing renamed it once yt-dlp had answered.
2026-10-10 02:43:13 +02:00

443 lines
16 KiB
Python
Raw Blame History

"""Ad-hoc clips (downloads/clip): the spec, the one ffmpeg command every clip goes through, the route,
and the worker path for both sources.
The ffmpeg command is judged at argv level: where the seek sits, which streams are mapped, what the
video filter does to an HDR source. The worker runs with ffmpeg / yt-dlp replaced by stand-ins that
write a file where they were told to, so what is checked is what the pipeline asks of them and what
it stores afterwards.
"""
from pathlib import Path
import pytest
from app.config import settings
from app.downloads import clip as clipmod
from app.downloads import pipeline, quota, runner, service
from app.downloads import probe as probemod
from app.downloads.edit import INPUT_GUARD
from app.models import Channel, DownloadJob, MediaAsset, PlexItem, PlexLibrary, PlexShow, Video
from app.plex import paths as plex_paths
# --- the spec -----------------------------------------------------------------------------------
def test_a_spec_is_canonical_so_the_same_marks_hit_the_same_clip():
a = clipmod.normalize("youtube", "abcdefghijk", 10.00049, 40.0)
b = clipmod.normalize("youtube", "abcdefghijk", 10.0, 40.0004)
assert a == b == {"source": "youtube", "ref": "abcdefghijk", "start_s": 10.0, "end_s": 40.0}
assert clipmod.sig(a) == clipmod.sig(b)
assert clipmod.sig(a) != clipmod.sig({**a, "end_s": 41.0})
@pytest.mark.parametrize(
("args", "reason"),
[
(("vimeo", "abcdefghijk", 0, 10), "bad_source"),
(("youtube", "short", 0, 10), "bad_source"),
(("youtube", "abcdefghij/", 0, 10), "bad_source"),
(("plex", "12a", 0, 10), "bad_source"),
(("youtube", "abcdefghijk", -1, 10), "bad_range"),
(("youtube", "abcdefghijk", 0, float("nan")), "bad_range"),
(("youtube", "abcdefghijk", 5, 5.5), "too_short"),
(("youtube", "abcdefghijk", 30, 10), "too_short"),
(("youtube", "abcdefghijk", 0, 600.001), "too_long"),
],
)
def test_a_range_it_cannot_cut_is_refused_with_its_reason(args, reason):
with pytest.raises(clipmod.ClipError) as e:
clipmod.normalize(*args)
assert e.value.reason == reason
def test_the_ten_minute_cap_itself_is_allowed():
assert clipmod.duration(clipmod.normalize("plex", "42", 0, 600)) == 600
def test_an_audio_track_is_kept_for_plex_only_and_must_be_sane():
assert clipmod.normalize("plex", "42", 0, 10, 2)["audio_track"] == 2
assert "audio_track" not in clipmod.normalize("youtube", "abcdefghijk", 0, 10, 2)
with pytest.raises(clipmod.ClipError, match="bad_audio"):
clipmod.normalize("plex", "42", 0, 10, -1)
def test_a_youtube_clip_links_its_video_at_the_clip_start_and_a_plex_clip_links_nothing():
yt = clipmod.normalize("youtube", "abcdefghijk", 75.9, 90)
assert clipmod.reference_url(yt) == "https://www.youtube.com/watch?v=abcdefghijk&t=75s"
assert clipmod.reference_url(clipmod.normalize("plex", "42", 0, 10)) is None
# --- the ffmpeg command ----------------------------------------------------------------------------
def _cmd(**kw):
args = {"audio_track": None, "hdr": False, "threads": 3, **kw}
return clipmod.build_cmd(Path("/m/in.mkv"), Path("/s/out.mp4"), 61.5, 30.0, **args)
def test_the_seek_is_on_the_input_so_the_clip_starts_on_the_marked_frame():
cmd = _cmd()
i = cmd.index("-i")
assert cmd[cmd.index("-ss") + 1] == "61.500" and cmd.index("-ss") < i
assert cmd[i + 1] == "/m/in.mkv"
assert cmd[cmd.index("-t") + 1] == "30.000" and cmd.index("-t") > i
# The single-file demuxer allowlist goes with every library read.
g = cmd.index(INPUT_GUARD[0])
assert cmd[g : g + len(INPUT_GUARD)] == INPUT_GUARD and g < i
assert cmd[-1] == "/s/out.mp4"
def test_it_maps_the_first_video_and_the_heard_audio_track_and_nothing_else():
cmd = _cmd(audio_track=2)
maps = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-map"]
assert maps == ["0:v:0", "0:a:2"]
assert "-sn" in cmd and "-dn" in cmd
# An mkv's chapters would ride into the .mp4 as a data track of their own (measured on a Blu-ray
# rip: `bin_data`), which -dn does not drop.
assert cmd[cmd.index("-map_chapters") + 1] == "-1"
# No track chosen: the first one, and a source with no audio at all still cuts.
maps = [c for i, c in enumerate(_cmd()) if i and _cmd()[i - 1] == "-map"]
assert maps == ["0:v:0", "0:a:0?"]
def test_the_output_is_phone_safe_stereo_h264_at_most_1080p():
cmd = _cmd()
assert cmd[cmd.index("-c:v") + 1] == "libx264"
assert cmd[cmd.index("-pix_fmt") + 1] == "yuv420p"
assert cmd[cmd.index("-c:a") + 1] == "aac" and cmd[cmd.index("-ac") + 1] == "2"
assert cmd[cmd.index("-threads") + 1] == "3"
vf = cmd[cmd.index("-vf") + 1]
assert "min(1080,ih)" in vf and "tonemap" not in vf
def test_an_hdr_source_is_tonemapped_to_sdr_before_it_is_scaled():
vf = _cmd(hdr=True)[_cmd(hdr=True).index("-vf") + 1]
assert vf.index("tonemap") < vf.index("zscale=t=bt709") < vf.index("min(1080,ih)")
@pytest.mark.parametrize(
("transfer", "hdr"),
[("smpte2084", True), ("arib-std-b67", True), ("bt709", False), (None, False)],
)
def test_hdr_is_read_from_the_first_video_stream(transfer, hdr):
data = {
"streams": [{"codec_type": "audio"}, {"codec_type": "video", "color_transfer": transfer}]
}
assert clipmod.is_hdr(data) is hdr
# --- the route ---------------------------------------------------------------------------------------
@pytest.fixture
def yt_video(db):
db.add(Channel(id="UC_clip", title="The Channel"))
db.flush()
db.add(
Video(
id="abcdefghijk",
channel_id="UC_clip",
title="A Funny Video",
duration_seconds=300,
thumbnail_url="https://i.ytimg.com/vi/abcdefghijk/hq.jpg",
)
)
db.commit()
return "abcdefghijk"
@pytest.fixture
def plex_episode(db, tmp_path, monkeypatch):
lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
db.add(lib)
db.flush()
show = PlexShow(rating_key="s9", library_id=lib.id, title="The Show")
db.add(show)
db.flush()
db.add(
PlexItem(
rating_key="777",
library_id=lib.id,
kind="episode",
title="Pilot",
show_id=show.id,
season_number=1,
episode_number=2,
duration_s=1800,
file_path="/data/shows/pilot.mkv",
)
)
db.commit()
media = tmp_path / "pilot.mkv"
media.write_bytes(b"m" * 8)
monkeypatch.setattr(
plex_paths,
"local_media_path",
lambda _db, p: media if p == "/data/shows/pilot.mkv" else None,
)
return media
def _clip(client, **body):
return client.post("/api/downloads/clip", json={"start_s": 10, "end_s": 40, **body})
def test_a_youtube_clip_is_queued_as_its_own_job_titled_after_its_video(
db, make_user, as_user, yt_video
):
r = _clip(as_user(make_user()), source="youtube", ref=yt_video)
assert r.status_code == 200, r.text
job = r.json()
assert (job["job_kind"], job["status"], job["display_name"]) == (
"clip",
"queued",
"A Funny Video (clip)",
)
assert job["edit_spec"] == {"source": "youtube", "ref": yt_video, "start_s": 10, "end_s": 40}
assert job["source_url"] == f"https://www.youtube.com/watch?v={yt_video}&t=10s"
asset = db.get(MediaAsset, job["asset"]["id"])
assert (asset.duration_s, asset.uploader, asset.media_kind) == (30, "The Channel", "video")
def test_the_same_range_again_is_the_same_clip_but_never_another_users(
db, make_user, as_user, yt_video
):
a, b = make_user(), make_user()
first = _clip(as_user(a), source="youtube", ref=yt_video).json()["asset"]["id"]
again = _clip(as_user(a), source="youtube", ref=yt_video).json()["asset"]["id"]
other = _clip(as_user(b), source="youtube", ref=yt_video).json()["asset"]["id"]
longer = _clip(as_user(a), source="youtube", ref=yt_video, end_s=41).json()["asset"]["id"]
assert first == again
assert len({first, other, longer}) == 3
def test_a_ready_clip_asked_for_again_is_done_at_once(db, make_user, as_user, yt_video):
u = make_user()
first = _clip(as_user(u), source="youtube", ref=yt_video).json()
asset = db.get(MediaAsset, first["asset"]["id"])
asset.status, asset.rel_path = "ready", ".edits/1/clip_1.mp4"
db.commit()
again = _clip(as_user(u), source="youtube", ref=yt_video).json()
assert (again["status"], again["progress"]) == ("done", 100)
@pytest.mark.parametrize(
("body", "status", "reason"),
[
({"end_s": 700}, 422, "too_long"),
({"start_s": 299.5, "end_s": 330}, 422, "past_end"),
({"ref": "nope"}, 422, "bad_source"),
],
)
def test_a_youtube_clip_it_cannot_cut_says_why(
db, make_user, as_user, yt_video, body, status, reason
):
r = _clip(as_user(make_user()), **{"source": "youtube", "ref": yt_video, **body})
assert r.status_code == status
assert r.json()["detail"] == {"code": "clip", "reason": reason}
assert db.query(DownloadJob).count() == 0
def test_a_plex_clip_is_titled_show_episode_title_and_keeps_its_audio_track(
db, make_user, as_user, plex_episode
):
r = _clip(as_user(make_user()), source="plex", ref="777", audio_track=1)
assert r.status_code == 200, r.text
job = r.json()
assert job["display_name"] == "The Show — S1·E2 — Pilot (clip)"
assert job["edit_spec"]["audio_track"] == 1
assert job["source_url"] is None
def test_a_plex_clip_of_an_unknown_item_is_a_404_and_of_an_unreachable_file_a_422(
db, make_user, as_user, plex_episode
):
c = as_user(make_user())
assert _clip(c, source="plex", ref="999").json()["detail"]["reason"] == "not_found"
assert _clip(c, source="plex", ref="999").status_code == 404
item = db.query(PlexItem).one()
item.file_path = "/data/elsewhere.mkv"
db.commit()
r = _clip(c, source="plex", ref="777")
assert (r.status_code, r.json()["detail"]["reason"]) == (422, "media_missing")
def test_a_clip_over_quota_is_refused_like_a_download(
db, make_user, as_user, yt_video, monkeypatch
):
def full(_db, _uid):
raise quota.QuotaExceeded("max_jobs", limit=5, current=5)
monkeypatch.setattr(quota, "check_enqueue", full)
r = _clip(as_user(make_user()), source="youtube", ref=yt_video)
assert r.status_code == 422
assert db.query(DownloadJob).count() == 0
def test_a_clip_is_not_redownloadable(db, make_user, as_user, yt_video):
c = as_user(make_user())
job = _clip(c, source="youtube", ref=yt_video).json()
row = db.get(DownloadJob, job["id"])
row.status = "done"
db.commit()
r = c.post(f"/api/downloads/{job['id']}/redownload")
assert (r.status_code, r.json()["detail"]["reason"]) == (409, "not_redownloadable")
# --- the worker ----------------------------------------------------------------------------------------
def test_a_clip_job_reaches_the_clip_entry(monkeypatch):
called = []
for name in ("_convert_upload", "_process_edit", "_download", "_process_clip"):
monkeypatch.setattr(pipeline, name, lambda *a, _n=name: called.append(_n))
runner._run(5, 7, "clip", "clip", "1:youtube:abcdefghijk", {}, 1)
assert called == ["_process_clip"]
@pytest.fixture
def worker(db, tmp_path, monkeypatch):
"""The worker path with its external programs replaced: ffmpeg writes the file it was given, the
probe answers what the test says, the phone-safe step and the poster pass through."""
monkeypatch.setattr(settings, "download_root", str(tmp_path / "root"))
seen: dict = {}
def ffmpeg(cmd, *_a, **_k):
seen["cmd"] = cmd
Path(cmd[-1]).write_bytes(b"clip")
monkeypatch.setattr(pipeline.editmod, "run_ffmpeg", ffmpeg)
monkeypatch.setattr(pipeline, "_ensure_phone_safe", lambda _j, p, *_a, **_k: p)
monkeypatch.setattr(pipeline.editmod, "ensure_poster", lambda *_a, **_k: None)
monkeypatch.setattr(probemod, "probe_media", lambda _p: ("h264", "aac", 1920, 1080, 30))
seen["probe"] = {"streams": [{"codec_type": "video"}, {"codec_type": "audio"}]}
monkeypatch.setattr(probemod, "probe_json", lambda _p: seen["probe"])
def run(job_json):
job = db.get(DownloadJob, job_json["id"])
asset = db.get(MediaAsset, job.asset_id)
job.status, asset.status = "running", "downloading"
db.commit()
pipeline._process_clip(job.id, asset.id)
db.expire_all()
return db.get(DownloadJob, job.id), db.get(MediaAsset, asset.id)
seen["run"] = run
return seen
def test_a_plex_clip_is_cut_from_the_library_file_with_its_audio_track(
db, make_user, as_user, plex_episode, worker
):
worker["probe"] = {
"streams": [
{"codec_type": "video", "color_transfer": "smpte2084"},
{"codec_type": "audio"},
{"codec_type": "audio"},
]
}
job = _clip(as_user(make_user()), source="plex", ref="777", audio_track=1).json()
job, asset = worker["run"](job)
assert (job.status, asset.status, asset.container, asset.height) == (
"done",
"ready",
"mp4",
1080,
)
cmd = worker["cmd"]
assert cmd[cmd.index("-i") + 1] == str(plex_episode)
assert cmd[cmd.index("-ss") + 1] == "10.000"
assert "0:a:1" in cmd and "tonemap" in cmd[cmd.index("-vf") + 1]
assert asset.rel_path.startswith(".edits/") and asset.rel_path.endswith(".mp4")
def test_a_plex_clip_of_a_track_the_file_does_not_have_fails_and_says_so(
db, make_user, as_user, plex_episode, worker
):
job = _clip(as_user(make_user()), source="plex", ref="777", audio_track=3).json()
job, asset = worker["run"](job)
assert (job.status, asset.status) == ("error", "error")
assert "audio track" in job.error
assert "cmd" not in worker
def test_a_youtube_clip_fetches_only_its_range_and_cuts_it_from_zero(
db, make_user, as_user, yt_video, worker, monkeypatch
):
asked: dict = {}
def extract(job_id, asset_id, url, spec, staging, adapter, **kw):
asked.update(url=url, spec=spec, **kw)
f = Path(staging) / "abcdefghijk.mp4"
f.write_bytes(b"range")
return {"requested_downloads": [{"filepath": str(f)}]}
monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)
job = _clip(as_user(make_user()), source="youtube", ref=yt_video).json()
job, asset = worker["run"](job)
assert (job.status, asset.status) == ("done", "ready")
assert asked["section"] == (10, 40)
assert asked["spec"]["max_height"] == 1080
assert asked["url"].endswith(yt_video)
cmd = worker["cmd"]
assert cmd[cmd.index("-ss") + 1] == "0.000" and cmd[cmd.index("-t") + 1] == "30.000"
assert cmd[cmd.index("-i") + 1].endswith("abcdefghijk.mp4")
def test_a_youtube_clip_hands_the_range_to_yt_dlp(monkeypatch, tmp_path):
seen = {}
class FakeYDL:
def __init__(self, opts):
seen.update(opts)
def __enter__(self):
return self
def __exit__(self, *_a):
return False
def extract_info(self, *_a, **_k):
return {}
monkeypatch.setattr(pipeline.yt_dlp, "YoutubeDL", FakeYDL)
monkeypatch.setattr(pipeline, "_touch_job", lambda *_a: None)
adapter = pipeline.adapters.resolve_adapter("youtube", "abcdefghijk")
pipeline._extract_with_fallback(
1,
2,
"https://www.youtube.com/watch?v=abcdefghijk",
pipeline.formats.normalize({}),
tmp_path,
adapter,
section=(10.0, 40.0),
)
ranges = list(seen["download_ranges"]({}, None))
assert [(r["start_time"], r["end_time"]) for r in ranges] == [(10.0, 40.0)]
assert seen["force_keyframes_at_cuts"] is False
def test_service_reference_url_follows_the_clip_spec(db, make_user, as_user, yt_video):
job = _clip(as_user(make_user()), source="youtube", ref=yt_video, start_s=65, end_s=90).json()
row = db.get(DownloadJob, job["id"])
assert service.reference_url(row, None) == f"https://www.youtube.com/watch?v={yt_video}&t=65s"
def test_a_youtube_clip_outside_the_catalog_takes_its_title_from_the_fetch(
db, make_user, as_user, worker, monkeypatch
):
def extract(job_id, asset_id, url, spec, staging, adapter, **kw):
f = Path(staging) / "zyxwvutsrqp.mp4"
f.write_bytes(b"range")
return {"requested_downloads": [{"filepath": str(f)}], "title": "Found On YouTube"}
monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)
job = _clip(as_user(make_user()), source="youtube", ref="zyxwvutsrqp").json()
assert job["display_name"] == "Clip (clip)"
job, asset = worker["run"](job)
assert (job.status, asset.title, job.display_name) == (
"done",
"Found On YouTube (clip)",
"Found On YouTube (clip)",
)
Reference in New Issue View Git Blame Copy Permalink