from lyra_worker.adapters.fakes import FakeQobuz, FakeSoulseek, FakeYouTube, FailingAdapter from lyra_worker.claim import claim_next from lyra_worker.pipeline import run_pipeline from tests.conftest import insert_request def _job_state(conn, job_id): with conn.cursor() as cur: cur.execute('SELECT state, "currentStage" FROM "Job" WHERE id = %s', (job_id,)) return cur.fetchone() def _request_status(conn, job_id): with conn.cursor() as cur: cur.execute( 'SELECT status FROM "Request" WHERE id = (SELECT "requestId" FROM "Job" WHERE id = %s)', (job_id,), ) return cur.fetchone()[0] def _download_progress(conn, job_id): with conn.cursor() as cur: cur.execute('SELECT "downloadProgress" FROM "Job" WHERE id = %s', (job_id,)) return cur.fetchone()[0] def test_success_picks_best_source_and_imports(conn): job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job_id, [FakeYouTube(), FakeSoulseek(), FakeQobuz()], dest_root="/tmp/lib") assert _job_state(conn, job_id) == ("imported", "import") assert _request_status(conn, job_id) == "completed" assert _download_progress(conn, job_id) == 1.0 # a successful download completes at 100% with conn.cursor() as cur: cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == "qobuz" # highest quality won cur.execute('SELECT source, format FROM "LibraryItem" WHERE artist = %s', ("Radiohead",)) row = cur.fetchone() assert row[0] == "qobuz" def test_falls_through_when_best_download_fails(conn): job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) # FailingAdapter is source 'qobuz' (tier 0) whose download fails; Soulseek should win. run_pipeline(conn, job_id, [FailingAdapter(), FakeSoulseek()], dest_root="/tmp/lib") assert _job_state(conn, job_id) == ("imported", "import") assert _download_progress(conn, job_id) == 1.0 # falls through, but still completes at 100% with conn.cursor() as cur: cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == "soulseek" def test_fallthrough_marks_exactly_one_chosen(conn): # `chosen` now marks the source in flight at each attempt; after falling through from a # failed source to a working one, exactly ONE candidate stays chosen (the winner) — the # attempted-then-failed source must be un-marked (exclusive _mark_chosen). job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job_id, [FailingAdapter(), FakeSoulseek()], dest_root="/tmp/lib") with conn.cursor() as cur: cur.execute('SELECT count(*) FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == 1 cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == "soulseek" def test_no_match_goes_to_needs_attention(conn): job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job_id, [FakeQobuz(matches=False)], dest_root="/tmp/lib") assert _job_state(conn, job_id)[0] == "needs_attention" assert _request_status(conn, job_id) == "needs_attention" def test_all_downloads_fail_goes_to_needs_attention(conn): job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job_id, [FailingAdapter()], dest_root="/tmp/lib") assert _job_state(conn, job_id)[0] == "needs_attention" assert _download_progress(conn, job_id) == 0.0 # never reached 1.0 — all attempts failed def test_incomplete_download_goes_to_needs_attention(conn): from lyra_worker.adapters.fakes import FakeQobuz from lyra_worker.types import DownloadResult class TruncatingQobuz(FakeQobuz): def download(self, candidate, dest, on_progress): on_progress(1.0) # reports fewer tracks than the candidate promised -> integrity check fails return DownloadResult(ok=True, path=dest, track_count=candidate.track_count - 1) job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job_id, [TruncatingQobuz()], dest_root="/tmp/lib") assert _job_state(conn, job_id)[0] == "needs_attention" assert _request_status(conn, job_id) == "needs_attention" def test_dedupe_skips_already_in_library(conn): # First acquisition job1 = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job1, [FakeQobuz()], dest_root="/tmp/lib") # Second request for the same album job2 = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job2, [FakeQobuz()], dest_root="/tmp/lib") assert _job_state(conn, job2) == ("imported", "import") with conn.cursor() as cur: # dedupe path creates no candidates for the second job cur.execute('SELECT count(*) FROM "Candidate" WHERE "jobId" = %s', (job2,)) assert cur.fetchone()[0] == 0 cur.execute('SELECT count(*) FROM "LibraryItem" WHERE artist = %s AND album = %s', ("Radiohead", "In Rainbows")) assert cur.fetchone()[0] == 1 # not duplicated def test_below_threshold_candidates_are_persisted_for_diagnosis(conn): from lyra_worker.adapters.fakes import FakeQobuz from lyra_worker.types import MBTarget class WrongAlbumQobuz(FakeQobuz): def search(self, target): # returns a candidate for a different album -> scores below the gate return super().search(MBTarget(artist=target.artist, album="Some Completely Unrelated Title")) job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) run_pipeline(conn, job_id, [WrongAlbumQobuz()], dest_root="/tmp/lib") assert _job_state(conn, job_id)[0] == "needs_attention" with conn.cursor() as cur: cur.execute('SELECT count(*) FROM "Candidate" WHERE "jobId" = %s', (job_id,)) assert cur.fetchone()[0] >= 1 # found-but-rejected candidate retained for diagnosis def test_duplicate_adapter_names_raise(conn): import pytest from lyra_worker.adapters.fakes import FakeQobuz job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) with pytest.raises(ValueError): run_pipeline(conn, job_id, [FakeQobuz(), FakeQobuz()], dest_root="/tmp/lib")