Library re-tag: seam tests for the job scan, apply handler, and source-id embed

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

@@ -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

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)