"""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"]