soulsync/tests/blocklist/test_blocklist_db.py

"""Blocklist DB layer + the add_to_wishlist enforcement guard.

Verified against a real SQLite DB: CRUD + profile scoping + dedup + backfill
update, the discovery→blocklist migration, and the end-to-end wishlist guard
(a banned artist's track is refused; cascade; profile isolation).
"""

from __future__ import annotations

import pytest

from database.music_database import MusicDatabase


@pytest.fixture()
def db(tmp_path):
    return MusicDatabase(str(tmp_path / "m.db"))


def _track(track_id, name, artist_id, artist_name, album_id="al0", album_name="Al", source="spotify"):
    return {
        "id": track_id, "name": name, "source": source,
        "artists": [{"id": artist_id, "name": artist_name}],
        "album": {"id": album_id, "name": album_name, "artists": [{"id": artist_id, "name": artist_name}]},
    }


# ── CRUD + scoping ───────────────────────────────────────────────────────────

def test_add_list_remove(db):
    eid = db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
    assert eid
    rows = db.get_blocklist(1)
    assert len(rows) == 1 and rows[0]["name"] == "Drake"
    assert db.remove_blocklist_entry(1, eid) is True
    assert db.get_blocklist(1) == []


def test_dedup_by_id_and_name(db):
    a = db.add_blocklist_entry(1, "artist", "Drake", spotify_id="sp")
    b = db.add_blocklist_entry(1, "artist", "Drake", spotify_id="sp")   # same id
    c = db.add_blocklist_entry(1, "artist", "drake")                     # same name (NOCASE)
    assert a == b == c
    assert len(db.get_blocklist(1)) == 1


def test_profile_isolation(db):
    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="sp")
    assert len(db.get_blocklist(1)) == 1
    assert db.get_blocklist(2) == []
    # remove is profile-scoped — profile 2 can't delete profile 1's row
    eid = db.get_blocklist(1)[0]["id"]
    assert db.remove_blocklist_entry(2, eid) is False
    assert len(db.get_blocklist(1)) == 1


def test_match_status_pending_vs_matched(db):
    one = db.add_blocklist_entry(1, "artist", "A", spotify_id="sp")
    two = db.add_blocklist_entry(1, "artist", "B", spotify_id="sp2", deezer_id="dz2")
    rows = {r["id"]: r for r in db.get_blocklist(1)}
    assert rows[one]["match_status"] == "pending"   # single id → needs backfill
    assert rows[two]["match_status"] == "matched"   # 2+ ids → already cross-known


def test_backfill_update_fills_only_nulls(db):
    eid = db.add_blocklist_entry(1, "artist", "A", spotify_id="sp")
    db.update_blocklist_entry_ids(eid, deezer_id="dz", spotify_id="SHOULD-NOT-OVERWRITE")
    row = db.get_blocklist(1)[0]
    assert row["spotify_id"] == "sp"        # COALESCE: existing kept
    assert row["deezer_id"] == "dz"
    assert row["match_status"] == "matched"
    assert db.get_blocklist_entries_needing_backfill() == []


# ── the wishlist guard, end to end ───────────────────────────────────────────

def test_blocked_artist_track_refused_from_wishlist(db):
    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
    ok = db.add_to_wishlist(
        spotify_track_data=_track("t1", "God's Plan", "drake-sp", "Drake"),
        profile_id=1)
    assert ok is False                       # refused
    # And nothing landed in the wishlist.
    assert db.get_wishlist_count(profile_id=1) == 0


def test_unblocked_artist_track_is_added(db):
    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
    ok = db.add_to_wishlist(
        spotify_track_data=_track("t2", "Hello", "adele-sp", "Adele"),
        profile_id=1)
    assert ok is True
    assert db.get_wishlist_count(profile_id=1) == 1


def test_block_is_profile_scoped_at_the_guard(db):
    db.add_blocklist_entry(1, "artist", "Drake", spotify_id="drake-sp")
    # Profile 2 has no such ban → the same track is allowed.
    ok = db.add_to_wishlist(
        spotify_track_data=_track("t3", "Nice For What", "drake-sp", "Drake"),
        profile_id=2)
    assert ok is True


def test_album_block_cascades_to_track_at_guard(db):
    db.add_blocklist_entry(1, "album", "Scorpion", spotify_id="scorp-sp")
    ok = db.add_to_wishlist(
        spotify_track_data=_track("t4", "Nonstop", "drake-sp", "Drake",
                                  album_id="scorp-sp", album_name="Scorpion"),
        profile_id=1)
    assert ok is False


def test_discovery_blacklist_migrated_into_blocklist(tmp_path):
    import database.music_database as mdb

    def _reinit(path):
        # An app upgrade = a fresh process with an empty init memo. Clear the
        # per-process "already initialized" set so init (and the migration)
        # actually re-runs against the existing DB file.
        mdb._database_initialized_paths.discard(str(mdb.Path(path).resolve()))
        return MusicDatabase(path)

    path = str(tmp_path / "mig.db")
    db = MusicDatabase(path)
    conn = db._get_connection()
    cur = conn.cursor()
    cur.execute("INSERT INTO discovery_artist_blacklist (artist_name, spotify_artist_id) "
                "VALUES ('Nickelback', 'nb-sp')")
    conn.commit()
    conn.close()

    db2 = _reinit(path)          # upgrade → migration runs (committed)
    rows = db2.get_blocklist(1)
    assert any(r["name"] == "Nickelback" and r["entity_type"] == "artist" for r in rows)

    db3 = _reinit(path)          # idempotent — a second upgrade doesn't duplicate
    nb = [r for r in db3.get_blocklist(1) if r["name"] == "Nickelback"]
    assert len(nb) == 1

    # The migrated ban actually enforces at the wishlist guard.
    ok = db3.add_to_wishlist(
        spotify_track_data=_track("t5", "Photograph", "nb-sp", "Nickelback"),
        profile_id=1)
    assert ok is False