Jonathan
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
from lyra_worker.adapters.qobuz import QobuzAdapter
|
|
from lyra_worker.quality import quality_class
|
|
from lyra_worker.types import MBTarget
|
|
|
|
TARGET = MBTarget(artist="John Mayer", album="Continuum", track_count=12)
|
|
|
|
|
|
class FakeQobuzClient:
|
|
def __init__(self, configured=True, results=None, fail=False):
|
|
self._configured = configured
|
|
self._results = results if results is not None else [
|
|
{"source_ref": "qz123", "title": "Continuum", "artist": "John Mayer",
|
|
"track_count": 12, "bit_depth": 24, "sampling_rate": 96.0}
|
|
]
|
|
self._fail = fail
|
|
|
|
def is_configured(self):
|
|
return self._configured
|
|
|
|
def search_album(self, artist, album):
|
|
return self._results
|
|
|
|
def download(self, source_ref, dest, on_progress):
|
|
if self._fail:
|
|
raise RuntimeError("qobuz boom")
|
|
on_progress(1.0)
|
|
return {"track_count": 12, "path": dest}
|
|
|
|
|
|
def test_name_and_tier():
|
|
a = QobuzAdapter(FakeQobuzClient())
|
|
assert a.name == "qobuz"
|
|
assert a.tier == 0
|
|
|
|
|
|
def test_health_reflects_configuration():
|
|
assert QobuzAdapter(FakeQobuzClient(configured=True)).health() is True
|
|
assert QobuzAdapter(FakeQobuzClient(configured=False)).health() is False
|
|
|
|
|
|
def test_search_maps_to_lossless_hires_candidate():
|
|
cands = QobuzAdapter(FakeQobuzClient()).search(TARGET)
|
|
assert len(cands) == 1
|
|
c = cands[0]
|
|
assert c.source == "qobuz"
|
|
assert c.source_ref == "qz123"
|
|
assert c.matched_artist == "John Mayer"
|
|
assert c.matched_album == "Continuum"
|
|
assert c.track_count == 12
|
|
assert c.source_tier == 0
|
|
assert c.quality.lossless is True
|
|
assert c.quality.fmt == "FLAC"
|
|
assert c.quality.sample_rate == 96000 # 96.0 kHz -> Hz
|
|
assert c.quality.bit_depth == 24
|
|
assert quality_class(c.quality) == 3 # hi-res
|
|
|
|
|
|
def test_search_empty_when_no_results():
|
|
assert QobuzAdapter(FakeQobuzClient(results=[])).search(TARGET) == []
|
|
|
|
|
|
def test_download_success_and_failure():
|
|
ok = QobuzAdapter(FakeQobuzClient()).download(
|
|
QobuzAdapter(FakeQobuzClient()).search(TARGET)[0], "/tmp/x", lambda _p: None)
|
|
assert ok.ok is True and ok.track_count == 12
|
|
bad = QobuzAdapter(FakeQobuzClient(fail=True)).download(
|
|
QobuzAdapter(FakeQobuzClient()).search(TARGET)[0], "/tmp/x", lambda _p: None)
|
|
assert bad.ok is False and bad.error
|