From 48debb792681ed3f7d7cc5cfc21ccbe33640374d Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Thu, 4 Jun 2026 09:46:35 -0700 Subject: [PATCH] Library re-tag: seam tests for the job scan, apply handler, and source-id embed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the kettui gap — the orchestration was unproven. Injected-fake seam tests (temp sqlite + real empty track files, no metadata APIs / no real tag writes): - embed_known_source_ids: builds the right canonical id_tags from flat db keys, honors the musicbrainz embed gate, no-ops when there's nothing to write. - library_retag scan: produces a detailed finding with the per-track old->new diff + stamped source ids, and skips an album that's already correct. - _add_source_ids: per-source key mapping. - _fix_library_retag apply: writes each track's payload, and reports failure when files are unreachable. 476 tests pass; ruff clean. --- tests/test_embed_known_source_ids.py | 69 ++++++++++++ tests/test_library_retag_job.py | 158 +++++++++++++++++++++++++++ 2 files changed, 227 insertions(+) create mode 100644 tests/test_embed_known_source_ids.py create mode 100644 tests/test_library_retag_job.py diff --git a/tests/test_embed_known_source_ids.py b/tests/test_embed_known_source_ids.py new file mode 100644 index 00000000..9aabf53e --- /dev/null +++ b/tests/test_embed_known_source_ids.py @@ -0,0 +1,69 @@ +"""Seam test: embed_known_source_ids builds the right id_tags and routes them +through the canonical frame writer (no API re-fetch).""" + +import sys +import types +from types import SimpleNamespace + +# Stub spotipy/config so core.metadata.source imports cleanly in tests. +if 'spotipy' not in sys.modules: + sp = types.ModuleType('spotipy'); oa = types.ModuleType('spotipy.oauth2') + sp.Spotify = type('S', (), {}); oa.SpotifyOAuth = oa.SpotifyClientCredentials = type('O', (), {}) + sp.oauth2 = oa; sys.modules['spotipy'] = sp; sys.modules['spotipy.oauth2'] = oa +if 'config.settings' not in sys.modules: + cm = types.ModuleType('config'); sm = types.ModuleType('config.settings') + + class _Cfg: + def get(self, k, d=None): return d + def get_active_media_server(self): return 'plex' + sm.config_manager = _Cfg(); cm.settings = sm + sys.modules['config'] = cm; sys.modules['config.settings'] = sm + +from core.metadata import source as src + + +def test_builds_id_tags_from_flat_keys_and_calls_writer(monkeypatch): + captured = {} + monkeypatch.setattr(src, 'get_mutagen_symbols', lambda: SimpleNamespace()) + monkeypatch.setattr(src, 'get_config_manager', lambda: SimpleNamespace(get=lambda k, d=None: d)) + + def _fake_write(audio, metadata, pp, cfg, symbols): + captured['id_tags'] = dict(pp['id_tags']) + monkeypatch.setattr(src, '_write_embedded_metadata', _fake_write) + + meta = { + 'spotify_track_id': 'sp_t', 'spotify_album_id': 'sp_a', + 'itunes_track_id': 'it_t', + 'musicbrainz_recording_id': 'mb_rec', 'musicbrainz_release_id': 'mb_rel', + } + written = src.embed_known_source_ids(object(), meta) + + assert captured['id_tags']['SPOTIFY_TRACK_ID'] == 'sp_t' + assert captured['id_tags']['SPOTIFY_ALBUM_ID'] == 'sp_a' + assert captured['id_tags']['ITUNES_TRACK_ID'] == 'it_t' + assert captured['id_tags']['MUSICBRAINZ_RECORDING_ID'] == 'mb_rec' + assert captured['id_tags']['MUSICBRAINZ_RELEASE_ID'] == 'mb_rel' + assert set(written) >= {'SPOTIFY_TRACK_ID', 'MUSICBRAINZ_RECORDING_ID'} + + +def test_no_ids_writes_nothing(monkeypatch): + called = {'n': 0} + monkeypatch.setattr(src, 'get_mutagen_symbols', lambda: SimpleNamespace()) + monkeypatch.setattr(src, 'get_config_manager', lambda: SimpleNamespace(get=lambda k, d=None: d)) + monkeypatch.setattr(src, '_write_embedded_metadata', + lambda *a, **k: called.__setitem__('n', called['n'] + 1)) + assert src.embed_known_source_ids(object(), {'title': 'x'}) == [] + assert called['n'] == 0 # nothing to embed → writer never called + + +def test_musicbrainz_gated_off(monkeypatch): + captured = {} + monkeypatch.setattr(src, 'get_mutagen_symbols', lambda: SimpleNamespace()) + # mb embed disabled + monkeypatch.setattr(src, 'get_config_manager', + lambda: SimpleNamespace(get=lambda k, d=None: False if k == 'musicbrainz.embed_tags' else d)) + monkeypatch.setattr(src, '_write_embedded_metadata', + lambda audio, m, pp, cfg, sym: captured.update(pp['id_tags'])) + src.embed_known_source_ids(object(), {'spotify_track_id': 'sp', 'musicbrainz_recording_id': 'mb'}) + assert 'SPOTIFY_TRACK_ID' in captured + assert 'MUSICBRAINZ_RECORDING_ID' not in captured # gated off diff --git a/tests/test_library_retag_job.py b/tests/test_library_retag_job.py new file mode 100644 index 00000000..28ab0c22 --- /dev/null +++ b/tests/test_library_retag_job.py @@ -0,0 +1,158 @@ +"""Seam tests for the Library Re-tag job scan + the apply handler. + +Injected fakes only — no metadata APIs, no real tag writes. A temp sqlite db ++ real (empty) track files exercise the orchestration: scan -> detailed +finding, and finding -> per-track write payload. +""" + +import sqlite3 +import sys +import types +from types import SimpleNamespace + +# Stub optional deps so the modules import in the test env. +if 'spotipy' not in sys.modules: + sp = types.ModuleType('spotipy'); oa = types.ModuleType('spotipy.oauth2') + sp.Spotify = type('S', (), {}); oa.SpotifyOAuth = oa.SpotifyClientCredentials = type('O', (), {}) + sp.oauth2 = oa; sys.modules['spotipy'] = sp; sys.modules['spotipy.oauth2'] = oa +if 'config.settings' not in sys.modules: + cm = types.ModuleType('config'); sm = types.ModuleType('config.settings') + + class _Cfg: + def get(self, k, d=None): return d + def get_active_media_server(self): return 'plex' + sm.config_manager = _Cfg(); cm.settings = sm + sys.modules['config'] = cm; sys.modules['config.settings'] = sm + +from core.repair_jobs import library_retag as lr + + +def _db_with_album(path, track_file, current_title='Old Title'): + conn = sqlite3.connect(path) + c = conn.cursor() + c.execute("CREATE TABLE artists (id INTEGER PRIMARY KEY, name TEXT)") + c.execute("""CREATE TABLE albums (id INTEGER PRIMARY KEY, title TEXT, artist_id INTEGER, + spotify_album_id TEXT, itunes_album_id TEXT, deezer_id TEXT, musicbrainz_release_id TEXT)""") + c.execute("""CREATE TABLE tracks (id INTEGER PRIMARY KEY, album_id INTEGER, title TEXT, + track_number INTEGER, disc_number INTEGER, file_path TEXT)""") + c.execute("INSERT INTO artists (id, name) VALUES (1, 'Real Artist')") + c.execute("INSERT INTO albums (id, title, artist_id, spotify_album_id) VALUES (1, 'Real Album', 1, 'sp_alb')") + c.execute("INSERT INTO tracks (id, album_id, title, track_number, disc_number, file_path) VALUES (1, 1, ?, 1, 1, ?)", + (current_title, track_file)) + conn.commit() + return conn + + +def _context(conn, settings): + findings = [] + return SimpleNamespace( + db=SimpleNamespace(_get_connection=lambda: conn), + config_manager=SimpleNamespace(get=lambda k, d=None: {f'repair.jobs.library_retag.settings': settings}.get(k, d)), + check_stop=lambda: False, wait_if_paused=lambda: False, + update_progress=lambda *a, **k: None, report_progress=lambda *a, **k: None, + create_finding=lambda **kw: (findings.append(kw) or True), + findings=findings, + ) + + +_ALBUM_META = {'name': 'Real Album', 'artists': [{'name': 'Real Artist'}], + 'year': '2021', 'genres': ['Rock'], 'total_tracks': 1, + 'image_url': 'http://art/cover.jpg'} +_SRC_TRACKS = [{'name': 'Real Title', 'track_number': 1, 'disc_number': 1, 'id': 'sp_trk'}] + + +def _patch_source(monkeypatch, current_tags): + monkeypatch.setattr(lr, 'get_album_for_source', lambda s, i: _ALBUM_META) + monkeypatch.setattr(lr, 'get_album_tracks_for_source', lambda s, i: list(_SRC_TRACKS)) + monkeypatch.setattr(lr, '_read_current_tags', lambda p: dict(current_tags)) + + +def test_scan_creates_detailed_finding(tmp_path, monkeypatch): + track = tmp_path / 'track.flac'; track.write_bytes(b'') + conn = _db_with_album(str(tmp_path / 'm.db'), str(track), current_title='Old Title') + ctx = _context(conn, {'mode': 'overwrite', 'cover_art': 'replace', 'source': 'spotify'}) + _patch_source(monkeypatch, { + 'title': 'Old Title', 'album_artist': 'Real Artist', 'album': 'Real Album', + 'year': '2021', 'genre': 'Rock', 'track_number': 1, 'disc_number': 1, + }) + + result = lr.LibraryRetagJob().scan(ctx) + + assert result.findings_created == 1 + d = ctx.findings[0]['details'] + assert ctx.findings[0]['finding_type'] == 'library_retag' + assert d['source'] == 'spotify' + assert d['cover_action'] == 'replace' + tp = d['tracks'][0] + assert tp['changes']['title'] == {'old': 'Old Title', 'new': 'Real Title'} + # source ids stamped onto the write payload + assert tp['db_data']['spotify_album_id'] == 'sp_alb' + assert tp['db_data']['spotify_track_id'] == 'sp_trk' + assert tp['db_data']['title'] == 'Real Title' + + +def test_scan_skips_album_already_correct(tmp_path, monkeypatch): + track = tmp_path / 'track.flac'; track.write_bytes(b'') + conn = _db_with_album(str(tmp_path / 'm.db'), str(track), current_title='Real Title') + # cover skipped + tags already match → nothing to do + ctx = _context(conn, {'mode': 'overwrite', 'cover_art': 'skip', 'source': 'spotify'}) + _patch_source(monkeypatch, { + 'title': 'Real Title', 'album_artist': 'Real Artist', 'album': 'Real Album', + 'year': '2021', 'genre': 'Rock', 'track_number': 1, 'disc_number': 1, + }) + + result = lr.LibraryRetagJob().scan(ctx) + assert result.findings_created == 0 + assert ctx.findings == [] + assert result.skipped >= 1 + + +def test_add_source_ids_maps_per_source(): + db = {} + lr._add_source_ids(db, 'spotify', 'AL', {'id': 'TR'}) + assert db == {'spotify_album_id': 'AL', 'spotify_track_id': 'TR'} + db2 = {} + lr._add_source_ids(db2, 'musicbrainz', 'REL', {'id': 'REC'}) + assert db2 == {'musicbrainz_release_id': 'REL', 'musicbrainz_recording_id': 'REC'} + + +# ── apply handler ── + +def test_fix_library_retag_writes_each_track(tmp_path, monkeypatch): + import core.repair_worker as rw + track = tmp_path / 'a.flac'; track.write_bytes(b'') + + worker = rw.RepairWorker.__new__(rw.RepairWorker) + worker.db = SimpleNamespace() + worker._config_manager = SimpleNamespace(get=lambda k, d=None: d) + worker.transfer_folder = str(tmp_path) + + monkeypatch.setattr(rw, '_resolve_file_path', lambda p, *a, **k: p) + writes = [] + monkeypatch.setattr('core.tag_writer.write_tags_to_file', + lambda fp, db_data, **k: writes.append((fp, db_data)) or {'success': True}) + + details = { + 'tracks': [{'file_path': str(track), 'db_data': {'title': 'Real Title', 'spotify_track_id': 'sp_trk'}}], + 'cover_action': None, 'cover_url': None, + } + res = worker._fix_library_retag('album', '1', None, details) + assert res['success'] is True + assert res['written'] == 1 + assert writes[0][1]['title'] == 'Real Title' + assert writes[0][1]['spotify_track_id'] == 'sp_trk' + + +def test_fix_library_retag_counts_unreachable(tmp_path, monkeypatch): + import core.repair_worker as rw + worker = rw.RepairWorker.__new__(rw.RepairWorker) + worker.db = SimpleNamespace() + worker._config_manager = SimpleNamespace(get=lambda k, d=None: d) + worker.transfer_folder = str(tmp_path) + monkeypatch.setattr(rw, '_resolve_file_path', lambda p, *a, **k: p) + monkeypatch.setattr('core.tag_writer.write_tags_to_file', lambda *a, **k: {'success': True}) + + details = {'tracks': [{'file_path': str(tmp_path / 'missing.flac'), 'db_data': {'title': 'x'}}], + 'cover_action': None, 'cover_url': None} + res = worker._fix_library_retag('album', '1', None, details) + assert res['success'] is False # nothing written (file missing)