Add regression tests for source-artist lookup + dedup_variants flag
Four targeted backend tests for behaviour added during the Search/Artists unification work: 1. _SOURCE_ID_FIELD mapping is parsed out of web_server.py via AST and compared against an explicit expectation, so silent renames break the test instead of silently breaking library-upgrade detection. 2. Every column in _SOURCE_ID_FIELD must exist on the real artists table after migrations run. This is the schema-vs-query contract that the `deezer_artist_id` typo would have failed instantly. 3. The two queries from the watchlist-config enrichment path execute verbatim against a fresh DB — separate ones for the artists table (deezer_id / discogs_id) and the watchlist_artists join (deezer_artist_id). Documents the column-name split that caused the original bug. 4. Static contract test for _build_source_only_artist_detail's response shape: every JSON key the frontend reads (success/artist/discography/ image_url/server_source/genres/lastfm_*) must appear in the function source, plus the dynamic source-id stamp and the dedup_variants=False opt-out. Plus a behavioural test for MetadataLookupOptions.dedup_variants=False in test_metadata_service_discography.py — proves the flag actually keeps variant releases that get_artist_detail_discography would otherwise collapse to a single canonical entry.
This commit is contained in:
Broque Thomas
parent
7625362c49
commit
12c23b6b89
2 changed files with 356 additions and 0 deletions
301
tests/test_artist_source_lookup.py
Normal file
301
tests/test_artist_source_lookup.py
Normal file
|
|
@ -0,0 +1,301 @@
|
|||
"""Regression tests for the source-artist → library lookup path and the
|
||||
source-only artist-detail response shape.
|
||||
|
||||
These tests 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.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
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",
|
||||
"deezer": "deezer_id",
|
||||
"discogs": "discogs_id",
|
||||
"hydrabase": "soul_id",
|
||||
"musicbrainz": "musicbrainz_id",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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):
|
||||
"""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())
|
||||
placeholders = ",".join("?" for _ in cols)
|
||||
with db._get_connection() as conn:
|
||||
conn.execute(
|
||||
f"INSERT INTO artists ({','.join(cols)}) VALUES ({placeholders})",
|
||||
vals,
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group A — _SOURCE_ID_FIELD contract
|
||||
# ===========================================================================
|
||||
|
||||
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."""
|
||||
|
||||
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."
|
||||
)
|
||||
|
||||
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."""
|
||||
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()
|
||||
if column not in existing
|
||||
}
|
||||
assert not missing, (
|
||||
"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
|
||||
# ===========================================================================
|
||||
|
||||
class TestLibraryArtistLookup:
|
||||
"""Replicates the two queries in _find_library_artist_for_source so we
|
||||
catch column-name drift or schema regressions immediately."""
|
||||
|
||||
@pytest.mark.parametrize("source,column", list(EXPECTED_SOURCE_ID_FIELD.items()))
|
||||
def test_lookup_by_source_id_column(self, db, source, column):
|
||||
source_value = f"{source}-test-artist-123"
|
||||
_insert_artist(
|
||||
db,
|
||||
artist_id=f"pk-{source}",
|
||||
name=f"{source.title()} Test Artist",
|
||||
**{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?"
|
||||
)
|
||||
assert row[0] == f"pk-{source}"
|
||||
|
||||
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
|
||||
|
||||
def test_name_fallback_matches_case_insensitively_within_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()
|
||||
|
||||
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."""
|
||||
_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
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group C — Watchlist-config enrichment query schema contract
|
||||
# ===========================================================================
|
||||
|
||||
class TestWatchlistConfigEnrichment:
|
||||
"""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,
|
||||
``deezer_artist_id`` on watchlist_artists). The queries must use the
|
||||
correct column per table."""
|
||||
|
||||
def test_artists_enrichment_query_executes(self, db):
|
||||
"""Run the exact SELECT from web_server.py verbatim — must not raise
|
||||
``no such column``."""
|
||||
with db._get_connection() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
SELECT banner_url, summary, style, mood, label, genres
|
||||
FROM artists
|
||||
WHERE spotify_artist_id = ?
|
||||
OR itunes_artist_id = ?
|
||||
OR deezer_id = ?
|
||||
OR discogs_id = ?
|
||||
LIMIT 1
|
||||
""",
|
||||
("x", "x", "x", "x"),
|
||||
)
|
||||
|
||||
def test_watchlist_join_query_executes(self, db):
|
||||
"""The paired query hits ``watchlist_artists`` where the Deezer column
|
||||
is ``deezer_artist_id`` — confirm that shape works too."""
|
||||
with db._get_connection() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
SELECT rr.album_name, rr.release_date, rr.album_cover_url, rr.track_count
|
||||
FROM recent_releases rr
|
||||
JOIN watchlist_artists wa ON rr.watchlist_artist_id = wa.id
|
||||
WHERE wa.spotify_artist_id = ?
|
||||
OR wa.itunes_artist_id = ?
|
||||
OR wa.deezer_artist_id = ?
|
||||
ORDER BY rr.release_date DESC
|
||||
LIMIT 6
|
||||
""",
|
||||
("x", "x", "x"),
|
||||
)
|
||||
|
||||
def test_artists_table_does_not_have_watchlist_column_names(self, db):
|
||||
"""Document the schema split that caused the original bug: these
|
||||
suffixed names only exist on ``watchlist_artists``, never ``artists``."""
|
||||
with db._get_connection() as conn:
|
||||
cursor = conn.execute("PRAGMA table_info(artists)")
|
||||
artists_cols = {row[1] for row in cursor.fetchall()}
|
||||
|
||||
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"
|
||||
)
|
||||
|
|
@ -483,6 +483,61 @@ def test_get_artist_detail_discography_dedups_variant_releases(monkeypatch):
|
|||
assert result["albums"][0]["track_count"] == 10
|
||||
|
||||
|
||||
def test_get_artist_detail_discography_keeps_variants_when_dedup_disabled(monkeypatch):
|
||||
"""MetadataLookupOptions.dedup_variants=False is the source-only artist
|
||||
detail code path — used so the standalone /artist-detail page can show
|
||||
every release the source returns (matching the retired inline Artists
|
||||
page behaviour)."""
|
||||
monkeypatch.setattr(
|
||||
metadata_service,
|
||||
"get_artist_discography",
|
||||
lambda artist_id, artist_name='', options=None: {
|
||||
"albums": [
|
||||
{
|
||||
"id": "album-standard",
|
||||
"name": "Variant Album",
|
||||
"album_type": "album",
|
||||
"image_url": "https://img.example/standard.jpg",
|
||||
"release_date": "2024-01-05",
|
||||
"total_tracks": 10,
|
||||
},
|
||||
{
|
||||
"id": "album-swedish",
|
||||
"name": "Variant Album (Swedish Edition)",
|
||||
"album_type": "album",
|
||||
"image_url": "https://img.example/swedish.jpg",
|
||||
"release_date": "2024-01-05",
|
||||
"total_tracks": 12,
|
||||
},
|
||||
{
|
||||
"id": "album-remaster",
|
||||
"name": "Variant Album (2023 Abbey Road Remaster)",
|
||||
"album_type": "album",
|
||||
"image_url": "https://img.example/remaster.jpg",
|
||||
"release_date": "2024-01-05",
|
||||
"total_tracks": 10,
|
||||
},
|
||||
],
|
||||
"singles": [],
|
||||
"source": "deezer",
|
||||
"source_priority": ["deezer", "spotify"],
|
||||
},
|
||||
)
|
||||
|
||||
result = metadata_service.get_artist_detail_discography(
|
||||
"artist-1",
|
||||
"Artist One",
|
||||
MetadataLookupOptions(dedup_variants=False),
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert [album["id"] for album in result["albums"]] == [
|
||||
"album-standard",
|
||||
"album-swedish",
|
||||
"album-remaster",
|
||||
]
|
||||
|
||||
|
||||
def test_get_artist_discography_keeps_provider_artist_ids(monkeypatch):
|
||||
class _SpotifyArtistIdClient(_FakeSourceClient):
|
||||
def get_artist_albums(self, artist_id, **kwargs):
|
||||
|
|
|
|||
Loading…
Reference in a new issue