peter
520b0afb90
An optional crop in clip mode: a Crop toggle puts a frame on the picture (free, 16:9, 9:16 or 1:1; the whole picture stays the default). The frame is sent in fractions of the box it was drawn in, with that box's shape; the worker fits the real picture into the box by its display aspect (sample aspect and rotation aware), drops the part over the bars, and crops first in the filter chain. A frame covering the whole box is no crop at all, and one off the picture is refused by name. The crop frame is a ui primitive (CropFrame) shared with the video editor, keyboard-operable. YouTube: the wide side panel shuts while clipping and comes back after; maximised, the clip panel docks in the stage. Plex: the picture sits above the clip panel so the frame lies on it.
653 lines
25 KiB
Python
653 lines
25 KiB
Python
"""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 crop ----------------------------------------------------------------------------------------
|
||
|
||
_FRAME = {"x": 0.5, "y": 0.25, "w": 0.25, "h": 0.5, "aspect": 16 / 9}
|
||
|
||
|
||
def test_a_crop_frame_is_canonical_and_a_frame_over_the_whole_box_is_no_crop():
|
||
a = clipmod.normalize("plex", "42", 0, 10, crop={**_FRAME, "x": 0.50004})
|
||
b = clipmod.normalize("plex", "42", 0, 10, crop=_FRAME)
|
||
assert a["crop"] == b["crop"] == {"x": 0.5, "y": 0.25, "w": 0.25, "h": 0.5, "aspect": 1.7778}
|
||
assert clipmod.sig(a) == clipmod.sig(b)
|
||
whole = clipmod.normalize(
|
||
"plex", "42", 0, 10, crop={"x": 0, "y": 0, "w": 1, "h": 1, "aspect": 1}
|
||
)
|
||
assert whole == clipmod.normalize("plex", "42", 0, 10)
|
||
# A far edge a hair past the box (the client's rounding) is the box's edge.
|
||
assert clipmod.normalize("plex", "42", 0, 10, crop={**_FRAME, "w": 0.5004})["crop"]["w"] == 0.5
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"frame",
|
||
[
|
||
{**_FRAME, "w": 0.05},
|
||
{**_FRAME, "h": 0.09},
|
||
{**_FRAME, "x": -0.1},
|
||
{**_FRAME, "x": 0.8},
|
||
{**_FRAME, "y": 0.6},
|
||
{**_FRAME, "aspect": 0},
|
||
{**_FRAME, "aspect": 9},
|
||
{**_FRAME, "x": float("nan")},
|
||
{"x": 0.1, "y": 0.1, "w": 0.5},
|
||
{**_FRAME, "h": "half"},
|
||
],
|
||
)
|
||
def test_a_crop_frame_it_cannot_cut_is_refused(frame):
|
||
with pytest.raises(clipmod.ClipError) as e:
|
||
clipmod.normalize("plex", "42", 0, 10, crop=frame)
|
||
assert e.value.reason == "bad_crop"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("stream", "dar"),
|
||
[
|
||
({"width": 1920, "height": 1080}, 16 / 9),
|
||
# An anamorphic DVD: 720×480 stored, shown 16:9.
|
||
({"width": 720, "height": 480, "sample_aspect_ratio": "32:27"}, 16 / 9),
|
||
({"width": 640, "height": 480, "sample_aspect_ratio": "0:1"}, 4 / 3),
|
||
# A phone video stored landscape, shown upright.
|
||
({"width": 1280, "height": 720, "side_data_list": [{"rotation": -90}]}, 9 / 16),
|
||
({"width": 1280, "height": 720, "tags": {"rotate": "270"}}, 9 / 16),
|
||
({"width": 1280, "height": 720, "side_data_list": [{"rotation": 180}]}, 16 / 9),
|
||
({"width": None, "height": 720}, None),
|
||
],
|
||
)
|
||
def test_the_display_aspect_is_the_picture_as_a_player_shows_it(stream, dar):
|
||
got = clipmod.display_aspect(
|
||
{"streams": [{"codec_type": "audio"}, {"codec_type": "video", **stream}]}
|
||
)
|
||
assert got == pytest.approx(dar) if dar else got is None
|
||
|
||
|
||
def _crop_args(f: str) -> list[float]:
|
||
# crop=trunc(iw*W/2)*2:trunc(ih*H/2)*2:trunc(iw*X/2)*2:trunc(ih*Y/2)*2 → [W, H, X, Y]
|
||
assert f.startswith("crop=")
|
||
return [float(p.split("*")[1].split("/")[0]) for p in f[5:].split(":")]
|
||
|
||
|
||
def test_a_frame_on_a_picture_that_fills_its_box_is_cut_as_drawn():
|
||
assert _crop_args(clipmod.crop_filter(_FRAME, 16 / 9)) == pytest.approx([0.25, 0.5, 0.5, 0.25])
|
||
|
||
|
||
def test_a_frame_drawn_in_a_16_9_box_over_a_4_3_picture_maps_onto_the_picture():
|
||
# The 4:3 picture fills 3/4 of the box's width from 1/8 in: a frame over its right half's middle.
|
||
frame = {"x": 0.5, "y": 0.25, "w": 0.1875, "h": 0.5, "aspect": 16 / 9}
|
||
assert _crop_args(clipmod.crop_filter(frame, 4 / 3)) == pytest.approx([0.25, 0.5, 0.5, 0.25])
|
||
|
||
|
||
def test_the_part_of_a_frame_over_a_bar_is_dropped_and_a_frame_on_a_bar_alone_is_no_cut():
|
||
over = {"x": 0.0, "y": 0.0, "w": 0.25, "h": 0.5, "aspect": 16 / 9}
|
||
w, h, x, y = _crop_args(clipmod.crop_filter(over, 4 / 3))
|
||
assert (w, h, x, y) == pytest.approx([1 / 6, 0.5, 0, 0], abs=1e-6)
|
||
assert clipmod.crop_filter({**over, "w": 0.13}, 4 / 3) is None
|
||
# A letterboxed film: the frame over the whole box is the whole picture.
|
||
whole = {"x": 0.0, "y": 0.0, "w": 1.0, "h": 1.0, "aspect": 16 / 9}
|
||
assert clipmod.crop_filter(whole, 2.39) == ""
|
||
|
||
|
||
def test_the_crop_comes_first_so_the_tonemap_and_the_scale_work_on_what_it_keeps():
|
||
f = clipmod.crop_filter(_FRAME, 16 / 9)
|
||
vf = _cmd(hdr=True, crop=f)[_cmd(hdr=True, crop=f).index("-vf") + 1]
|
||
assert vf.startswith(f + ",") and vf.index("tonemap") < vf.index("min(1080,ih)")
|
||
assert "crop" not in _cmd()[_cmd().index("-vf") + 1]
|
||
|
||
|
||
# --- 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_cropped_plex_clip_cuts_the_frame_from_the_picture_and_is_its_own_clip(
|
||
db, make_user, as_user, plex_episode, worker
|
||
):
|
||
worker["probe"] = {
|
||
"streams": [
|
||
{"codec_type": "video", "width": 720, "height": 480, "sample_aspect_ratio": "32:27"}
|
||
]
|
||
}
|
||
c = as_user(make_user())
|
||
plain = _clip(c, source="plex", ref="777").json()
|
||
job = _clip(c, source="plex", ref="777", crop=_FRAME).json()
|
||
assert job["asset"]["id"] != plain["asset"]["id"]
|
||
assert job["edit_spec"]["crop"]["w"] == 0.25
|
||
job, asset = worker["run"](job)
|
||
assert (job.status, asset.status) == ("done", "ready")
|
||
vf = worker["cmd"][worker["cmd"].index("-vf") + 1]
|
||
# Stored to four decimals, the box's 16:9 is a hair off the file's: well under a pixel.
|
||
assert _crop_args(vf.split(",")[0]) == pytest.approx([0.25, 0.5, 0.5, 0.25], abs=1e-4)
|
||
|
||
|
||
def test_a_crop_frame_that_misses_the_picture_fails_and_says_so(
|
||
db, make_user, as_user, plex_episode, worker
|
||
):
|
||
# A 4:3 file, a frame drawn over its pillar bar in a 16:9 box.
|
||
worker["probe"] = {"streams": [{"codec_type": "video", "width": 640, "height": 480}]}
|
||
frame = {"x": 0.0, "y": 0.0, "w": 0.12, "h": 0.5, "aspect": 16 / 9}
|
||
job = _clip(as_user(make_user()), source="plex", ref="777", crop=frame).json()
|
||
job, asset = worker["run"](job)
|
||
assert (job.status, asset.status) == ("error", "error")
|
||
assert "crop" in job.error
|
||
assert "cmd" not in worker
|
||
|
||
|
||
def test_a_crop_frame_the_route_cannot_cut_says_why(db, make_user, as_user, yt_video):
|
||
r = _clip(as_user(make_user()), source="youtube", ref=yt_video, crop={**_FRAME, "w": 0.01})
|
||
assert r.status_code == 422
|
||
assert r.json()["detail"] == {"code": "clip", "reason": "bad_crop"}
|
||
|
||
|
||
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)",
|
||
)
|
||
|
||
|
||
# --- a YouTube clip goes with the user's own cookie, and its failure is diagnosed -----------------
|
||
|
||
_YT_COOKIES = "# Netscape HTTP Cookie File\n.youtube.com\tTRUE\t/\tTRUE\t0\tSID\tabc\n"
|
||
|
||
|
||
@pytest.fixture
|
||
def fernet(monkeypatch):
|
||
from cryptography.fernet import Fernet
|
||
|
||
from app import security
|
||
|
||
monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))
|
||
|
||
|
||
def _capture_extract(monkeypatch, asked: dict):
|
||
def extract(job_id, asset_id, url, spec, staging, adapter, **kw):
|
||
cookiefile = kw.get("cookiefile")
|
||
asked["cookiefile"] = cookiefile
|
||
asked["cookie_text"] = Path(cookiefile).read_text() if cookiefile else None
|
||
f = Path(staging) / "abcdefghijk.mp4"
|
||
f.write_bytes(b"range")
|
||
return {"requested_downloads": [{"filepath": str(f)}]}
|
||
|
||
monkeypatch.setattr(pipeline, "_extract_with_fallback", extract)
|
||
|
||
|
||
def test_a_youtube_clip_goes_signed_in_with_the_users_own_cookie_and_leaves_none_behind(
|
||
db, make_user, as_user, yt_video, worker, monkeypatch, fernet
|
||
):
|
||
from app.downloads import credentials
|
||
|
||
user = make_user()
|
||
credentials.upsert_credential(db, user.id, "youtube", _YT_COOKIES)
|
||
db.commit()
|
||
asked: dict = {}
|
||
_capture_extract(monkeypatch, asked)
|
||
job, _asset = worker["run"](_clip(as_user(user), source="youtube", ref=yt_video).json())
|
||
assert job.status == "done"
|
||
assert asked["cookie_text"] == _YT_COOKIES
|
||
assert not Path(asked["cookiefile"]).exists()
|
||
|
||
|
||
def test_a_youtube_clip_never_goes_with_another_users_cookie(
|
||
db, make_user, as_user, yt_video, worker, monkeypatch, fernet
|
||
):
|
||
from app.downloads import credentials
|
||
|
||
other = make_user()
|
||
credentials.upsert_credential(db, other.id, "youtube", _YT_COOKIES)
|
||
db.commit()
|
||
asked: dict = {}
|
||
_capture_extract(monkeypatch, asked)
|
||
job, _asset = worker["run"](_clip(as_user(make_user()), source="youtube", ref=yt_video).json())
|
||
assert job.status == "done" and asked["cookiefile"] is None
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("with_cookie", "detail"), [(False, "youtube.bot_wall"), (True, "youtube.cookie_rejected")]
|
||
)
|
||
def test_a_youtube_clip_on_the_bot_wall_is_diagnosed_like_a_download(
|
||
db, make_user, as_user, yt_video, worker, monkeypatch, fernet, with_cookie, detail
|
||
):
|
||
from app.downloads import credentials
|
||
|
||
user = make_user()
|
||
if with_cookie:
|
||
credentials.upsert_credential(db, user.id, "youtube", _YT_COOKIES)
|
||
db.commit()
|
||
|
||
def walled(*_a, **_k):
|
||
raise RuntimeError(f"ERROR: [youtube] {yt_video}: Sign in to confirm you’re not a bot.")
|
||
|
||
monkeypatch.setattr(pipeline, "_extract_with_fallback", walled)
|
||
job, asset = worker["run"](_clip(as_user(user), source="youtube", ref=yt_video).json())
|
||
assert (job.status, asset.status) == ("error", "error")
|
||
assert (job.error_code, job.error_detail) == ("login_required", detail)
|