Jonathan
4da0db4a1d
Probe each album exactly once and thread the ProbeResult through for tags, quality, and format instead of hardcoding LibraryItem.format to "FLAC" and re-probing on the tag-fallback path. Also skip album-level directories starting with "." (previously only artist-level dirs were skipped).
120 lines
5.2 KiB
Python
120 lines
5.2 KiB
Python
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), 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(), 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, 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(), 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"),
|
|
})
|
|
scan_library(conn, resolver, FakeProbe(quality_class=3), dest_root=str(tmp_path))
|
|
second = scan_library(conn, resolver, FakeProbe(quality_class=3), 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), 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(), 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}
|