116 lines
5.0 KiB
Python
116 lines
5.0 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"
|
|
|
|
|
|
def test_non_direct_link_resolves_and_fetches(client, auth, monkeypatch):
|
|
from server import mf
|
|
lid = mf.Hit(source="lidarr", kind="album", title="A Moment Apart",
|
|
artist="ODESZA", album="A Moment Apart", year="2017",
|
|
payload={"album": {"id": 9}})
|
|
yt = mf.Hit(source="youtube", kind="track", title="Bloom", artist="ODESZA",
|
|
payload={"url": "https://music.youtube.com/watch?v=YYY"})
|
|
monkeypatch.setattr("server.app.mf.resolve_link_hits",
|
|
lambda url, limit: ("ODESZA - Bloom", [lid, yt]))
|
|
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": None, "lidarr_album_id": 9})
|
|
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "queued"
|
|
assert body["hit"]["album"] == "A Moment Apart"
|
|
done = _wait_done(client, auth, body["job_id"])
|
|
assert done["status"] == "done"
|
|
|
|
|
|
def test_non_direct_link_resolve_failure_422(client, auth, monkeypatch):
|
|
from server import mf as mf_mod
|
|
|
|
def boom(url, limit):
|
|
raise mf_mod.OdesliError(url)
|
|
|
|
monkeypatch.setattr("server.app.mf.resolve_link_hits", boom)
|
|
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/bad"}, headers=auth)
|
|
assert r.status_code == 422
|
|
assert "resolve" in r.json()["message"].lower()
|
|
|
|
|
|
def test_non_direct_link_no_hits_404(client, auth, monkeypatch):
|
|
monkeypatch.setattr("server.app.mf.resolve_link_hits",
|
|
lambda url, limit: ("ODESZA - Bloom", []))
|
|
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
|
|
assert r.status_code == 404
|