soulsync/tests/test_personalized_playlists_id_gate.py

"""Tests for the lifted PersonalizedPlaylistsService selectors and the
mandatory ID-validity gate every section must enforce.

Context: discover-page selection methods used to return tracks/albums
with all source IDs NULL — the UI displayed them, the user clicked
download, the download silently failed because there was nothing to
look up. This test file pins the gate at the helper level so a future
section can't accidentally bypass it.

Coverage:
- `_select_discovery_tracks` filters out rows where every source ID is NULL
- `_select_discovery_tracks` honors source filter + blacklist filter
- `_apply_diversity_filter` caps per-album + per-artist counts
- `_compute_adaptive_diversity_limits` returns the right tier for the
  unique-artist count + relaxed flag
- The 5 discovery_pool methods (decade / genre / popular_picks /
  hidden_gems / discovery_shuffle) each filter NULL-id rows
"""

from __future__ import annotations

import sqlite3
from contextlib import contextmanager
from unittest.mock import patch

import pytest

from core.personalized_playlists import PersonalizedPlaylistsService


# ---------------------------------------------------------------------------
# Fake DB
# ---------------------------------------------------------------------------


class _FakeDatabase:
    """Wraps an in-memory sqlite connection so the service's
    `database._get_connection()` calls work the same as in production.

    The schema mirrors the real `discovery_pool` + `tracks` + `albums`
    + `artists` shape just enough for the selection methods to exercise.
    """

    def __init__(self):
        self._conn = sqlite3.connect(":memory:")
        self._conn.row_factory = sqlite3.Row
        self._setup_schema()

    def _setup_schema(self):
        cursor = self._conn.cursor()
        cursor.executescript("""
            CREATE TABLE discovery_pool (
                id INTEGER PRIMARY KEY,
                source TEXT NOT NULL,
                spotify_track_id TEXT,
                itunes_track_id TEXT,
                deezer_track_id TEXT,
                track_name TEXT,
                artist_name TEXT,
                album_name TEXT,
                album_cover_url TEXT,
                duration_ms INTEGER,
                popularity INTEGER,
                release_date TEXT,
                artist_genres TEXT,
                track_data_json TEXT
            );
            CREATE TABLE discovery_artist_blacklist (
                artist_name TEXT PRIMARY KEY
            );
        """)
        self._conn.commit()

    @contextmanager
    def _get_connection(self):
        # Match the production interface (`with database._get_connection() as conn`)
        try:
            yield self._conn
        finally:
            pass

    def insert_discovery_track(self, **kwargs):
        cols = ", ".join(kwargs.keys())
        placeholders = ", ".join(["?"] * len(kwargs))
        self._conn.execute(
            f"INSERT INTO discovery_pool ({cols}) VALUES ({placeholders})",
            tuple(kwargs.values()),
        )
        self._conn.commit()

    def blacklist(self, artist_name):
        self._conn.execute(
            "INSERT INTO discovery_artist_blacklist (artist_name) VALUES (?)",
            (artist_name,),
        )
        self._conn.commit()


@pytest.fixture
def service():
    """Service with a fresh in-memory DB. `_get_active_source` patched
    to return 'spotify' so every selector targets the same source."""
    db = _FakeDatabase()
    svc = PersonalizedPlaylistsService(db)
    with patch.object(svc, '_get_active_source', return_value='spotify'):
        yield svc, db


# ---------------------------------------------------------------------------
# `_select_discovery_tracks` — the helper everyone goes through
# ---------------------------------------------------------------------------


def test_discovery_helper_filters_null_id_rows(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='Has Spotify ID',
        artist_name='A', album_name='A', popularity=50,
    )
    db.insert_discovery_track(
        source='spotify', spotify_track_id=None, itunes_track_id=None,
        track_name='No IDs', artist_name='B', album_name='B', popularity=50,
    )
    db.insert_discovery_track(
        source='spotify', itunes_track_id='it1', track_name='Has iTunes ID',
        artist_name='C', album_name='C', popularity=50,
    )

    tracks = svc._select_discovery_tracks(
        source='spotify',
        order_by='track_name',
        fetch_limit=100,
    )
    names = sorted(t['track_name'] for t in tracks)
    assert names == ['Has Spotify ID', 'Has iTunes ID']


def test_discovery_helper_accepts_deezer_only_id_rows(service):
    """Discovery pool rows with NULL spotify + NULL itunes but a populated
    deezer_track_id MUST pass the gate. Regression test — early version
    of the gate only checked Spotify + iTunes, which silently filtered
    out every row for Deezer-primary users (entire Time Machine /
    Genre / Hidden Gems / Shuffle / Popular Picks rendered empty)."""
    svc, db = service
    db.insert_discovery_track(
        source='deezer', deezer_track_id='dz1',
        spotify_track_id=None, itunes_track_id=None,
        track_name='Deezer Only', artist_name='A', album_name='X',
        popularity=50, release_date='2024-01-01',
    )
    with patch.object(svc, '_get_active_source', return_value='deezer'):
        tracks = svc._select_discovery_tracks(
            source='deezer',
            order_by='track_name',
            fetch_limit=100,
        )
    assert [t['track_name'] for t in tracks] == ['Deezer Only']
    assert tracks[0]['deezer_track_id'] == 'dz1'


def test_discovery_helper_filters_blacklisted_artists(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='Keep',
        artist_name='Good Artist', album_name='X',
    )
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp2', track_name='Drop',
        artist_name='Bad Artist', album_name='X',
    )
    db.blacklist('bad artist')

    tracks = svc._select_discovery_tracks(
        source='spotify',
        order_by='track_name',
        fetch_limit=100,
    )
    assert [t['track_name'] for t in tracks] == ['Keep']


def test_discovery_helper_honors_source_filter(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='SP',
        artist_name='A', album_name='X',
    )
    db.insert_discovery_track(
        source='itunes', itunes_track_id='it1', track_name='IT',
        artist_name='A', album_name='X',
    )

    tracks = svc._select_discovery_tracks(
        source='spotify',
        order_by='track_name',
        fetch_limit=100,
    )
    assert [t['track_name'] for t in tracks] == ['SP']


def test_discovery_helper_honors_extra_where(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='Pop60',
        artist_name='A', album_name='X', popularity=60,
    )
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp2', track_name='Pop20',
        artist_name='A', album_name='X', popularity=20,
    )

    tracks = svc._select_discovery_tracks(
        source='spotify',
        extra_where='AND popularity >= 50',
        order_by='popularity DESC',
        fetch_limit=100,
    )
    assert [t['track_name'] for t in tracks] == ['Pop60']


# ---------------------------------------------------------------------------
# Diversity filter
# ---------------------------------------------------------------------------


def test_diversity_filter_caps_per_album():
    svc = PersonalizedPlaylistsService(_FakeDatabase())
    tracks = [
        {'track_name': f't{i}', 'artist_name': 'A', 'album_name': 'Album1'}
        for i in range(10)
    ]
    out = svc._apply_diversity_filter(
        tracks, max_per_album=3, max_per_artist=10, limit=10,
    )
    assert len(out) == 3


def test_diversity_filter_caps_per_artist():
    svc = PersonalizedPlaylistsService(_FakeDatabase())
    tracks = [
        {'track_name': f't{i}', 'artist_name': 'OnlyArtist', 'album_name': f'Album{i}'}
        for i in range(10)
    ]
    out = svc._apply_diversity_filter(
        tracks, max_per_album=10, max_per_artist=2, limit=10,
    )
    assert len(out) == 2


def test_diversity_filter_stops_at_limit():
    svc = PersonalizedPlaylistsService(_FakeDatabase())
    tracks = [
        {'track_name': f't{i}', 'artist_name': f'a{i}', 'album_name': f'b{i}'}
        for i in range(20)
    ]
    out = svc._apply_diversity_filter(
        tracks, max_per_album=10, max_per_artist=10, limit=5,
    )
    assert len(out) == 5


# ---------------------------------------------------------------------------
# Adaptive diversity limits
# ---------------------------------------------------------------------------


def test_adaptive_limits_high_variety():
    svc = PersonalizedPlaylistsService(_FakeDatabase())
    tracks = [{'artist_name': f'a{i}'} for i in range(30)]
    max_album, max_artist = svc._compute_adaptive_diversity_limits(tracks)
    # High variety tier — strict limits
    assert (max_album, max_artist) == (3, 5)


def test_adaptive_limits_low_variety():
    svc = PersonalizedPlaylistsService(_FakeDatabase())
    tracks = [{'artist_name': f'a{i % 3}'} for i in range(30)]  # only 3 unique artists
    max_album, max_artist = svc._compute_adaptive_diversity_limits(tracks)
    # Low variety tier — much more lenient (matches existing decade-style limits)
    assert max_album >= 4
    assert max_artist >= 8


def test_adaptive_limits_relaxed_flag_loosens_genre_tier():
    svc = PersonalizedPlaylistsService(_FakeDatabase())
    # 15 unique artists triggers the moderate tier
    tracks = [{'artist_name': f'a{i}'} for i in range(15)]
    strict = svc._compute_adaptive_diversity_limits(tracks, relaxed=False)
    relaxed = svc._compute_adaptive_diversity_limits(tracks, relaxed=True)
    assert relaxed[1] >= strict[1]


# ---------------------------------------------------------------------------
# Public methods enforce the gate (smoke test on each)
# ---------------------------------------------------------------------------


def test_get_hidden_gems_filters_null_id_rows(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='Gem',
        artist_name='A', album_name='X', popularity=20,
    )
    db.insert_discovery_track(
        source='spotify', spotify_track_id=None, itunes_track_id=None,
        track_name='Nogem', artist_name='A', album_name='X', popularity=20,
    )
    out = svc.get_hidden_gems(limit=10)
    assert [t['track_name'] for t in out] == ['Gem']


def test_get_discovery_shuffle_filters_null_id_rows(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='Yes',
        artist_name='A', album_name='X',
    )
    db.insert_discovery_track(
        source='spotify', spotify_track_id=None, itunes_track_id=None,
        track_name='No', artist_name='A', album_name='X',
    )
    out = svc.get_discovery_shuffle(limit=10)
    assert [t['track_name'] for t in out] == ['Yes']


def test_get_popular_picks_filters_null_id_rows(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='Yes',
        artist_name='A', album_name='X', popularity=80,
    )
    db.insert_discovery_track(
        source='spotify', spotify_track_id=None, itunes_track_id=None,
        track_name='No', artist_name='A', album_name='X', popularity=80,
    )
    out = svc.get_popular_picks(limit=10)
    assert [t['track_name'] for t in out] == ['Yes']


def test_get_decade_playlist_filters_null_id_rows(service):
    svc, db = service
    db.insert_discovery_track(
        source='spotify', spotify_track_id='sp1', track_name='Yes',
        artist_name='A', album_name='X', release_date='2024-06-01',
    )
    db.insert_discovery_track(
        source='spotify', spotify_track_id=None, itunes_track_id=None,
        track_name='No', artist_name='A', album_name='X', release_date='2024-06-01',
    )
    out = svc.get_decade_playlist(2020, limit=10)
    assert [t['track_name'] for t in out] == ['Yes']