danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge pull request #741 from Nezreka/refactor/db-schema-hardening

Browse source 
Refactor/db schema hardening
This commit is contained in:
BoulderBadgeDad 2026-05-29 13:57:24 -07:00 committed by GitHub
commit 2d68843343
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 736 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
core/artist_source_lookup.py
View file

@ -21,6 +21,8 @@ from __future__ import annotations
import logging
from typing import Optional
from core.source_ids import id_column as _artist_id_column
logger = logging.getLogger("artist_source_lookup")
@ -29,14 +31,14 @@ SOURCE_ONLY_ARTIST_SOURCES = frozenset({
})
# The per-source column on the ``artists`` table, derived from the canonical
# source-ID registry (the single source of truth). Values are unchanged from the
# previous hardcoded map — this just stops duplicating that knowledge here.
SOURCE_ID_FIELD = {
"spotify": "spotify_artist_id",
"itunes": "itunes_artist_id",
"deezer": "deezer_id",
"discogs": "discogs_id",
"hydrabase": "soul_id",
"musicbrainz": "musicbrainz_id",
"amazon": "amazon_id",
source: _artist_id_column(source, "artist")
for source in (
"spotify", "itunes", "deezer", "discogs", "hydrabase", "musicbrainz", "amazon",
)
}

144
core/source_ids.py Normal file
View file

@ -0,0 +1,144 @@
"""Canonical registry of external/source ID column names.
SoulSync stores each metadata provider's ID for an artist/album/track under a
column whose NAME is inconsistent across tables — e.g. Deezer's artist id is
``deezer_id`` on the ``artists`` table but ``deezer_artist_id`` on
``watchlist_artists`` and ``album_deezer_id`` / ``similar_artist_deezer_id`` on
the discovery tables. Spotify/iTunes keep an entity qualifier on the core tables
while Deezer/Amazon/Tidal/... don't, and MusicBrainz uses three different nouns.
The result is code that checks 2–5 property-name variants everywhere.
This module is the single source of truth for "(provider, entity) → column".
It does NOT rename any database column — these ARE the real names today; the
registry just centralizes the knowledge and offers accessors that read an ID
from a dict / sqlite3.Row robustly (canonical column first, then known aliases),
so callers stop hand-rolling variant checks.
"""
from __future__ import annotations
from typing import Any, Dict, Iterable, Optional
# Entity types this registry knows about.
ENTITIES = ("artist", "album", "track")
# Canonical column name on the CORE table (artists / albums / tracks) for each
# (entity, provider). This is the name to prefer when reading/writing.
_CORE_ID_COLUMNS: Dict[str, Dict[str, str]] = {
"artist": {
"spotify": "spotify_artist_id",
"itunes": "itunes_artist_id",
"deezer": "deezer_id",
"musicbrainz": "musicbrainz_id",
"discogs": "discogs_id",
"amazon": "amazon_id",
"tidal": "tidal_id",
"qobuz": "qobuz_id",
"audiodb": "audiodb_id",
"genius": "genius_id",
"hydrabase": "soul_id",
},
"album": {
"spotify": "spotify_album_id",
"itunes": "itunes_album_id",
"deezer": "deezer_id",
"musicbrainz": "musicbrainz_release_id",
"discogs": "discogs_id",
"amazon": "amazon_id",
"tidal": "tidal_id",
"qobuz": "qobuz_id",
"audiodb": "audiodb_id",
"hydrabase": "soul_id",
},
"track": {
"spotify": "spotify_track_id",
"itunes": "itunes_track_id",
"deezer": "deezer_id",
"musicbrainz": "musicbrainz_recording_id",
"amazon": "amazon_id",
"tidal": "tidal_id",
"qobuz": "qobuz_id",
"audiodb": "audiodb_id",
"genius": "genius_id",
"hydrabase": "soul_id",
},
}
# Other column / dict-key names the SAME (entity, provider) ID appears under
# elsewhere (satellite tables, API payloads). Accessors check the canonical
# column first, then these, so a read works regardless of where the row/dict
# came from. Keyed by (entity, provider).
_ALIASES: Dict[tuple, tuple] = {
("artist", "spotify"): ("similar_artist_spotify_id",),
("artist", "itunes"): ("artist_itunes_id", "similar_artist_itunes_id"),
("artist", "deezer"): ("deezer_artist_id", "artist_deezer_id", "similar_artist_deezer_id"),
("artist", "musicbrainz"): ("musicbrainz_artist_id", "similar_artist_musicbrainz_id"),
("artist", "discogs"): ("discogs_artist_id",),
("artist", "amazon"): ("amazon_artist_id",),
("album", "spotify"): ("album_spotify_id",),
("album", "itunes"): ("album_itunes_id",),
("album", "deezer"): ("deezer_album_id", "album_deezer_id"),
("album", "discogs"): ("discogs_release_id",),
("track", "deezer"): ("deezer_track_id",),
}
def id_column(provider: str, entity: str = "artist") -> Optional[str]:
"""Canonical core-table column for this provider + entity, or None if the
provider isn't tracked for that entity."""
return _CORE_ID_COLUMNS.get(entity, {}).get(provider)
def id_keys(provider: str, entity: str = "artist") -> tuple:
"""All known key names (canonical first, then aliases) the ID may live
under. Useful for code that needs the full variant list explicitly."""
keys = []
canon = id_column(provider, entity)
if canon:
keys.append(canon)
for alias in _ALIASES.get((entity, provider), ()): # preserve order, no dups
if alias not in keys:
keys.append(alias)
return tuple(keys)
def _read(data: Any, key: str) -> Any:
"""Read ``key`` from a dict or sqlite3.Row, returning None if absent."""
try:
keys = data.keys() # dict and sqlite3.Row both support .keys()
except AttributeError:
return None
if key in keys:
try:
return data[key]
except (KeyError, IndexError):
return None
return None
def get_id(data: Any, provider: str, entity: str = "artist") -> Optional[str]:
"""Read this provider's ID for ``entity`` from a dict / sqlite3.Row.
Tries the canonical column first, then every known alias, and returns the
first non-empty value (or None). Replaces hand-rolled
``row.get('deezer_id') or row.get('deezer_artist_id')`` chains.
"""
for key in id_keys(provider, entity):
value = _read(data, key)
if value:
return value
return None
def source_id_map(
data: Any,
entity: str = "artist",
providers: Optional[Iterable[str]] = None,
) -> Dict[str, Optional[str]]:
"""Build a ``{provider: id}`` dict for ``entity`` from a row/dict — the
common "artist_source_ids" pattern. Defaults to every provider known for the
entity; pass ``providers`` to restrict/order the result.
"""
if providers is None:
providers = list(_CORE_ID_COLUMNS.get(entity, {}).keys())
return {p: get_id(data, p, entity) for p in providers}

143
database/music_database.py
View file

@ -312,6 +312,19 @@ class MusicDatabase:
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Migration ledger — a single, readable record of which one-time
# migrations have run. ADDITIVE backstop only: existing migrations
# keep their own idempotency gates (PRAGMA checks, marker tables,
# metadata flags); this table just unifies that scattered state so a
# half-migrated DB is detectable. Nothing GATES on it. Paired with
# PRAGMA user_version (set at the end of init).
cursor.execute("""
CREATE TABLE IF NOT EXISTS schema_migrations (
name TEXT PRIMARY KEY,
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Wishlist table for storing failed download tracks for retry
cursor.execute("""
@ -337,6 +350,7 @@ class MusicDatabase:
deezer_artist_id TEXT,
discogs_artist_id TEXT,
musicbrainz_artist_id TEXT,
amazon_artist_id TEXT,
artist_name TEXT NOT NULL,
date_added TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_scan_timestamp TIMESTAMP,
@ -792,6 +806,10 @@ class MusicDatabase:
logger.error(f"Personalized-playlist schema init failed: {ps_err}")
self._ensure_core_media_schema_columns(cursor)
self._normalize_genres_to_json(cursor)
# Unify scattered migration state into the ledger + stamp the schema
# version. Additive backstop — runs last, gates nothing.
self._sync_migration_ledger(cursor)
conn.commit()
logger.info("Database initialized successfully")
@ -802,6 +820,122 @@ class MusicDatabase:
self._init_manual_library_match_table()
# Bump when the schema's generation meaningfully changes. Stamped into
# PRAGMA user_version as a backstop indicator; nothing GATES on it yet.
SCHEMA_VERSION = 1
# Maps a ledger name to the EXISTING idempotency signal that proves a
# one-time migration ran: ('table', <marker table>) or ('flag', <metadata
# key>). Used to back-fill the ledger for DBs created before it existed.
# The ledger is a non-gating backstop, so this can grow lazily — a missing
# entry just means that migration isn't surfaced in the ledger (harmless).
_KNOWN_MIGRATION_SIGNALS = {
'id_columns_to_text': ('flag', 'id_columns_migrated'),
'genres_json': ('flag', 'genres_json_normalized'),
'metadata_cache_v1': ('flag', 'metadata_cache_v1'),
'repair_worker_v2': ('flag', 'repair_worker_v2'),
'spotify_library_cache_v1': ('flag', 'spotify_library_cache_v1'),
'profiles_v1': ('flag', 'profiles_migration_v1'),
'profiles_v2': ('flag', 'profiles_migration_v2'),
'profiles_v3': ('flag', 'profiles_migration_v3'),
'profiles_v4': ('flag', 'profiles_migration_v4'),
'discovery_cache_v2': ('table', '_discovery_cache_v2_migrated'),
'deezer_cache_v2': ('table', '_deezer_cache_v2_migrated'),
'cache_junk_artist_purged': ('table', '_cache_junk_artist_purged'),
'genius_search_fix': ('table', '_genius_search_fix_applied'),
}
def _record_migration(self, cursor, name):
"""Record a one-time migration in the schema_migrations ledger.
Idempotent (INSERT OR IGNORE). New one-time migrations should call this
when they complete; the ledger is a non-gating backstop, so a failure to
record never affects correctness.
"""
try:
cursor.execute(
"INSERT OR IGNORE INTO schema_migrations (name, applied_at) "
"VALUES (?, CURRENT_TIMESTAMP)", (name,)
)
except Exception as e:
logger.debug("Could not record migration %s in ledger: %s", name, e)
def _sync_migration_ledger(self, cursor):
"""Back-fill the ledger from existing idempotency signals and stamp
PRAGMA user_version.
ADDITIVE + non-gating: this only RECORDS state that already exists (which
marker tables / metadata flags are present); it never decides whether a
migration runs. Safe to run on every startup.
"""
try:
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = {r[0] for r in cursor.fetchall()}
for ledger_name, (kind, signal) in self._KNOWN_MIGRATION_SIGNALS.items():
if kind == 'table':
present = signal in tables
else: # 'flag' — a metadata row
cursor.execute("SELECT 1 FROM metadata WHERE key = ? LIMIT 1", (signal,))
present = cursor.fetchone() is not None
if present:
self._record_migration(cursor, ledger_name)
# Backstop version stamp; nothing gates on it.
cursor.execute(f"PRAGMA user_version = {int(self.SCHEMA_VERSION)}")
except Exception as e:
logger.error(f"Error syncing migration ledger: {e}")
def _normalize_genres_to_json(self, cursor):
"""One-time: rewrite legacy comma-separated genres to canonical JSON arrays.
``artists.genres`` / ``albums.genres`` historically stored EITHER a JSON
array (new writes) OR a comma-separated string (old writes), so every
reader has to try-JSON-then-split. This normalizes existing rows to JSON
in place. It mirrors the readers' exact parse (JSON list, else
comma-split/strip/drop-empties), so the genre VALUES are unchanged — only
the storage format. Marker-gated to run once, and per-row diffed so a
re-run (or a row already in JSON form) is a no-op. Non-fatal on error,
like the other migrations.
"""
import json
try:
cursor.execute("SELECT value FROM metadata WHERE key = 'genres_json_normalized' LIMIT 1")
if cursor.fetchone():
return
def _to_list(raw):
# Identical semantics to the genres readers elsewhere in this file.
try:
parsed = json.loads(raw)
return parsed if isinstance(parsed, list) else [str(parsed)]
except (json.JSONDecodeError, ValueError, TypeError):
return [g.strip() for g in raw.split(',') if g.strip()]
total = 0
for table in ('artists', 'albums'):
cursor.execute(
f"SELECT id, genres FROM {table} "
f"WHERE genres IS NOT NULL AND TRIM(genres) != ''"
)
pending = []
for row in cursor.fetchall():
rid, raw = row[0], row[1]
canonical = json.dumps(_to_list(raw))
if canonical != raw: # leave already-canonical rows untouched
pending.append((canonical, rid))
for canonical, rid in pending:
cursor.execute(f"UPDATE {table} SET genres = ? WHERE id = ?", (canonical, rid))
total += 1
cursor.execute(
"INSERT OR REPLACE INTO metadata (key, value, updated_at) "
"VALUES ('genres_json_normalized', 'true', CURRENT_TIMESTAMP)"
)
self._record_migration(cursor, 'genres_json')
if total:
logger.info("Normalized %d legacy genres value(s) to JSON", total)
except Exception as e:
logger.error(f"Error normalizing genres to JSON: {e}")
def _ensure_core_media_schema_columns(self, cursor):
"""Repair required media-library columns that older migrations may miss.
@ -1826,6 +1960,7 @@ class MusicDatabase:
deezer_artist_id TEXT,
discogs_artist_id TEXT,
musicbrainz_artist_id TEXT,
amazon_artist_id TEXT,
profile_id INTEGER DEFAULT 1,
UNIQUE(profile_id, spotify_artist_id),
UNIQUE(profile_id, itunes_artist_id)
@ -1854,7 +1989,8 @@ class MusicDatabase:
itunes_artist_id TEXT,
deezer_artist_id TEXT,
discogs_artist_id TEXT,
musicbrainz_artist_id TEXT
musicbrainz_artist_id TEXT,
amazon_artist_id TEXT
)
""")
@ -1867,7 +2003,7 @@ class MusicDatabase:
'include_remixes', 'include_acoustic', 'include_compilations',
'include_instrumentals', 'lookback_days',
'itunes_artist_id', 'deezer_artist_id', 'discogs_artist_id',
'musicbrainz_artist_id', 'profile_id']
'musicbrainz_artist_id', 'amazon_artist_id', 'profile_id']
shared_cols = [c for c in new_cols if c in old_cols]
cols_str = ', '.join(shared_cols)
cursor.execute(f"INSERT INTO watchlist_artists_new ({cols_str}) SELECT {cols_str} FROM watchlist_artists")
@ -2711,6 +2847,7 @@ class MusicDatabase:
deezer_artist_id TEXT,
discogs_artist_id TEXT,
musicbrainz_artist_id TEXT,
amazon_artist_id TEXT,
profile_id INTEGER DEFAULT 1,
UNIQUE(profile_id, spotify_artist_id),
UNIQUE(profile_id, itunes_artist_id)
@ -2724,7 +2861,7 @@ class MusicDatabase:
'include_remixes', 'include_acoustic', 'include_compilations',
'include_instrumentals', 'lookback_days',
'itunes_artist_id', 'deezer_artist_id', 'discogs_artist_id',
'musicbrainz_artist_id', 'profile_id']
'musicbrainz_artist_id', 'amazon_artist_id', 'profile_id']
shared_cols = [c for c in new_cols if c in col_names]
cols_str = ', '.join(shared_cols)

120
tests/test_db_genres_json_normalization.py Normal file
View file

@ -0,0 +1,120 @@
"""Tests for the one-time genres CSV->JSON normalization migration.
artists.genres / albums.genres historically stored either a JSON array (new
writes) or a legacy comma-separated string (old writes). _normalize_genres_to_json
rewrites legacy rows to canonical JSON, mirroring the readers' exact parse so the
genre VALUES are unchanged — only the storage format. These tests drive the real
method on a temp database.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from database.music_database import MusicDatabase
def _fresh_db(tmp_path: Path) -> MusicDatabase:
# Init creates the schema and (harmlessly) runs the normalization once on the
# empty DB, setting the marker. Tests clear the marker + seed, then call the
# method directly so they exercise the real normalization logic.
return MusicDatabase(str(tmp_path / "library.db"))
def _seed_and_normalize(db: MusicDatabase, artists, albums=()):
"""Insert (id, name, genres) artists and (id, artist_id, title, genres) albums
with the marker cleared, then run the real migration. Returns nothing."""
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute("DELETE FROM metadata WHERE key = 'genres_json_normalized'")
for aid, name, genres in artists:
cur.execute(
"INSERT INTO artists (id, name, genres) VALUES (?, ?, ?)",
(aid, name, genres),
)
for alid, artist_id, title, genres in albums:
cur.execute(
"INSERT INTO albums (id, artist_id, title, genres) VALUES (?, ?, ?, ?)",
(alid, artist_id, title, genres),
)
conn.commit()
db._normalize_genres_to_json(cur)
conn.commit()
def _get_genres(db: MusicDatabase, table: str, rid: str):
with db._get_connection() as conn:
row = conn.execute(f"SELECT genres FROM {table} WHERE id = ?", (rid,)).fetchone()
return row[0]
def test_csv_genres_normalized_to_json(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
_seed_and_normalize(db, [("a1", "Artist One", "Rock, Pop, Jazz")])
stored = _get_genres(db, "artists", "a1")
assert json.loads(stored) == ["Rock", "Pop", "Jazz"]
def test_existing_json_genres_left_unchanged(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
canonical = json.dumps(["Hip-Hop", "Soul"])
_seed_and_normalize(db, [("a1", "Artist One", canonical)])
# Byte-for-byte identical — no needless churn on already-canonical rows.
assert _get_genres(db, "artists", "a1") == canonical
def test_single_genre_without_comma(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
_seed_and_normalize(db, [("a1", "Artist One", "Electronic")])
assert json.loads(_get_genres(db, "artists", "a1")) == ["Electronic"]
def test_csv_whitespace_and_empties_dropped(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
_seed_and_normalize(db, [("a1", "Artist One", " Rock ,, Pop , ")])
assert json.loads(_get_genres(db, "artists", "a1")) == ["Rock", "Pop"]
def test_albums_table_also_normalized(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
_seed_and_normalize(
db,
artists=[("a1", "Artist One", "Rock")],
albums=[("al1", "a1", "Album One", "Soul, Funk")],
)
assert json.loads(_get_genres(db, "albums", "al1")) == ["Soul", "Funk"]
def test_values_match_legacy_reader_semantics(tmp_path: Path) -> None:
"""The normalized list must equal what the legacy CSV reader would produce,
so downstream genre values are identical pre- and post-migration."""
db = _fresh_db(tmp_path)
raw = "Rock, Pop, Hip-Hop/Rap"
_seed_and_normalize(db, [("a1", "Artist One", raw)])
legacy = [g.strip() for g in raw.split(",") if g.strip()]
assert json.loads(_get_genres(db, "artists", "a1")) == legacy
def test_idempotent_rerun(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
_seed_and_normalize(db, [("a1", "Artist One", "Rock, Pop")])
first = _get_genres(db, "artists", "a1")
# Marker is now set; a second run must be a no-op and leave the value identical.
with db._get_connection() as conn:
cur = conn.cursor()
db._normalize_genres_to_json(cur)
conn.commit()
assert _get_genres(db, "artists", "a1") == first
assert json.loads(first) == ["Rock", "Pop"]
def test_marker_set_after_fresh_init(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
with db._get_connection() as conn:
row = conn.execute(
"SELECT value FROM metadata WHERE key = 'genres_json_normalized'"
).fetchone()
assert row is not None and row[0] == "true"

96
tests/test_db_migration_ledger.py Normal file
View file

@ -0,0 +1,96 @@
"""Tests for the additive schema_migrations ledger + PRAGMA user_version backstop.
The ledger unifies the previously-scattered migration state (marker tables +
metadata flags) into one readable place so a half-migrated DB is detectable.
It is NON-GATING: nothing decides whether a migration runs based on it.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from database.music_database import MusicDatabase
def _fresh_db(tmp_path: Path) -> MusicDatabase:
return MusicDatabase(str(tmp_path / "library.db"))
def test_schema_migrations_table_exists(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
with db._get_connection() as conn:
row = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='schema_migrations'"
).fetchone()
assert row is not None
def test_user_version_stamped(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
with db._get_connection() as conn:
version = conn.execute("PRAGMA user_version").fetchone()[0]
assert version == MusicDatabase.SCHEMA_VERSION == 1
def test_record_migration_is_idempotent(tmp_path: Path) -> None:
db = _fresh_db(tmp_path)
with db._get_connection() as conn:
cur = conn.cursor()
db._record_migration(cur, "unit_test_mig")
db._record_migration(cur, "unit_test_mig")
conn.commit()
n = cur.execute(
"SELECT COUNT(*) FROM schema_migrations WHERE name = 'unit_test_mig'"
).fetchone()[0]
assert n == 1
def test_genres_migration_recorded_on_fresh_init(tmp_path: Path) -> None:
"""The forward pattern: the genres migration records itself in the ledger."""
db = _fresh_db(tmp_path)
with db._get_connection() as conn:
row = conn.execute(
"SELECT 1 FROM schema_migrations WHERE name = 'genres_json'"
).fetchone()
assert row is not None
def test_ledger_backfills_from_existing_signals(tmp_path: Path) -> None:
"""Back-fill records both metadata-flag and marker-table migrations that are
already present, under their canonical ledger names."""
db = _fresh_db(tmp_path)
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute("DELETE FROM schema_migrations")
# A metadata-flag-style signal and a marker-table-style signal.
cur.execute(
"INSERT OR REPLACE INTO metadata (key, value, updated_at) "
"VALUES ('metadata_cache_v1', '1', CURRENT_TIMESTAMP)"
)
cur.execute(
"CREATE TABLE IF NOT EXISTS _genius_search_fix_applied "
"(applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)"
)
conn.commit()
db._sync_migration_ledger(cur)
conn.commit()
names = {r[0] for r in cur.execute("SELECT name FROM schema_migrations")}
assert "metadata_cache_v1" in names # from the metadata flag
assert "genius_search_fix" in names # from the marker table
def test_ledger_does_not_record_absent_signals(tmp_path: Path) -> None:
"""A migration whose signal is absent must NOT be recorded as applied."""
db = _fresh_db(tmp_path)
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute("DELETE FROM schema_migrations")
# Ensure the deezer-cache marker table does not exist.
cur.execute("DROP TABLE IF EXISTS _deezer_cache_v2_migrated")
conn.commit()
db._sync_migration_ledger(cur)
conn.commit()
names = {r[0] for r in cur.execute("SELECT name FROM schema_migrations")}
assert "deezer_cache_v2" not in names

120
tests/test_db_watchlist_amazon_id_migration.py Normal file
View file

@ -0,0 +1,120 @@
"""Regression for the watchlist_artists rebuild dropping amazon_artist_id.
`amazon_artist_id` is added to watchlist_artists via ALTER (music_database.py
~1732), but the table-rebuild migrations (the spotify_id-nullable fix and the
profile-scoped UNIQUE rebuild) recreated the table from a hardcoded column list
that omitted amazon_artist_id — so on upgrade the column AND any stored Amazon
artist IDs were silently dropped.
These tests drive the REAL migrations through MusicDatabase() against a fresh
temp database that starts in the pre-migration shape (no profile-scoped UNIQUE,
amazon_artist_id present with data), then assert the column and its data survive.
Proven differential: with database/music_database.py reverted to pre-fix,
test_amazon_artist_id_data_survives_rebuild FAILS (the column/data are dropped);
with the fix it passes.
"""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from database.music_database import MusicDatabase
# A pre-profile-migration watchlist_artists schema (no UNIQUE(profile_id, ...),
# i.e. exactly the state that triggers the rebuild path) that ALREADY carries
# amazon_artist_id + the other source-id columns — mirroring a real DB that ran
# the 1732 ALTER before the rebuild migrations existed.
_OLD_WATCHLIST_SCHEMA = """
CREATE TABLE watchlist_artists (
id INTEGER PRIMARY KEY AUTOINCREMENT,
spotify_artist_id TEXT UNIQUE,
itunes_artist_id TEXT,
deezer_artist_id TEXT,
discogs_artist_id TEXT,
musicbrainz_artist_id TEXT,
amazon_artist_id TEXT,
artist_name TEXT NOT NULL,
date_added TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_scan_timestamp TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
def _seed_old_db(db_path: Path) -> None:
"""Create a pre-migration watchlist_artists with an Amazon-tagged row."""
conn = sqlite3.connect(str(db_path))
try:
conn.execute(_OLD_WATCHLIST_SCHEMA)
conn.execute(
"INSERT INTO watchlist_artists "
"(spotify_artist_id, amazon_artist_id, artist_name) VALUES (?, ?, ?)",
("spfy_abc", "B0AMAZONXYZ", "Amazon Tagged Artist"),
)
conn.commit()
finally:
conn.close()
def _watchlist_columns(db_path: Path) -> list:
conn = sqlite3.connect(str(db_path))
try:
return [r[1] for r in conn.execute("PRAGMA table_info(watchlist_artists)")]
finally:
conn.close()
def test_amazon_artist_id_column_survives_rebuild(tmp_path: Path) -> None:
"""After the real migrations run, watchlist_artists must still have the
amazon_artist_id column (the rebuild must not drop it)."""
db_path = tmp_path / "old_library.db"
_seed_old_db(db_path)
# Driving MusicDatabase against this path runs the real _initialize_database,
# which fires the watchlist_artists rebuild(s).
MusicDatabase(str(db_path))
cols = _watchlist_columns(db_path)
assert "amazon_artist_id" in cols, (
"amazon_artist_id was dropped by the watchlist_artists rebuild; "
f"columns are: {cols}"
)
# The rebuild's whole purpose: profile-scoped uniqueness must still apply.
assert "profile_id" in cols
def test_amazon_artist_id_data_survives_rebuild(tmp_path: Path) -> None:
"""The stored Amazon artist ID must be carried across the rebuild, not lost.
This is the test that FAILS on pre-fix code."""
db_path = tmp_path / "old_library.db"
_seed_old_db(db_path)
MusicDatabase(str(db_path))
conn = sqlite3.connect(str(db_path))
try:
row = conn.execute(
"SELECT amazon_artist_id FROM watchlist_artists WHERE artist_name = ?",
("Amazon Tagged Artist",),
).fetchone()
finally:
conn.close()
assert row is not None, "the watchlist row was lost entirely during rebuild"
assert row[0] == "B0AMAZONXYZ", (
f"amazon_artist_id data was not preserved across rebuild (got {row[0]!r})"
)
def test_fresh_db_has_amazon_artist_id_column(tmp_path: Path) -> None:
"""A brand-new database (no pre-existing table) must also end up with the
amazon_artist_id column, so fresh installs match upgraded ones."""
db_path = tmp_path / "fresh_library.db"
MusicDatabase(str(db_path))
assert "amazon_artist_id" in _watchlist_columns(db_path)

99
tests/test_source_ids_registry.py Normal file
View file

@ -0,0 +1,99 @@
"""Tests for the canonical source-ID registry (core/source_ids.py)."""
from __future__ import annotations
import sqlite3
import pytest
from core import source_ids as sid
def test_canonical_columns_match_real_schema():
# Spot-check the canonical names against the actual DB columns.
assert sid.id_column("spotify", "artist") == "spotify_artist_id"
assert sid.id_column("deezer", "artist") == "deezer_id"
assert sid.id_column("musicbrainz", "artist") == "musicbrainz_id"
assert sid.id_column("hydrabase", "artist") == "soul_id"
assert sid.id_column("spotify", "album") == "spotify_album_id"
assert sid.id_column("musicbrainz", "album") == "musicbrainz_release_id"
assert sid.id_column("deezer", "album") == "deezer_id"
assert sid.id_column("spotify", "track") == "spotify_track_id"
assert sid.id_column("musicbrainz", "track") == "musicbrainz_recording_id"
def test_id_column_unknown_returns_none():
assert sid.id_column("nonesuch", "artist") is None
assert sid.id_column("spotify", "playlist") is None
def test_id_keys_canonical_first_then_aliases():
keys = sid.id_keys("deezer", "artist")
assert keys[0] == "deezer_id" # canonical first
assert "deezer_artist_id" in keys # watchlist/pool alias
assert "similar_artist_deezer_id" in keys
def test_get_id_reads_canonical_column():
row = {"deezer_id": "525046", "name": "Artist"}
assert sid.get_id(row, "deezer", "artist") == "525046"
def test_get_id_falls_back_to_alias():
# A watchlist-shaped row uses deezer_artist_id, not deezer_id.
row = {"deezer_artist_id": "999", "artist_name": "X"}
assert sid.get_id(row, "deezer", "artist") == "999"
def test_get_id_prefers_canonical_over_alias():
row = {"deezer_id": "canon", "deezer_artist_id": "alias"}
assert sid.get_id(row, "deezer", "artist") == "canon"
def test_get_id_missing_and_empty_return_none():
assert sid.get_id({"name": "X"}, "deezer", "artist") is None
assert sid.get_id({"deezer_id": ""}, "deezer", "artist") is None
assert sid.get_id({"deezer_id": None}, "deezer", "artist") is None
def test_get_id_works_with_sqlite_row():
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("CREATE TABLE t (spotify_artist_id TEXT, name TEXT)")
conn.execute("INSERT INTO t VALUES ('spfy123', 'Artist')")
row = conn.execute("SELECT * FROM t").fetchone()
conn.close()
assert sid.get_id(row, "spotify", "artist") == "spfy123"
# A column not present on the row must not raise — just None.
assert sid.get_id(row, "deezer", "artist") is None
def test_source_id_map_builds_provider_dict():
row = {
"spotify_artist_id": "s1",
"deezer_id": "d1",
"itunes_artist_id": None,
}
result = sid.source_id_map(row, "artist", providers=["spotify", "deezer", "itunes"])
assert result == {"spotify": "s1", "deezer": "d1", "itunes": None}
def test_source_id_map_default_covers_all_providers():
result = sid.source_id_map({"deezer_id": "d1"}, "artist")
assert result["deezer"] == "d1"
assert "spotify" in result and result["spotify"] is None
def test_source_id_field_unchanged_after_registry_refactor():
"""artist_source_lookup.SOURCE_ID_FIELD must keep its exact prior mapping
after being folded into the registry (no behavior change)."""
from core.artist_source_lookup import SOURCE_ID_FIELD
assert SOURCE_ID_FIELD == {
"spotify": "spotify_artist_id",
"itunes": "itunes_artist_id",
"deezer": "deezer_id",
"discogs": "discogs_id",
"hydrabase": "soul_id",
"musicbrainz": "musicbrainz_id",
"amazon": "amazon_id",
}

16
web_server.py
View file

@ -7601,15 +7601,15 @@ def get_artist_detail(artist_id):
try:
from core.metadata.lookup import MetadataLookupOptions
from core.metadata_service import get_artist_detail_discography as _get_artist_detail_discography
from core.source_ids import source_id_map
artist_source_ids = {
'spotify': artist_info.get('spotify_artist_id'),
'deezer': artist_info.get('deezer_id'),
'itunes': artist_info.get('itunes_artist_id'),
'discogs': artist_info.get('discogs_id'),
'hydrabase': artist_info.get('soul_id'),
'amazon': artist_info.get('amazon_id'),
}
# Per-source artist IDs, read via the canonical source-ID registry
# (same columns as before: spotify_artist_id / deezer_id /
# itunes_artist_id / discogs_id / soul_id / amazon_id).
artist_source_ids = source_id_map(
artist_info, 'artist',
providers=('spotify', 'deezer', 'itunes', 'discogs', 'hydrabase', 'amazon'),
)
artist_detail_discography = _get_artist_detail_discography(
artist_id,