diff --git a/core/artist_source_lookup.py b/core/artist_source_lookup.py new file mode 100644 index 00000000..32eac787 --- /dev/null +++ b/core/artist_source_lookup.py @@ -0,0 +1,88 @@ +"""Source-artist → library lookup helpers. + +Extracted from `web_server.py` so the logic can be imported and unit-tested +without booting the Flask app, Spotify client, Soulseek connection, etc. + +Two concepts live here: + + * ``SOURCE_ID_FIELD`` — the per-source column on the ``artists`` table that + stores the external service ID (Spotify track ID, Deezer artist ID, …). + This map is what ties a result clicked in the source-aware Search results + back to a library record so we can serve the richer library view. + + * ``find_library_artist_for_source`` — given a source-aware click (e.g. + ``deezer:525046``), try to locate a matching library artist. First by + direct column match against the source's ID column, then by case- + insensitive name match scoped to the active media server. +""" + +from __future__ import annotations + +import logging +from typing import Optional + +logger = logging.getLogger("artist_source_lookup") + + +SOURCE_ONLY_ARTIST_SOURCES = frozenset({ + "spotify", "itunes", "deezer", "discogs", "hydrabase", "musicbrainz", +}) + + +SOURCE_ID_FIELD = { + "spotify": "spotify_artist_id", + "itunes": "itunes_artist_id", + "deezer": "deezer_id", + "discogs": "discogs_id", + "hydrabase": "soul_id", + "musicbrainz": "musicbrainz_id", +} + + +def find_library_artist_for_source( + database, + source: str, + source_artist_id: str, + artist_name: str, + active_server: Optional[str] = None, +) -> Optional[str]: + """Return the library PK of an artist matching the source-aware click. + + Lookup order: + 1. Direct match on the source-specific ID column (server-agnostic — any + library record with the right external ID is a hit). + 2. Case-insensitive name match within ``active_server`` (defaults to the + active media server when not provided), so we don't jump the user + across server contexts on a name collision. + + Returns ``None`` on miss or on any database error. + """ + column = SOURCE_ID_FIELD.get(source) + if not column: + return None + + try: + with database._get_connection() as conn: + cursor = conn.cursor() + cursor.execute( + f"SELECT id, name FROM artists WHERE {column} = ? LIMIT 1", + (str(source_artist_id),), + ) + row = cursor.fetchone() + if row: + return row[0] + + if artist_name and active_server: + cursor.execute( + "SELECT id FROM artists " + "WHERE LOWER(name) = LOWER(?) AND server_source = ? LIMIT 1", + (artist_name, active_server), + ) + row = cursor.fetchone() + if row: + return row[0] + except Exception as e: + logger.debug( + f"Library upgrade lookup failed for {source}:{source_artist_id}: {e}" + ) + return None diff --git a/tests/test_artist_source_lookup.py b/tests/test_artist_source_lookup.py index cbcf07d8..c33d864f 100644 --- a/tests/test_artist_source_lookup.py +++ b/tests/test_artist_source_lookup.py @@ -1,36 +1,29 @@ -"""Regression tests for the source-artist → library lookup path and the -source-only artist-detail response shape. +"""Tests for the source-artist → library lookup helpers in +``core/artist_source_lookup.py``. -These tests exist to catch the class of bug we hit in April 2026 where the +These exist to catch the class of bug we hit in April 2026 where the watchlist-config enrichment query referenced a column name (``deezer_artist_id``) that lived on ``watchlist_artists`` but NOT on ``artists``, producing a ``no such column`` error on every request. -``web_server.py`` cannot be imported at test time (it initialises Spotify, -Soulseek, Plex, etc.), so the ``_SOURCE_ID_FIELD`` map and the -``_build_source_only_artist_detail`` response contract are verified by reading -``web_server.py`` as source text and parsing the dict literal. +The earlier version of this file AST-parsed ``web_server.py`` because the +logic lived inline there and could not be imported at test time. The logic +has since been extracted to a side-effect-free module, so we can just import +and call it directly. """ from __future__ import annotations -import ast -import re -from pathlib import Path - import pytest +from core.artist_source_lookup import ( + SOURCE_ID_FIELD, + SOURCE_ONLY_ARTIST_SOURCES, + find_library_artist_for_source, +) from database.music_database import MusicDatabase -_ROOT = Path(__file__).resolve().parent.parent -_WEB_SERVER = _ROOT / "web_server.py" - - -# --------------------------------------------------------------------------- -# Expected mapping — must match web_server.py::_SOURCE_ID_FIELD -# --------------------------------------------------------------------------- - EXPECTED_SOURCE_ID_FIELD = { "spotify": "spotify_artist_id", "itunes": "itunes_artist_id", @@ -41,39 +34,13 @@ EXPECTED_SOURCE_ID_FIELD = { } -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - -def _extract_source_id_field_dict() -> dict[str, str]: - """Parse web_server.py and return the _SOURCE_ID_FIELD dict literal.""" - source = _WEB_SERVER.read_text(encoding="utf-8", errors="replace") - tree = ast.parse(source) - for node in ast.walk(tree): - if isinstance(node, ast.Assign): - for target in node.targets: - if isinstance(target, ast.Name) and target.id == "_SOURCE_ID_FIELD": - return ast.literal_eval(node.value) - raise AssertionError("_SOURCE_ID_FIELD not found in web_server.py") - - -def _extract_function_source(fn_name: str) -> str: - """Return the full source text of a top-level function in web_server.py.""" - source = _WEB_SERVER.read_text(encoding="utf-8", errors="replace") - tree = ast.parse(source) - for node in ast.walk(tree): - if isinstance(node, ast.FunctionDef) and node.name == fn_name: - return ast.get_source_segment(source, node) or "" - raise AssertionError(f"function {fn_name!r} not found in web_server.py") - - @pytest.fixture def db(tmp_path): """Fresh MusicDatabase — runs all migrations so source-id columns exist.""" return MusicDatabase(str(tmp_path / "music.db")) -def _insert_artist(db, *, artist_id: str, name: str, server_source: str = "plex", **extra): +def _insert_artist(db, *, artist_id, name, server_source="plex", **extra): """Insert a row into the artists table with the given extra columns.""" cols = ["id", "name", "server_source"] + list(extra.keys()) vals = [artist_id, name, server_source] + list(extra.values()) @@ -87,46 +54,50 @@ def _insert_artist(db, *, artist_id: str, name: str, server_source: str = "plex" # =========================================================================== -# Group A — _SOURCE_ID_FIELD contract +# Group A — SOURCE_ID_FIELD constants # =========================================================================== class TestSourceIdFieldMapping: - """The mapping web_server.py uses to join source artists back to the - library ``artists`` table must stay in sync with this test's expectations - AND with the real column names on the table.""" + """The mapping the lookup uses to join source artists back to the library + ``artists`` table must stay in sync with this test's expectations AND with + the real column names on the table.""" def test_mapping_matches_expected(self): - actual = _extract_source_id_field_dict() - assert actual == EXPECTED_SOURCE_ID_FIELD, ( - "web_server.py::_SOURCE_ID_FIELD changed; update " - "EXPECTED_SOURCE_ID_FIELD (and the test body) to match." + assert SOURCE_ID_FIELD == EXPECTED_SOURCE_ID_FIELD, ( + "SOURCE_ID_FIELD changed; update EXPECTED_SOURCE_ID_FIELD " + "(and the test body) to match." ) + def test_source_only_set_matches_mapping_keys(self): + """Sources eligible for the source-only fallback must all have a + column to look them up by — otherwise the upgrade path silently + returns None.""" + assert SOURCE_ONLY_ARTIST_SOURCES == frozenset(SOURCE_ID_FIELD.keys()) + def test_every_mapped_column_exists_on_artists_table(self, db): """Regression for the 2026-04 ``deezer_artist_id`` typo: every column - referenced by _SOURCE_ID_FIELD must exist on the ``artists`` table.""" + referenced by SOURCE_ID_FIELD must exist on the ``artists`` table.""" with db._get_connection() as conn: cursor = conn.execute("PRAGMA table_info(artists)") existing = {row[1] for row in cursor.fetchall()} missing = { source: column - for source, column in EXPECTED_SOURCE_ID_FIELD.items() + for source, column in SOURCE_ID_FIELD.items() if column not in existing } assert not missing, ( - "Columns declared in _SOURCE_ID_FIELD are missing from the " + "Columns declared in SOURCE_ID_FIELD are missing from the " f"artists table: {missing}. Available columns: {sorted(existing)}" ) # =========================================================================== -# Group B — _find_library_artist_for_source lookup behaviour +# Group B — find_library_artist_for_source behaviour # =========================================================================== -class TestLibraryArtistLookup: - """Replicates the two queries in _find_library_artist_for_source so we - catch column-name drift or schema regressions immediately.""" +class TestFindLibraryArtistForSource: + """Behavioural tests against a real (in-memory) MusicDatabase.""" @pytest.mark.parametrize("source,column", list(EXPECTED_SOURCE_ID_FIELD.items())) def test_lookup_by_source_id_column(self, db, source, column): @@ -138,59 +109,78 @@ class TestLibraryArtistLookup: **{column: source_value}, ) - with db._get_connection() as conn: - cursor = conn.execute( - f"SELECT id, name FROM artists WHERE {column} = ? LIMIT 1", - (source_value,), - ) - row = cursor.fetchone() - - assert row is not None, ( - f"Lookup by {column} returned no row — schema/query mismatch?" + result = find_library_artist_for_source( + db, source, source_value, artist_name="" ) - assert row[0] == f"pk-{source}" + assert result == f"pk-{source}" + + def test_unknown_source_returns_none(self, db): + assert find_library_artist_for_source( + db, "made-up-source", "anything", artist_name="Anything" + ) is None def test_lookup_misses_when_source_id_unknown(self, db): - with db._get_connection() as conn: - cursor = conn.execute( - "SELECT id FROM artists WHERE deezer_id = ? LIMIT 1", - ("does-not-exist",), - ) - assert cursor.fetchone() is None + _insert_artist(db, artist_id="pk-real", name="Real Artist", deezer_id="dz-real") + assert find_library_artist_for_source( + db, "deezer", "dz-not-real", artist_name="" + ) is None - def test_name_fallback_matches_case_insensitively_within_server(self, db): + def test_name_fallback_matches_within_active_server(self, db): _insert_artist(db, artist_id="pk-a", name="Kendrick Lamar", server_source="plex") _insert_artist(db, artist_id="pk-b", name="KENDRICK LAMAR", server_source="jellyfin") - with db._get_connection() as conn: - cursor = conn.execute( - "SELECT id FROM artists " - "WHERE LOWER(name) = LOWER(?) AND server_source = ? LIMIT 1", - ("kendrick lamar", "plex"), - ) - row = cursor.fetchone() + result = find_library_artist_for_source( + db, "deezer", "no-id-match", artist_name="kendrick lamar", + active_server="plex", + ) + assert result == "pk-a" - assert row is not None and row[0] == "pk-a" - - def test_name_fallback_respects_server_scope(self, db): - """Only the active-server artist should match; the other server's copy - is deliberately ignored to avoid cross-server context jumps.""" + def test_name_fallback_skips_other_servers(self, db): + """Active-server scope is required so we don't jump the user across + server contexts on a name collision.""" _insert_artist(db, artist_id="pk-jelly", name="Taylor Swift", server_source="jellyfin") - with db._get_connection() as conn: - cursor = conn.execute( - "SELECT id FROM artists " - "WHERE LOWER(name) = LOWER(?) AND server_source = ? LIMIT 1", - ("Taylor Swift", "plex"), - ) - assert cursor.fetchone() is None + result = find_library_artist_for_source( + db, "deezer", "no-id-match", artist_name="Taylor Swift", + active_server="plex", + ) + assert result is None + + def test_name_fallback_requires_active_server(self, db): + """Without an active_server we shouldn't fall through to a global + name match — too easy to land the user on the wrong record.""" + _insert_artist(db, artist_id="pk-x", name="Some Artist", server_source="plex") + + result = find_library_artist_for_source( + db, "deezer", "no-id-match", artist_name="Some Artist", + active_server=None, + ) + assert result is None + + def test_id_match_wins_over_name_match(self, db): + """If both a source-id match and a name match exist, the id match + should take priority — it's the more reliable signal.""" + _insert_artist( + db, artist_id="pk-id-match", name="Different Name", + deezer_id="dz-shared", server_source="plex", + ) + _insert_artist( + db, artist_id="pk-name-match", name="The Searched Artist", + server_source="plex", + ) + + result = find_library_artist_for_source( + db, "deezer", "dz-shared", artist_name="The Searched Artist", + active_server="plex", + ) + assert result == "pk-id-match" # =========================================================================== # Group C — Watchlist-config enrichment query schema contract # =========================================================================== -class TestWatchlistConfigEnrichment: +class TestWatchlistConfigEnrichmentQueries: """The watchlist-config GET (web_server.py ~line 42196) joins ``watchlist_artists`` against ``artists``. Both tables use different column names for the same external IDs (``deezer_id`` on artists, @@ -241,61 +231,3 @@ class TestWatchlistConfigEnrichment: assert "deezer_artist_id" not in artists_cols assert "discogs_artist_id" not in artists_cols - - -# =========================================================================== -# Group D — _build_source_only_artist_detail response-shape contract -# =========================================================================== - -class TestSourceOnlyArtistDetailContract: - """Contract test for the JSON response produced by - _build_source_only_artist_detail. We assert the function source contains - the response-key identifiers we rely on from the frontend. If web_server - drops or renames one, the test fires before the JS tries to read a - ``undefined`` field.""" - - @pytest.fixture(autouse=True) - def _load_source(self): - self.src = _extract_function_source("_build_source_only_artist_detail") - - @pytest.mark.parametrize("key", [ - # Top-level response shape - "success", - "artist", - "discography", - "enrichment_coverage", - # artist_info fields - "image_url", - "server_source", - "genres", - # Last.fm enrichment - "lastfm_bio", - "lastfm_listeners", - "lastfm_playcount", - "lastfm_url", - ]) - def test_function_references_response_key(self, key): - assert f"'{key}'" in self.src or f'"{key}"' in self.src, ( - f"_build_source_only_artist_detail no longer references " - f"response key {key!r}" - ) - - def test_function_sets_source_specific_id_field(self): - """The function must look up _SOURCE_ID_FIELD and stamp the - appropriate column name into artist_info so the correct service - badge renders. Regression guard for a refactor that drops the - dynamic assignment.""" - assert "_SOURCE_ID_FIELD" in self.src - # Should assign via dict-style setattr on artist_info - assert re.search( - r"artist_info\[\s*source_id_field\s*\]\s*=\s*artist_id", - self.src, - ), "expected dynamic artist_info[source_id_field] = artist_id assignment" - - def test_function_disables_variant_dedup(self): - """Source-only view must pass ``dedup_variants=False`` so every - release surfaces — matching the prior inline Artists-page behaviour - the user explicitly requested.""" - assert "dedup_variants=False" in self.src, ( - "_build_source_only_artist_detail must opt out of variant dedup" - ) diff --git a/web_server.py b/web_server.py index 7396b39c..d58f0fe8 100644 --- a/web_server.py +++ b/web_server.py @@ -11316,64 +11316,22 @@ def test_artist_endpoint(artist_id): "message": f"Test endpoint working for artist ID: {artist_id}" }) -_SOURCE_ONLY_ARTIST_SOURCES = frozenset({ - 'spotify', 'itunes', 'deezer', 'discogs', 'hydrabase', 'musicbrainz', -}) - - -_SOURCE_ID_FIELD = { - 'spotify': 'spotify_artist_id', - 'itunes': 'itunes_artist_id', - 'deezer': 'deezer_id', - 'discogs': 'discogs_id', - 'hydrabase': 'soul_id', - 'musicbrainz': 'musicbrainz_id', -} +from core.artist_source_lookup import ( + SOURCE_ID_FIELD as _SOURCE_ID_FIELD, + SOURCE_ONLY_ARTIST_SOURCES as _SOURCE_ONLY_ARTIST_SOURCES, + find_library_artist_for_source as _core_find_library_artist_for_source, +) def _find_library_artist_for_source(database, source, source_artist_id, artist_name): - """Try to upgrade a source-artist click to a library lookup. - - Returns the library PK of an artist that matches either the source-specific - ID column (e.g. WHERE deezer_id = source_artist_id) or, as a fallback, the - artist name in the active server. Returns None if no match. - """ - column = _SOURCE_ID_FIELD.get(source) - if not column: - return None + """Thin wrapper that injects the active-server context for the core lookup.""" try: - with database._get_connection() as conn: - cursor = conn.cursor() - # Match by source-specific ID column. Server-source-agnostic: any - # library record (Plex/Jellyfin/Navidrome/SoulSync) that has the - # right external ID is a hit. - cursor.execute( - f"SELECT id, name FROM artists WHERE {column} = ? LIMIT 1", - (str(source_artist_id),) - ) - row = cursor.fetchone() - if row: - return row[0] - - # Fallback: case-insensitive name match within the active server only, - # to avoid jumping the user across server contexts unintentionally. - if artist_name: - try: - active_server = config_manager.get_active_media_server() - except Exception: - active_server = None - if active_server: - cursor.execute( - "SELECT id FROM artists " - "WHERE LOWER(name) = LOWER(?) AND server_source = ? LIMIT 1", - (artist_name, active_server) - ) - row = cursor.fetchone() - if row: - return row[0] - except Exception as e: - logger.debug(f"Library upgrade lookup failed for {source}:{source_artist_id}: {e}") - return None + active_server = config_manager.get_active_media_server() + except Exception: + active_server = None + return _core_find_library_artist_for_source( + database, source, source_artist_id, artist_name, active_server=active_server + ) def _build_source_only_artist_detail(artist_id, artist_name, source):