soulsync/tests/metadata/test_metadata_gap_filler.py

import sqlite3
import sys
import types
from types import SimpleNamespace

# Stub optional Spotify dependency so the metadata package can import in tests.
if 'spotipy' not in sys.modules:
    spotipy = types.ModuleType('spotipy')
    oauth2 = types.ModuleType('spotipy.oauth2')

    class _DummySpotify:
        pass

    class _DummyOAuth:
        pass

    spotipy.Spotify = _DummySpotify
    oauth2.SpotifyOAuth = _DummyOAuth
    oauth2.SpotifyClientCredentials = _DummyOAuth
    spotipy.oauth2 = oauth2
    sys.modules['spotipy'] = spotipy
    sys.modules['spotipy.oauth2'] = oauth2

if 'config.settings' not in sys.modules:
    config_mod = types.ModuleType('config')
    settings_mod = types.ModuleType('config.settings')

    class _DummyConfigManager:
        def get(self, key, default=None):
            return default

        def get_active_media_server(self):
            return "plex"

    settings_mod.config_manager = _DummyConfigManager()
    config_mod.settings = settings_mod
    sys.modules['config'] = config_mod
    sys.modules['config.settings'] = settings_mod

from core.repair_jobs import metadata_gap_filler as mgf


class _FakeTrackClient:
    def __init__(self, source_name, isrc=None):
        self.source_name = source_name
        self.isrc = isrc
        self.calls = []

    def get_track_details(self, track_id):
        self.calls.append(track_id)
        if self.isrc is None:
            return None
        return {
            'id': track_id,
            'external_ids': {'isrc': self.isrc},
        }


class _FakeMBClient:
    def __init__(self):
        self.calls = []

    def search_recording(self, title, artist_name=None, limit=1):
        self.calls.append((title, artist_name, limit))
        return [{'id': 'mb-recording'}]


def _make_db():
    conn = sqlite3.connect(':memory:')
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    cursor.execute(
        """
        CREATE TABLE artists (
            id INTEGER PRIMARY KEY,
            name TEXT,
            thumb_url TEXT
        )
        """
    )
    cursor.execute(
        """
        CREATE TABLE albums (
            id INTEGER PRIMARY KEY,
            title TEXT,
            thumb_url TEXT
        )
        """
    )
    cursor.execute(
        """
        CREATE TABLE tracks (
            id INTEGER PRIMARY KEY,
            title TEXT,
            artist_id INTEGER,
            album_id INTEGER,
            spotify_track_id TEXT,
            itunes_track_id TEXT,
            deezer_track_id TEXT,
            isrc TEXT,
            musicbrainz_recording_id TEXT
        )
        """
    )
    cursor.execute("INSERT INTO artists (id, name, thumb_url) VALUES (1, 'Artist', '')")
    cursor.execute("INSERT INTO albums (id, title, thumb_url) VALUES (1, 'Album', '')")
    cursor.execute(
        """
        INSERT INTO tracks
            (id, title, artist_id, album_id, spotify_track_id, itunes_track_id, deezer_track_id, isrc, musicbrainz_recording_id)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (1, 'Track Title', 1, 1, 'sp-1', None, 'dz-1', '', ''),
    )
    conn.commit()
    return conn


def _make_context(conn):
    findings = []
    return SimpleNamespace(
        db=SimpleNamespace(_get_connection=lambda: conn),
        config_manager=SimpleNamespace(get=lambda key, default=None: default),
        check_stop=lambda: False,
        wait_if_paused=lambda: False,
        update_progress=lambda *args, **kwargs: None,
        report_progress=lambda *args, **kwargs: None,
        sleep_or_stop=lambda seconds: False,
        mb_client=_FakeMBClient(),
        create_finding=lambda **kwargs: (findings.append(kwargs) or True),
        findings=findings,
    )


def test_metadata_gap_filler_prefers_primary_track_source(monkeypatch):
    conn = _make_db()
    context = _make_context(conn)

    spotify_client = _FakeTrackClient('spotify', isrc='SP-ISRC')
    deezer_client = _FakeTrackClient('deezer', isrc='DZ-ISRC')
    itunes_client = _FakeTrackClient('itunes', isrc=None)

    monkeypatch.setattr(mgf, 'get_primary_source', lambda: 'deezer')
    monkeypatch.setattr(
        mgf,
        'get_client_for_source',
        lambda source: {'spotify': spotify_client, 'deezer': deezer_client, 'itunes': itunes_client}.get(source),
    )

    result = mgf.MetadataGapFillerJob().scan(context)

    assert result.findings_created == 1
    assert deezer_client.calls == ['dz-1']
    assert spotify_client.calls == []
    assert context.findings[0]['details']['found_fields']['isrc'] == 'DZ-ISRC'
    assert context.findings[0]['details']['resolved_source'] == 'deezer'
    assert context.findings[0]['details']['resolved_track_id'] == 'dz-1'


def test_metadata_gap_filler_skips_track_detail_lookup_when_isrc_disabled(monkeypatch):
    conn = _make_db()
    context = _make_context(conn)

    spotify_client = _FakeTrackClient('spotify', isrc='SP-ISRC')
    deezer_client = _FakeTrackClient('deezer', isrc='DZ-ISRC')

    monkeypatch.setattr(mgf, 'get_primary_source', lambda: 'deezer')
    monkeypatch.setattr(
        mgf,
        'get_client_for_source',
        lambda source: {'spotify': spotify_client, 'deezer': deezer_client}.get(source),
    )

    job = mgf.MetadataGapFillerJob()
    monkeypatch.setattr(job, '_get_settings', lambda context: {'fill_isrc': False, 'fill_musicbrainz_id': True})

    result = job.scan(context)

    assert result.findings_created == 1
    assert spotify_client.calls == []
    assert deezer_client.calls == []
    assert context.findings[0]['details']['found_fields']['musicbrainz_recording_id'] == 'mb-recording'