251 lines
11 KiB
Python
251 lines
11 KiB
Python
"""The two-step Siri flow on the single entry point: free text at POST /fetch
|
|
parks a choice list, and POST /choose sends the picked release to the download
|
|
client. A URL grabs directly instead — see test_api_url.py."""
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from server import actions, jobs as jobs_mod, mf
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_jobs():
|
|
jobs_mod.JOBS.clear()
|
|
yield
|
|
jobs_mod.JOBS.clear()
|
|
|
|
|
|
def _wait(client, auth, job_id, status, timeout=3.0):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
body = client.get(f"/jobs/{job_id}", headers=auth).json()
|
|
if body["status"] == status:
|
|
return body
|
|
time.sleep(0.01)
|
|
raise AssertionError(f"job {job_id} never reached {status}")
|
|
|
|
|
|
def _release(guid="g1", seeders=24, quality="FLAC"):
|
|
return {"guid": guid, "indexerId": 2, "title": f"Overgrown-Lay_Cold-{guid}",
|
|
"size": 312 * 1024 * 1024, "seeders": seeders, "indexer": "Redacted",
|
|
"quality": {"quality": {"name": quality}}}
|
|
|
|
|
|
def _album_hit():
|
|
return mf.Hit(source="lidarr", kind="album", title="Lay Cold", album="Lay Cold",
|
|
artist="Overgrown",
|
|
payload={"album": {"title": "Lay Cold",
|
|
"artist": {"artistName": "Overgrown"}}})
|
|
|
|
|
|
@pytest.fixture
|
|
def lidarr(monkeypatch):
|
|
"""A Lidarr that already holds Overgrown and offers two releases."""
|
|
state = {"grabbed": [], "releases": [_release("a", 90), _release("b", 10, "MP3-320")],
|
|
"hits": [_album_hit()]}
|
|
monkeypatch.setattr("server.app.mf.build_combined_hits",
|
|
lambda q, limit, yt_first, lidarr_only, yt_only: state["hits"])
|
|
monkeypatch.setattr(actions.mf, "ensure_album", lambda album, root, search_all, dry_run: {
|
|
"album": {"id": 5}, "album_id": 5, "artist_name": "Overgrown",
|
|
"album_title": "Lay Cold", "artist_state": "existing", "album_state": "existing"})
|
|
monkeypatch.setattr(actions.mf, "list_releases", lambda album_id, **kw: state["releases"])
|
|
monkeypatch.setattr(actions.mf, "grab_release",
|
|
lambda rel, dry_run=False: state["grabbed"].append(rel["guid"]) or True)
|
|
return state
|
|
|
|
|
|
# --- find ------------------------------------------------------------------
|
|
def test_find_returns_a_job_immediately(client, auth, lidarr):
|
|
r = client.post("/fetch", params={"q": "Overgrown - Lay Cold"}, headers=auth)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "queued"
|
|
assert body["job_id"]
|
|
assert "Searching indexers now." in body["message"]
|
|
assert body["hit"]["album"] == "Lay Cold"
|
|
assert "choices" not in body
|
|
|
|
|
|
def test_find_parks_the_job_with_a_choice_list(client, auth, lidarr):
|
|
job_id = client.post("/fetch", params={"q": "Overgrown - Lay Cold"},
|
|
headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "awaiting_choice")
|
|
assert "Which one?" in body["message"]
|
|
labels = [c["label"] for c in body["choices"]]
|
|
assert labels == ["FLAC, 312 MB, 90 seeders, Redacted",
|
|
"MP3-320, 312 MB, 10 seeders, Redacted"]
|
|
assert [c["n"] for c in body["choices"]] == [1, 2]
|
|
# Shortcuts can only "Choose from List" plain strings, so each choice also
|
|
# carries a numbered label the client can read the number back off.
|
|
assert [c["pick_label"] for c in body["choices"]] == [
|
|
"1. FLAC, 312 MB, 90 seeders, Redacted",
|
|
"2. MP3-320, 312 MB, 10 seeders, Redacted"]
|
|
assert lidarr["grabbed"] == [] # nothing sent until the user picks
|
|
|
|
|
|
def test_find_choices_never_leak_the_raw_guid(client, auth, lidarr):
|
|
job_id = client.post("/fetch", params={"q": "x"}, headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "awaiting_choice")
|
|
assert all("guid" not in c for c in body["choices"])
|
|
assert "guid" not in str(body["result"])
|
|
assert "pending" not in body["result"]
|
|
|
|
|
|
def test_find_caps_the_list(client, auth, lidarr, monkeypatch):
|
|
lidarr["releases"] = [_release(f"g{i}") for i in range(30)]
|
|
monkeypatch.setattr(actions, "MAX_CHOICES", 5)
|
|
job_id = client.post("/fetch", params={"q": "x"}, headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "awaiting_choice")
|
|
assert len(body["choices"]) == 5
|
|
|
|
|
|
def test_find_auto_grabs_the_best_release_without_asking(client, auth, lidarr):
|
|
job_id = client.post("/fetch", params={"q": "x", "auto": "true"},
|
|
headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "done")
|
|
assert lidarr["grabbed"] == ["a"]
|
|
assert "download client" in body["message"]
|
|
|
|
|
|
def test_find_reports_no_release_without_pretending_to_download(client, auth, lidarr):
|
|
lidarr["releases"] = [] # Lidarr-only: no YouTube hit to fall back to
|
|
job_id = client.post("/fetch", params={"q": "x"}, headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "done")
|
|
assert "no indexer has a release" in body["message"]
|
|
|
|
|
|
def test_a_url_grabs_directly_and_never_asks(client, auth, lidarr, monkeypatch):
|
|
""""If a URL, grab directly" — the choice list is for free text only."""
|
|
monkeypatch.setattr("server.app.mf._is_direct_url", lambda url: False)
|
|
monkeypatch.setattr("server.app.mf.resolve_link_hits",
|
|
lambda url, limit: ("Overgrown - Lay Cold", lidarr["hits"]))
|
|
job_id = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"},
|
|
headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "done")
|
|
assert lidarr["grabbed"] == ["a"]
|
|
assert "download client" in body["message"]
|
|
|
|
|
|
def test_youtube_source_skips_lidarr(client, auth, lidarr, monkeypatch):
|
|
yt_hit = mf.Hit(source="youtube", kind="track", title="Lay Cold", artist="Overgrown",
|
|
album="Lay Cold", payload={"videoId": "abc"})
|
|
lidarr["hits"] = [yt_hit]
|
|
monkeypatch.setattr(actions.mf, "act_youtube", lambda *a, **kw: True)
|
|
job_id = client.post("/fetch", params={"q": "x", "source": "youtube"},
|
|
headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "done")
|
|
assert lidarr["grabbed"] == []
|
|
assert body["message"] == "Finished downloading 'Lay Cold' by Overgrown."
|
|
|
|
|
|
def test_find_adds_a_missing_artist_only(client, auth, lidarr, monkeypatch):
|
|
added = []
|
|
lidarr["hits"] = [mf.Hit(source="lidarr", kind="artist", title="Overgrown",
|
|
artist="Overgrown",
|
|
payload={"artist": {"artistName": "Overgrown"}})]
|
|
monkeypatch.setattr(actions.mf, "ensure_artist", lambda meta, root, search_all, dry_run:
|
|
(added.append(meta["artistName"]),
|
|
{"artist": {"id": 9}, "state": "added", "name": "Overgrown"})[1])
|
|
job_id = client.post("/fetch", params={"q": "Overgrown"}, headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "done")
|
|
assert added == ["Overgrown"]
|
|
assert "Added Overgrown to Lidarr" in body["message"]
|
|
assert lidarr["grabbed"] == []
|
|
|
|
|
|
def test_no_hits_is_a_404_not_a_job(client, auth, lidarr):
|
|
lidarr["hits"] = []
|
|
r = client.post("/fetch", params={"q": "zzz"}, headers=auth)
|
|
assert r.status_code == 404
|
|
assert "zzz" in r.json()["message"]
|
|
|
|
|
|
def test_failure_message_explains_why(client, auth, lidarr, monkeypatch):
|
|
monkeypatch.setattr(actions.mf, "ensure_album", lambda album, root, search_all, dry_run: {
|
|
"album": None, "album_id": None, "artist_name": "Overgrown",
|
|
"album_title": "Lay Cold", "artist_state": "existing", "album_state": "missing"})
|
|
job_id = client.post("/fetch", params={"q": "x"}, headers=auth).json()["job_id"]
|
|
body = _wait(client, auth, job_id, "failed")
|
|
assert "no album called 'Lay Cold'" in body["message"]
|
|
|
|
|
|
def test_find_needs_the_api_key(client, lidarr):
|
|
assert client.post("/fetch", params={"q": "x"}).status_code == 401
|
|
|
|
|
|
# --- choose ----------------------------------------------------------------
|
|
def _parked(client, auth, lidarr):
|
|
job_id = client.post("/fetch", params={"q": "x"}, headers=auth).json()["job_id"]
|
|
_wait(client, auth, job_id, "awaiting_choice")
|
|
return job_id
|
|
|
|
|
|
def test_choose_sends_the_picked_release(client, auth, lidarr):
|
|
job_id = _parked(client, auth, lidarr)
|
|
r = client.post("/choose", params={"job_id": job_id, "n": 2}, headers=auth)
|
|
assert r.status_code == 200
|
|
assert "Sending 'Lay Cold' by Overgrown" in r.json()["message"]
|
|
body = _wait(client, auth, job_id, "done")
|
|
assert lidarr["grabbed"] == ["b"]
|
|
assert body["message"] == ("Sent 'Lay Cold' by Overgrown to the download client. "
|
|
"MP3-320, 312 MB, 10 seeders, Redacted.")
|
|
assert "choices" not in body
|
|
|
|
|
|
def test_choose_twice_is_refused(client, auth, lidarr):
|
|
job_id = _parked(client, auth, lidarr)
|
|
client.post("/choose", params={"job_id": job_id, "n": 1}, headers=auth)
|
|
_wait(client, auth, job_id, "done")
|
|
r = client.post("/choose", params={"job_id": job_id, "n": 1}, headers=auth)
|
|
assert r.status_code == 409
|
|
assert lidarr["grabbed"] == ["a"]
|
|
|
|
|
|
def test_choose_out_of_range_is_refused(client, auth, lidarr):
|
|
job_id = _parked(client, auth, lidarr)
|
|
r = client.post("/choose", params={"job_id": job_id, "n": 9}, headers=auth)
|
|
assert r.status_code == 409
|
|
assert "between 1 and 2" in r.json()["message"]
|
|
assert lidarr["grabbed"] == []
|
|
|
|
|
|
def test_choose_unknown_job_is_404(client, auth):
|
|
r = client.post("/choose", params={"job_id": "nope", "n": 1}, headers=auth)
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_the_old_split_endpoints_are_gone(client, auth):
|
|
assert client.post("/lidarr/find", params={"q": "x"}, headers=auth).status_code == 404
|
|
assert client.post("/lidarr/choose", params={"job_id": "x", "n": 1},
|
|
headers=auth).status_code == 404
|
|
|
|
|
|
def test_choose_zero_is_rejected_by_validation(client, auth, lidarr):
|
|
job_id = _parked(client, auth, lidarr)
|
|
assert client.post("/choose", params={"job_id": job_id, "n": 0},
|
|
headers=auth).status_code == 422
|
|
|
|
|
|
def test_choose_needs_the_api_key(client, auth, lidarr):
|
|
job_id = _parked(client, auth, lidarr)
|
|
assert client.post("/choose", params={"job_id": job_id, "n": 1}).status_code == 401
|
|
|
|
|
|
def test_a_parked_job_expires(client, auth, lidarr, monkeypatch):
|
|
job_id = _parked(client, auth, lidarr)
|
|
jobs_mod.JOBS[job_id].expires_at = time.time() - 1
|
|
body = client.get(f"/jobs/{job_id}", headers=auth).json()
|
|
assert body["status"] == "expired"
|
|
assert "expired" in body["message"]
|
|
r = client.post("/choose", params={"job_id": job_id, "n": 1}, headers=auth)
|
|
assert r.status_code == 409
|
|
assert lidarr["grabbed"] == []
|
|
|
|
|
|
def test_choose_reports_a_lidarr_refusal(client, auth, lidarr, monkeypatch):
|
|
job_id = _parked(client, auth, lidarr)
|
|
monkeypatch.setattr(actions.mf, "grab_release", lambda rel, dry_run=False: False)
|
|
client.post("/choose", params={"job_id": job_id, "n": 1}, headers=auth)
|
|
body = _wait(client, auth, job_id, "failed")
|
|
assert "refused the release" in body["message"]
|