Jonathan
0c5e73eb17
Soulseek's slskd search polls up to ~2 min; searching adapters sequentially added that on top of Qobuz's instant search. Run the per-adapter searches concurrently via a ThreadPoolExecutor so total match time is the slowest single source, not their sum. Only Qobuz drives streamrip's shared asyncio loop and there's one Qobuz adapter, so no two threads use that loop at once. Results collected in adapter order (map preserves order) for deterministic candidate ordering; per-adapter failures still swallowed (down source = no candidates). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
233 lines
10 KiB
Python
233 lines
10 KiB
Python
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
|
|
|
|
|
|
class _DurationResolver:
|
|
"""Resolver that stamps a known MB total duration on the target (the real MB path)."""
|
|
def __init__(self, total_s, tracks=10):
|
|
self._total, self._tracks = total_s, tracks
|
|
def resolve(self, artist, album):
|
|
from lyra_worker.types import MBTarget
|
|
return MBTarget(artist=artist, album=album, track_count=self._tracks, total_duration_s=self._total)
|
|
|
|
|
|
def test_rejects_a_download_far_shorter_than_the_mb_total(conn):
|
|
job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
|
|
claim_next(conn)
|
|
# MB says ~40 min; only ~5 min of audio staged (truncated/preview tracks) → reject.
|
|
run_pipeline(conn, job_id, [FakeQobuz()], resolver=_DurationResolver(2400),
|
|
measure_duration=lambda _s: 300.0, dest_root="/tmp/lib")
|
|
state, _stage = _job_state(conn, job_id)
|
|
assert state == "needs_attention"
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT error FROM "Job" WHERE id = %s', (job_id,))
|
|
assert "too short" in cur.fetchone()[0]
|
|
cur.execute('SELECT count(*) FROM "LibraryItem" WHERE album = %s', ("In Rainbows",))
|
|
assert cur.fetchone()[0] == 0 # nothing imported
|
|
|
|
|
|
def test_accepts_a_download_within_the_duration_tolerance(conn):
|
|
job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
|
|
claim_next(conn)
|
|
# ~39 of ~40 min → within the 85% tolerance → imports normally.
|
|
run_pipeline(conn, job_id, [FakeQobuz()], resolver=_DurationResolver(2400),
|
|
measure_duration=lambda _s: 2340.0, dest_root="/tmp/lib")
|
|
assert _job_state(conn, job_id) == ("imported", "import")
|
|
|
|
|
|
def test_force_reacquires_an_album_already_in_library(conn):
|
|
# First acquisition puts the album in the library.
|
|
job1 = insert_request(conn, artist="Radiohead", album="In Rainbows")
|
|
claim_next(conn)
|
|
run_pipeline(conn, job1, [FakeQobuz()], dest_root="/tmp/lib")
|
|
# A forced re-acquire skips the dedupe → it searches + persists candidates again.
|
|
job2 = insert_request(conn, artist="Radiohead", album="In Rainbows", force=True)
|
|
claim_next(conn)
|
|
run_pipeline(conn, job2, [FakeQobuz()], dest_root="/tmp/lib", upgrade_cutoff=2)
|
|
|
|
assert _job_state(conn, job2) == ("imported", "import")
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT count(*) FROM "Candidate" WHERE "jobId" = %s', (job2,))
|
|
assert cur.fetchone()[0] >= 1 # dedupe bypassed — it actually ran the search
|
|
cur.execute('SELECT count(*) FROM "LibraryItem" WHERE artist = %s AND album = %s',
|
|
("Radiohead", "In Rainbows"))
|
|
assert cur.fetchone()[0] == 1 # still not duplicated (keep-if-better import)
|
|
|
|
|
|
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_searches_run_concurrently_across_adapters(conn):
|
|
# Two sources that each take ~0.5s to search must run in parallel: total match time is ~the
|
|
# slowest single search (~0.5s), not the sum (~1.0s). Also verifies concurrency drops no
|
|
# candidates — both sources' contributions are persisted.
|
|
import time as _t
|
|
|
|
class _SlowQobuz(FakeQobuz):
|
|
def search(self, target):
|
|
_t.sleep(0.5)
|
|
return super().search(target)
|
|
|
|
class _SlowSoulseek(FakeSoulseek):
|
|
def search(self, target):
|
|
_t.sleep(0.5)
|
|
return super().search(target)
|
|
|
|
job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
|
|
claim_next(conn)
|
|
start = _t.monotonic()
|
|
run_pipeline(conn, job_id, [_SlowQobuz(), _SlowSoulseek()], dest_root="/tmp/lib")
|
|
elapsed = _t.monotonic() - start
|
|
assert elapsed < 0.9 # concurrent (~0.5s), not sequential (~1.0s)
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT count(DISTINCT source) FROM "Candidate" WHERE "jobId" = %s', (job_id,))
|
|
assert cur.fetchone()[0] == 2 # both adapters contributed candidates
|
|
|
|
|
|
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")
|