66 lines
2.7 KiB
Python
66 lines
2.7 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 test_playlist_url_batch_job(client, auth, monkeypatch):
|
|
monkeypatch.setattr("server.app.mf.download_playlist",
|
|
lambda url, root, quality, dry_run: (2, 3, "My Mix"))
|
|
r = client.post("/fetch", params={"q": "https://music.youtube.com/playlist?list=PLx"}, 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.download_playlist",
|
|
lambda url, root, quality, dry_run: (0, 3, "Dead Mix"))
|
|
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.download_single",
|
|
lambda url, root, quality, dry_run: {"title": "Song", "artist": "A", "ok": True})
|
|
body = client.post("/fetch", params={"q": "https://music.youtube.com/watch?v=abc"}, 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"
|