"""Export source wiring (#903): waterfall order + cache write-back. build_resolve_fn assembles cache -> DB -> file -> MusicBrainz and writes a fresh (non-cache) hit back to the cache. Pins: a cache hit short-circuits everything and is NOT re-written; a DB/MB hit IS written back; misses fall through; the resolving label is returned. """ from __future__ import annotations from core.exports.export_sources import build_resolve_fn from core.exports.mbid_resolver import SRC_CACHE, SRC_DB, SRC_MUSICBRAINZ MBID = "e8f9b188-f819-4e43-ab0f-4bd26ce9ff56" def _wire(db=None, file=None, mb=None, cache=None): recorded = {} store = dict(cache or {}) fn = build_resolve_fn( db_fn=lambda a, t: (db or {}).get((a, t)), file_fn=lambda a, t: (file or {}).get((a, t)), mb_fn=lambda a, t: (mb or {}).get((a, t)), cache_lookup=lambda k: store.get(k), cache_record=lambda k, m: recorded.__setitem__(k, m) or True, ) return fn, recorded def test_cache_hit_short_circuits_and_is_not_rewritten(): from core.exports.mbid_resolver import normalize_key fn, recorded = _wire( cache={normalize_key("A", "T"): MBID}, db={("A", "T"): "should-not-reach"}, ) mbid, label = fn("A", "T") assert (mbid, label) == (MBID, SRC_CACHE) assert recorded == {} # cache hit -> no write-back def test_db_hit_is_written_back_to_cache(): from core.exports.mbid_resolver import normalize_key fn, recorded = _wire(db={("A", "T"): MBID}) mbid, label = fn("A", "T") assert (mbid, label) == (MBID, SRC_DB) assert recorded == {normalize_key("A", "T"): MBID} # fresh hit cached for next time def test_falls_through_to_musicbrainz_and_caches(): fn, recorded = _wire(db={}, file={}, mb={("A", "T"): MBID}) mbid, label = fn("A", "T") assert (mbid, label) == (MBID, SRC_MUSICBRAINZ) assert list(recorded.values()) == [MBID] def test_all_miss_returns_none_and_no_write(): fn, recorded = _wire() assert fn("A", "T") == (None, None) assert recorded == {} # ── service track-id resolver (#945 export to Spotify/Deezer) ── from core.exports.export_sources import ( db_service_track_id, build_service_resolve_fn, _SERVICE_ID_COLUMNS, ) def test_service_id_column_mapping(): assert _SERVICE_ID_COLUMNS == {'spotify': 'spotify_track_id', 'deezer': 'deezer_id'} def test_db_service_track_id_unknown_service_is_none(): assert db_service_track_id('A', 'X', 'tidal') is None assert db_service_track_id('A', 'X', '') is None def test_db_service_track_id_no_title_is_none(): assert db_service_track_id('A', '', 'spotify') is None def test_build_service_resolve_fn_returns_id_and_source(monkeypatch): import core.exports.export_sources as es monkeypatch.setattr(es, 'db_service_track_id', lambda a, t, s: 'spid-99' if t == 'Hit' else None) fn = build_service_resolve_fn('spotify') assert fn('Artist', 'Hit') == ('spid-99', 'library') assert fn('Artist', 'Miss') == (None, None) def test_db_service_track_id_real_sql_executes(tmp_path, monkeypatch): """Run the ACTUAL query against a real (temp) tracks/artists schema — the broad except→None in db_service_track_id would otherwise mask a column/join typo as 'no match' for every track (#945 verification).""" import sqlite3 import types import core.exports.export_sources as es dbfile = tmp_path / "lib.db" con = sqlite3.connect(str(dbfile)) con.executescript( "CREATE TABLE artists (id TEXT PRIMARY KEY, name TEXT);" "CREATE TABLE tracks (id TEXT, artist_id TEXT, title TEXT, " "spotify_track_id TEXT, deezer_id TEXT);" "INSERT INTO artists VALUES ('a1','Kendrick Lamar');" "INSERT INTO tracks VALUES ('t1','a1','Not Like Us','spid-NLU','dz-NLU');" ) con.commit() con.close() # fresh connection per call (db_service_track_id closes it in finally) fake_db = types.SimpleNamespace(_get_connection=lambda: sqlite3.connect(str(dbfile))) monkeypatch.setattr("database.music_database.get_database", lambda: fake_db) assert es.db_service_track_id("Kendrick Lamar", "Not Like Us", "spotify") == "spid-NLU" assert es.db_service_track_id("kendrick lamar", "not like us", "deezer") == "dz-NLU" # case-insensitive assert es.db_service_track_id("Kendrick Lamar", "Unknown Song", "spotify") is None # ── discovery-cache resolution (#945: use the already-discovered IDs, no API call) ── import json as _json from core.exports.export_sources import ( service_id_from_extra_data, resolve_service_track_ids, ) def _extra(service, tid, discovered=True, provider=None): return {'extra_data': _json.dumps({'discovered': discovered, 'provider': provider or service, 'matched_data': {'id': tid}})} def test_extra_data_id_when_discovered_to_that_service(): assert service_id_from_extra_data(_extra('deezer', 111), 'deezer') == '111' # dict (not str) extra_data also works raw = {'extra_data': {'discovered': True, 'provider': 'spotify', 'matched_data': {'id': 'spX'}}} assert service_id_from_extra_data(raw, 'spotify') == 'spX' def test_extra_data_provider_must_match_service(): # discovered to Spotify, exporting to Deezer → don't reuse the (wrong-service) id assert service_id_from_extra_data(_extra('spotify', 111), 'deezer') is None def test_extra_data_wing_it_fallback_is_not_trusted(): track = _extra('deezer', 111, provider='wing_it_fallback') assert service_id_from_extra_data(track, 'deezer') is None def test_extra_data_misc_none_cases(): assert service_id_from_extra_data({}, 'deezer') is None # no extra_data assert service_id_from_extra_data({'extra_data': 'not json{'}, 'deezer') is None # bad json assert service_id_from_extra_data(_extra('deezer', 111, discovered=False), 'deezer') is None def test_resolve_waterfall_cache_then_library_then_unmatched(): tracks = [ _extra('deezer', 111) | {'artist_name': 'A', 'track_name': 'Cached'}, # cache hit {'artist_name': 'A', 'track_name': 'InLib'}, # library hit (db_fn) {'artist_name': 'A', 'track_name': 'Nowhere'}, # unmatched ] db_fn = lambda a, t, s: 'lib-222' if t == 'InLib' else None out = resolve_service_track_ids(tracks, 'deezer', db_fn=db_fn) ids = [r['service_track_id'] for r in out['resolved']] assert ids == ['111', 'lib-222', None] s = out['stats'] assert s == {'total': 3, 'resolved': 2, 'unmatched': 1, 'from_cache': 1, 'from_library': 1, 'from_search': 0} # ── backfill: confident live-search match for the un-cached/un-enriched tail (#945) ── from core.metadata.types import Track as _Track from core.exports.export_sources import search_service_track_id, BACKFILL_MIN_SCORE def _cand(name, artist, tid, album_type='album'): return _Track(id=tid, name=name, artists=[artist], album='A', duration_ms=200000, album_type=album_type) def test_backfill_exact_match_returned(): search = lambda q: [_cand('Not Like Us', 'Kendrick Lamar', 'dz-NLU')] assert search_service_track_id('Kendrick Lamar', 'Not Like Us', search_fn=search) == 'dz-NLU' def test_backfill_wrong_artist_rejected(): """SAFETY: an exact-title hit by the WRONG artist scores below the floor (no 1.5x exact- artist boost) → None, so backfill never adds someone else's same-named track.""" search = lambda q: [_cand('Not Like Us', 'Some Other Guy', 'wrong-id')] assert search_service_track_id('Kendrick Lamar', 'Not Like Us', search_fn=search) is None def test_backfill_karaoke_cover_rejected(): """SAFETY: a karaoke/cover version is buried (x0.05) below the floor → None.""" search = lambda q: [_cand('Not Like Us (Karaoke Version)', 'Karaoke All Stars', 'kar-id')] assert search_service_track_id('Kendrick Lamar', 'Not Like Us', search_fn=search) is None def test_backfill_picks_real_over_cover(): search = lambda q: [ _cand('Not Like Us (Karaoke Version)', 'Karaoke All Stars', 'kar-id'), _cand('Not Like Us', 'Kendrick Lamar', 'real-id'), ] assert search_service_track_id('Kendrick Lamar', 'Not Like Us', search_fn=search) == 'real-id' def test_backfill_empty_and_error_and_no_title(): assert search_service_track_id('A', 'X', search_fn=lambda q: []) is None def boom(q): raise RuntimeError('deezer flaked') assert search_service_track_id('A', 'X', search_fn=boom) is None # fail-safe assert search_service_track_id('A', '', search_fn=lambda q: [_cand('X', 'A', 'i')]) is None def test_resolve_waterfall_uses_search_only_when_cache_and_library_miss(): tracks = [ _extra('deezer', 111) | {'artist_name': 'A', 'track_name': 'Cached'}, {'artist_name': 'A', 'track_name': 'InLib'}, {'artist_name': 'A', 'track_name': 'OnlyOnSvc'}, ] db_fn = lambda a, t, s: 'lib-2' if t == 'InLib' else None search_id_fn = lambda a, t: 'srch-3' if t == 'OnlyOnSvc' else None out = resolve_service_track_ids(tracks, 'deezer', db_fn=db_fn, search_id_fn=search_id_fn) assert [r['service_track_id'] for r in out['resolved']] == ['111', 'lib-2', 'srch-3'] s = out['stats'] assert (s['from_cache'], s['from_library'], s['from_search'], s['unmatched']) == (1, 1, 1, 0) def test_resolve_no_search_fn_leaves_tail_unmatched(): out = resolve_service_track_ids([{'artist_name': 'A', 'track_name': 'X'}], 'deezer', db_fn=lambda a, t, s: None) # search_id_fn omitted assert out['resolved'][0]['service_track_id'] is None assert out['stats']['unmatched'] == 1 and out['stats']['from_search'] == 0