Jonathan
41 lines
1.7 KiB
Python
41 lines
1.7 KiB
Python
from lyra_worker.pipeline import _import
|
|
from lyra_worker.types import Candidate, Quality
|
|
from tests.conftest import insert_request
|
|
|
|
|
|
def _winner(fmt, lossless, bit_depth=None, sample_rate=None):
|
|
q = Quality(fmt=fmt, lossless=lossless, bit_depth=bit_depth, sample_rate=sample_rate)
|
|
return Candidate(source="qobuz", source_ref="ref", matched_artist="A", matched_album="B",
|
|
quality=q, track_count=10, source_tier=0)
|
|
|
|
|
|
def _library_row(conn, artist, album):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT path, "qualityClass", source FROM "LibraryItem" WHERE artist = %s AND album = %s',
|
|
(artist, album),
|
|
)
|
|
return cur.fetchone()
|
|
|
|
|
|
def test_higher_quality_replaces(conn):
|
|
from lyra_worker.types import MBTarget
|
|
|
|
target = MBTarget(artist="A", album="B")
|
|
job1 = insert_request(conn, artist="A", album="B")
|
|
_import(conn, job1, target, _winner("MP3", False), "/m/lossy") # qualityClass 1
|
|
job2 = insert_request(conn, artist="A", album="B")
|
|
_import(conn, job2, target, _winner("FLAC", True, 24, 96000), "/m/hires") # qualityClass 3
|
|
assert _library_row(conn, "A", "B") == ("/m/hires", 3, "qobuz")
|
|
|
|
|
|
def test_lower_quality_does_not_replace(conn):
|
|
from lyra_worker.types import MBTarget
|
|
|
|
target = MBTarget(artist="A", album="B")
|
|
job1 = insert_request(conn, artist="A", album="B")
|
|
_import(conn, job1, target, _winner("FLAC", True, 24, 96000), "/m/hires") # qualityClass 3
|
|
job2 = insert_request(conn, artist="A", album="B")
|
|
_import(conn, job2, target, _winner("MP3", False), "/m/lossy") # qualityClass 1
|
|
assert _library_row(conn, "A", "B") == ("/m/hires", 3, "qobuz")
|