Jonathan
aafd8b15d5
A followed artist previously showed only the one owned album ("of 1")
because scan_library recorded just the resolved release. Now scan
browses each artist's full MusicBrainz release-group list once per
run and stores it as unmonitored MonitoredRelease rows, matching
web-follow behavior. The owned album keeps its monitored/fulfilled
state via ON CONFLICT DO NOTHING.
145 lines
6.8 KiB
Python
145 lines
6.8 KiB
Python
from lyra_worker.adapters.fakes import FakeMbBrowser
|
|
from lyra_worker.browser import ReleaseGroupInfo
|
|
from lyra_worker.probe import ProbeResult
|
|
from lyra_worker.scan import scan_library
|
|
from lyra_worker.types import MBTarget
|
|
|
|
|
|
class FakeProbe:
|
|
"""Returns a canned ProbeResult regardless of path (tests control identification via folders)."""
|
|
def __init__(self, artist="", album="", fmt="FLAC", quality_class=3):
|
|
self._r = ProbeResult(artist=artist, album=album, fmt=fmt, quality_class=quality_class)
|
|
def probe(self, path):
|
|
return self._r
|
|
|
|
|
|
class FakeResolver:
|
|
"""resolve(artist, album) -> MBTarget or None, keyed by (artist, album)."""
|
|
def __init__(self, table):
|
|
self._t = dict(table)
|
|
def resolve(self, artist, album):
|
|
return self._t.get((artist, album))
|
|
|
|
|
|
def _album(tmp_path, artist, album_folder, files=("01 Track.flac",)):
|
|
d = tmp_path / artist / album_folder
|
|
d.mkdir(parents=True)
|
|
for f in files:
|
|
(d / f).write_bytes(b"")
|
|
return d
|
|
|
|
|
|
def _counts(conn):
|
|
with conn.cursor() as cur:
|
|
out = {}
|
|
for t in ("LibraryItem", "MonitoredRelease", "WatchedArtist"):
|
|
cur.execute(f'SELECT count(*) FROM "{t}"')
|
|
out[t] = cur.fetchone()[0]
|
|
return out
|
|
|
|
|
|
def test_scan_records_have_monitored_and_follows_artist(conn, tmp_path):
|
|
_album(tmp_path, "John Mayer", "Continuum (2006)")
|
|
resolver = FakeResolver({
|
|
("John Mayer", "Continuum"): MBTarget(
|
|
artist="John Mayer", album="Continuum", year=2006, rg_mbid="rg1", artist_mbid="a1"),
|
|
})
|
|
result = scan_library(conn, resolver, FakeProbe(quality_class=3), FakeMbBrowser(), dest_root=str(tmp_path))
|
|
|
|
assert (result.imported, result.skipped) == (1, 0)
|
|
assert _counts(conn) == {"LibraryItem": 1, "MonitoredRelease": 1, "WatchedArtist": 1}
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT source, "qualityClass" FROM "LibraryItem" WHERE artist=%s AND album=%s',
|
|
("John Mayer", "Continuum"))
|
|
assert cur.fetchone() == ("scan", 3)
|
|
cur.execute('SELECT monitored, state, "currentQualityClass" FROM "MonitoredRelease" WHERE "rgMbid"=%s',
|
|
("rg1",))
|
|
assert cur.fetchone() == (True, "fulfilled", 3)
|
|
cur.execute('SELECT "autoMonitorFuture" FROM "WatchedArtist" WHERE mbid=%s', ("a1",))
|
|
assert cur.fetchone()[0] is False
|
|
|
|
|
|
def test_scan_skips_unmatched_albums(conn, tmp_path):
|
|
_album(tmp_path, "Obscure", "Nope (1999)")
|
|
result = scan_library(conn, FakeResolver({}), FakeProbe(), FakeMbBrowser(), dest_root=str(tmp_path))
|
|
assert (result.imported, result.skipped) == (0, 1)
|
|
assert _counts(conn) == {"LibraryItem": 0, "MonitoredRelease": 0, "WatchedArtist": 0}
|
|
|
|
|
|
def test_scan_falls_back_to_tags_when_folder_does_not_resolve(conn, tmp_path):
|
|
_album(tmp_path, "GARBLE", "CD1") # folder names won't resolve
|
|
resolver = FakeResolver({
|
|
("Radiohead", "In Rainbows"): MBTarget(
|
|
artist="Radiohead", album="In Rainbows", year=2007, rg_mbid="rg9", artist_mbid="a9"),
|
|
})
|
|
probe = FakeProbe(artist="Radiohead", album="In Rainbows", quality_class=3) # embedded tags
|
|
result = scan_library(conn, resolver, probe, FakeMbBrowser(), dest_root=str(tmp_path))
|
|
assert (result.imported, result.skipped) == (1, 0)
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT count(*) FROM "MonitoredRelease" WHERE "rgMbid"=%s', ("rg9",))
|
|
assert cur.fetchone()[0] == 1
|
|
|
|
|
|
def test_scan_skips_folders_without_audio(conn, tmp_path):
|
|
d = tmp_path / "Artist" / "Album"
|
|
d.mkdir(parents=True)
|
|
(d / "cover.jpg").write_bytes(b"") # no audio
|
|
result = scan_library(conn, FakeResolver({}), FakeProbe(), FakeMbBrowser(), dest_root=str(tmp_path))
|
|
assert (result.imported, result.skipped) == (0, 0)
|
|
|
|
|
|
def test_scan_is_idempotent(conn, tmp_path):
|
|
_album(tmp_path, "John Mayer", "Continuum (2006)")
|
|
resolver = FakeResolver({
|
|
("John Mayer", "Continuum"): MBTarget(
|
|
artist="John Mayer", album="Continuum", year=2006, rg_mbid="rg1", artist_mbid="a1"),
|
|
})
|
|
browser = FakeMbBrowser()
|
|
scan_library(conn, resolver, FakeProbe(quality_class=3), browser, dest_root=str(tmp_path))
|
|
second = scan_library(conn, resolver, FakeProbe(quality_class=3), browser, dest_root=str(tmp_path))
|
|
assert second.imported == 0 # already recorded
|
|
assert _counts(conn) == {"LibraryItem": 1, "MonitoredRelease": 1, "WatchedArtist": 1}
|
|
|
|
|
|
def test_scan_records_the_probed_format(conn, tmp_path):
|
|
_album(tmp_path, "John Mayer", "Sob Rock (2021)")
|
|
resolver = FakeResolver({
|
|
("John Mayer", "Sob Rock"): MBTarget(
|
|
artist="John Mayer", album="Sob Rock", year=2021, rg_mbid="rg2", artist_mbid="a1"),
|
|
})
|
|
scan_library(conn, resolver, FakeProbe(fmt="MP3", quality_class=1), FakeMbBrowser(), dest_root=str(tmp_path))
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT format, "qualityClass" FROM "LibraryItem" WHERE album=%s', ("Sob Rock",))
|
|
assert cur.fetchone() == ("MP3", 1) # the probed format is recorded, not a hardcoded FLAC
|
|
|
|
|
|
def test_scan_skips_hidden_album_dirs(conn, tmp_path):
|
|
hidden = tmp_path / "John Mayer" / ".sync"
|
|
hidden.mkdir(parents=True)
|
|
(hidden / "01 Track.flac").write_bytes(b"")
|
|
result = scan_library(conn, FakeResolver({}), FakeProbe(), FakeMbBrowser(), dest_root=str(tmp_path))
|
|
assert (result.imported, result.skipped) == (0, 0) # hidden album dir not scanned
|
|
assert _counts(conn) == {"LibraryItem": 0, "MonitoredRelease": 0, "WatchedArtist": 0}
|
|
|
|
|
|
def test_scan_populates_full_discography_unmonitored(conn, tmp_path):
|
|
from lyra_worker.adapters.fakes import FakeMbBrowser
|
|
from lyra_worker.browser import ReleaseGroupInfo
|
|
_album(tmp_path, "50 Cent", "Get Rich or Die Tryin' (2003)")
|
|
resolver = FakeResolver({
|
|
("50 Cent", "Get Rich or Die Tryin'"): MBTarget(
|
|
artist="50 Cent", album="Get Rich or Die Tryin'", year=2003, rg_mbid="rg-owned", artist_mbid="a50"),
|
|
})
|
|
browser = FakeMbBrowser(releases={"a50": [
|
|
ReleaseGroupInfo("rg-owned", "Get Rich or Die Tryin'", "Album", (), "2003-02-06"),
|
|
ReleaseGroupInfo("rg-other", "The Massacre", "Album", (), "2005-03-03"),
|
|
]})
|
|
scan_library(conn, resolver, FakeProbe(quality_class=2), browser, dest_root=str(tmp_path))
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT count(*) FROM "MonitoredRelease" WHERE "artistName"=%s', ("50 Cent",))
|
|
assert cur.fetchone()[0] == 2 # full discography populated, not just the owned album
|
|
cur.execute('SELECT monitored, state FROM "MonitoredRelease" WHERE "rgMbid"=%s', ("rg-owned",))
|
|
assert cur.fetchone() == (True, "fulfilled") # owned album stays monitored+fulfilled
|
|
cur.execute('SELECT monitored, state FROM "MonitoredRelease" WHERE "rgMbid"=%s', ("rg-other",))
|
|
assert cur.fetchone() == (False, "wanted") # the rest are unmonitored
|