Jonathan
1c1b98967a
The verify step checked track COUNT but not playtime, so a truncated file or a short preview substituted for a real track could import as "complete". Add a total-duration check: sum the staged audio's playtime (mutagen) and fail the job when it falls below 85% of MusicBrainz's expected total (target.total_duration_s, already resolved). Generous tolerance so edition/encoding differences don't false-positive; bonus tracks (over-long) always pass. The duration reader is injected (measure_duration param, default the real mutagen-based reader) so the fail/pass paths are unit-tested; it returns None for the offline fakes (no real audio), skipping the check exactly like the staged-count guard. worker 233 tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
206 lines
9.3 KiB
Python
206 lines
9.3 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_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")
|