zebra
f103b6c253
Generalize URL handling beyond YouTube to any yt-dlp-supported site (SoundCloud, Bandcamp, etc), single tracks and playlists/sets/albums. - probe_url(): one yt-dlp --flat-playlist probe classifies playlist vs track and returns per-entry Hits; YouTube playlists still use ytmusicapi. - _track_url(): YouTube tracks keep the music.youtube album-art URL; other platforms download via their native entry URL (no more videoId reconstruction). - Per-source folders: <root>/<artist>/<extractor>/ (soundcloud/bandcamp/youtube) instead of hardcoded youtube; download_single derives source from metadata. - download_hits() downloads pre-probed Hits; API probes once and passes hits into the job closure. Replaces YouTube-only is_playlist_url/expand_playlist. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
import time
|
|
import pytest
|
|
from server import jobs as jobs_mod
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_jobs():
|
|
jobs_mod.JOBS.clear()
|
|
yield
|
|
jobs_mod.JOBS.clear()
|
|
|
|
|
|
def _wait_done(client, auth, job_id, timeout=2.0):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
b = client.get(f"/jobs/{job_id}", headers=auth).json()
|
|
if b["status"] in ("done", "failed"):
|
|
return b
|
|
time.sleep(0.01)
|
|
raise AssertionError("job never finished")
|
|
|
|
|
|
def _mk_hit():
|
|
from server import mf
|
|
return mf.Hit(source="youtube", kind="track", title="t", artist="a", payload={"videoId": "1"})
|
|
|
|
|
|
def test_playlist_url_batch_job(client, auth, monkeypatch):
|
|
monkeypatch.setattr("server.app.mf.probe_url",
|
|
lambda url: ("playlist", "My Mix", [_mk_hit(), _mk_hit(), _mk_hit()]))
|
|
monkeypatch.setattr("server.app.mf.download_hits",
|
|
lambda hits, root, quality, dry_run: (2, 3))
|
|
r = client.post("/fetch", params={"q": "https://soundcloud.com/dj/sets/mix"}, headers=auth)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "queued"
|
|
assert body["hit"]["kind"] == "playlist"
|
|
done = _wait_done(client, auth, body["job_id"])
|
|
assert done["status"] == "done"
|
|
assert "2/3" in done["message"]
|
|
assert done["result"]["ok"] == 2
|
|
|
|
|
|
def test_playlist_zero_success_fails(client, auth, monkeypatch):
|
|
monkeypatch.setattr("server.app.mf.probe_url",
|
|
lambda url: ("playlist", "Dead Mix", [_mk_hit()]))
|
|
monkeypatch.setattr("server.app.mf.download_hits",
|
|
lambda hits, root, quality, dry_run: (0, 3))
|
|
body = client.post("/fetch", params={"q": "https://www.youtube.com/playlist?list=PLy"}, headers=auth).json()
|
|
done = _wait_done(client, auth, body["job_id"])
|
|
assert done["status"] == "failed"
|
|
|
|
|
|
def test_single_video_url_download(client, auth, monkeypatch):
|
|
monkeypatch.setattr("server.app.mf.probe_url", lambda url: ("track", "Song", []))
|
|
monkeypatch.setattr("server.app.mf.download_single",
|
|
lambda url, root, quality, dry_run: {"title": "Song", "artist": "A", "ok": True})
|
|
body = client.post("/fetch", params={"q": "https://soundcloud.com/a/song"}, headers=auth).json()
|
|
assert body["hit"]["kind"] == "track"
|
|
done = _wait_done(client, auth, body["job_id"])
|
|
assert done["status"] == "done"
|
|
assert "Song" in done["message"]
|
|
|
|
|
|
def test_search_query_still_works(client, auth, monkeypatch):
|
|
from server import mf
|
|
hit = mf.Hit(source="youtube", kind="track", title="T", artist="A", payload={"videoId": "x"})
|
|
monkeypatch.setattr("server.app.mf.build_combined_hits",
|
|
lambda q, limit, yt_first, lidarr_only, yt_only: [hit])
|
|
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
|
|
monkeypatch.setattr("server.app.actions.perform_fetch",
|
|
lambda chosen, hits, quality, root: {"path": "/x", "lidarr_album_id": None})
|
|
r = client.post("/fetch", params={"q": "Daft Punk - Discovery"}, headers=auth)
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "queued"
|