peter
f221c1962a
A clip fetched its range signed out, so a video behind YouTube's per-video "not a bot" wall could never be clipped while its download went through with the user's cookie - and the failure copy told the user to add a cookie the clip would not use. The clip asset is private by construction (`clip.asset_ref` names its user), so `usable_clip_cookie` attaches the user's YouTube cookie on that identity, and a failed range fetch now carries the download catalog's diagnosis (`youtube.bot_wall` / `youtube.cookie_rejected`) instead of a bare category.
521 lines
19 KiB
Python
521 lines
19 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 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)",
|
||
)
|
||
|
||
|
||
# --- 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)
|