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

Extract source-artist lookup helpers from web_server.py to core module

Browse source 
Cin pointed out that the prior version of test_artist_source_lookup.py
AST-parsed web_server.py to verify a constant and to string-match a
function's response keys. That was a workaround for the fact that
web_server.py can't be imported at test time (it boots Spotify,
Soulseek, Plex, etc.) — the right answer is to move the logic into a
side-effect-free module so it can be imported and tested directly.

This commit:
  - adds core/artist_source_lookup.py containing the SOURCE_ID_FIELD
    map, the SOURCE_ONLY_ARTIST_SOURCES set, and find_library_artist_for_source
  - replaces the inline definitions in web_server.py with imports +
    a thin wrapper that injects the active media server
  - rewrites the tests to import from the core module directly:
      * mapping correctness is now a plain equality assertion
      * lookup behaviour is exercised against a real MusicDatabase
      * the AST parse and the string-matching contract test class are
        gone
  - drops the _build_source_only_artist_detail contract test entirely
    (the weakest of the four — it was just string-matching the function
    body); when that function moves to core/ it can get a real
    behavioural test alongside.

Test runtime drops from ~161s to ~5.8s. All 18 tests pass.
This commit is contained in:
Broque Thomas 2026-04-22 22:07:23 -07:00
parent f684bd603d
commit a097cf3d5a
3 changed files with 188 additions and 210 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

88
core/artist_source_lookup.py Normal file
View file

@ -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

244
tests/test_artist_source_lookup.py
View file

@ -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"
)

66
web_server.py
View file

@ -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):