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

Address Cin review: extract helpers, indexed pool fetch, tidy nits

Browse source 
Three changes folded into one perf+cleanup pass:

1. Indexed fast path for the per-artist pool fetch. The previous
   `search_tracks(artist=name)` call hit
   `unidecode_lower(artists.name) LIKE ?`, a function-in-WHERE that
   can't use `idx_artists_name`. New `MusicDatabase.get_artist_tracks_indexed`
   does a two-step lookup: exact-name match (indexed) plus a
   case-insensitive fallback, then `tracks WHERE artist_id IN (...)`
   via `idx_tracks_artist_id`. Drops per-artist fetch from seconds to
   milliseconds for the common case. The sync helper falls back to
   the old LIKE-based `search_tracks` only when the indexed lookup
   finds nothing, preserving diacritic recall and `tracks.track_artist`
   feature-artist matches with zero regression.

2. Public text-normalization helper. Lifted the body of
   `MusicDatabase._normalize_for_comparison` into
   `core/text/normalize.py:normalize_for_comparison` so callers outside
   the database layer (matching engine, sync pool, future import-side
   comparisons) don't reach across the module boundary into a
   leading-underscore "private" method. The DB method now delegates,
   so existing internal call sites stay untouched. Sync's lazy pool
   now imports the public helper.

3. Artist-name walker extracted. `_artist_name` at module level in
   `services/sync_service.py` replaces two near-identical inline
   str-or-dict-or-fallback walkers (one in `sync_playlist`, one in
   `_find_track_in_media_server`). Returns `''` for None instead of
   the literal string `'None'`.

Plus three small tidies from the same review:

- `_POOL_FETCH_LIMIT = 10000` constant in place of the literal at the
  pool-fetch call site.
- Trimmed the verbose docstring + comment block on the pool helper.
- Set-intersection predicate for the trigger-shape reset in
  `core/automation/api.py` instead of a two-line `or` chain.

Also removed the duplicate `_get_active_media_client()` call at
sync_service.py:212/214 — pre-existing wart that was sitting in the
same block I was editing.

Tests: 21 new tests across `tests/database/`, `tests/sync/`, and
`tests/text/`, plus updates to the existing pool tests to cover the
new fast/fallback split. Full suite stays green (3953 passing).
This commit is contained in:
Broque Thomas 2026-05-24 23:33:55 -07:00
parent 687bb0ca2c
commit feb6778af4
10 changed files with 412 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
core/automation/api.py
View file

@ -225,11 +225,10 @@ def update_automation(
if cycle_path:
return {'error': f'Signal cycle detected: {cycle_path}. This would cause an infinite loop.'}, 400
trigger_changed = (
'trigger_type' in update_fields
or 'trigger_config' in update_fields
)
if trigger_changed:
# Schedule-shape changes must invalidate the stored next_run so the
# scheduler recomputes it; otherwise restart-survival logic keeps the
# leftover timestamp from the previous interval.
if {'trigger_type', 'trigger_config'} & update_fields.keys():
update_fields['next_run'] = None
success = database.update_automation(automation_id, **update_fields)

0
core/text/__init__.py Normal file
View file

41
core/text/normalize.py Normal file
View file

@ -0,0 +1,41 @@
"""Shared text-normalization helpers.
Extracted from `MusicDatabase._normalize_for_comparison` so callers
outside the database layer (matching engine, sync candidate pool,
import comparisons) don't have to reach across the module boundary
into a leading-underscore "private" method.
Pure functions, no I/O.
"""
from __future__ import annotations
import logging
logger = logging.getLogger(__name__)
try:
from unidecode import unidecode as _unidecode
_HAS_UNIDECODE = True
except ImportError:
_unidecode = None # type: ignore[assignment]
_HAS_UNIDECODE = False
logger.warning("unidecode not available, accent matching may be limited")
def normalize_for_comparison(text: str) -> str:
"""Lowercase + strip whitespace + fold accents to ASCII.
``é → e``, ``ñ → n``, ``Björk → bjork``. Used as the dictionary key
for the sync candidate pool and for fuzzy library lookups where
diacritic differences must NOT split a single artist into two pool
entries.
Empty / falsy input returns ``""`` so callers can blindly key dicts
with the result.
"""
if not text:
return ""
if _HAS_UNIDECODE:
text = _unidecode(text)
return text.lower().strip()

70
database/music_database.py
View file

@ -6370,6 +6370,54 @@ class MusicDatabase:
logger.error(f"Error fetching candidate albums for artist '{artist}': {e}")
return candidates
def get_artist_tracks_indexed(self, name: str, server_source: Optional[str] = None, limit: int = 10000) -> List[DatabaseTrack]:
"""Indexed two-step lookup: artist_id by exact name (then case-insensitive
fallback), then tracks via `artist_id IN (...)`. Avoids the function-in-WHERE
pattern in search_tracks that defeats the artists.name index. Returns []
when the artist isn't in the library — caller can decide to fall back to
the slower LIKE-based path for track_artist / diacritic recall."""
if not name:
return []
try:
conn = self._get_connection()
cursor = conn.cursor()
# Step 1: exact case-sensitive match — hits idx_artists_name in O(log n).
# Spotify's canonical artist names match the library 90%+ of the time.
cursor.execute("SELECT id FROM artists WHERE name = ?", (name,))
artist_ids = [r['id'] for r in cursor.fetchall()]
# Step 2: case-insensitive fallback if exact missed. Full scan, but only
# runs on the (uncommon) miss path so amortized cost stays low.
if not artist_ids:
cursor.execute("SELECT id FROM artists WHERE LOWER(name) = LOWER(?)", (name,))
artist_ids = [r['id'] for r in cursor.fetchall()]
if not artist_ids:
return []
placeholders = ','.join('?' for _ in artist_ids)
where = f"t.artist_id IN ({placeholders})"
params: list = list(artist_ids)
if server_source:
where += " AND t.server_source = ?"
params.append(server_source)
params.append(limit)
cursor.execute(f"""
SELECT t.*, a.name as artist_name, al.title as album_title,
al.thumb_url as album_thumb_url
FROM tracks t
JOIN artists a ON a.id = t.artist_id
JOIN albums al ON al.id = t.album_id
WHERE {where}
LIMIT ?
""", params)
return self._rows_to_tracks(cursor.fetchall())
except Exception as e:
logger.error(f"Error fetching indexed artist tracks for '{name}': {e}")
return []
def get_candidate_tracks_for_albums(self, album_ids: List) -> List[DatabaseTrack]:
"""
Fetch every track belonging to the given set of album IDs in a single query.
@ -6897,22 +6945,12 @@ class MusicDatabase:
return unique_variations
def _normalize_for_comparison(self, text: str) -> str:
"""Normalize text for comparison with Unicode accent handling"""
if not text:
return ""
# Try to use unidecode for accent normalization, fallback to basic if not available
try:
from unidecode import unidecode
# Convert accents: é→e, ñ→n, ü→u, etc.
normalized = unidecode(text)
except ImportError:
# Fallback: basic normalization without accent handling
normalized = text
logger.warning("unidecode not available, accent matching may be limited")
# Convert to lowercase and strip
return normalized.lower().strip()
"""Delegates to `core.text.normalize.normalize_for_comparison`.
Kept as an instance method so existing internal callers don't need
to be touched — new code should import the public helper directly.
"""
from core.text.normalize import normalize_for_comparison
return normalize_for_comparison(text)
def _calculate_track_confidence(self, search_title: str, search_artist: str, db_track: DatabaseTrack) -> float:
"""Calculate confidence score for track match with enhanced cleaning and Unicode normalization"""

80
services/sync_service.py
View file

@ -10,6 +10,24 @@ from core.matching_engine import MusicMatchingEngine, MatchResult
logger = get_logger("sync_service")
# Per-artist track pool cap. High enough that no plausible artist hits it
# (the largest catalogs in our test libraries sit in the low thousands), low
# enough to avoid pathological pulls if the DB ever returns garbage.
_POOL_FETCH_LIMIT = 10000
def _artist_name(artist) -> str:
"""Pull the display name out of a Spotify artist entry — they come back
as bare strings on some endpoints and ``{name: ...}`` dicts on others.
Falls back to ``str(artist)`` so callers never get None."""
if isinstance(artist, str):
return artist
if isinstance(artist, dict):
name = artist.get('name')
if isinstance(name, str):
return name
return str(artist) if artist is not None else ''
@dataclass
class SyncResult:
playlist_name: str
@ -209,17 +227,12 @@ class PlaylistSyncService:
return self._create_error_result(playlist.name, ["Sync cancelled"])
total_tracks = len(playlist.tracks)
media_client, server_type = self._get_active_media_client()
media_client, server_type = self._get_active_media_client()
self._update_progress(playlist.name, f"Matching tracks against {server_type.title()} library", "", 20, 5, 2, total_tracks=total_tracks)
# Per-artist track pool, populated lazily inside _find_track_in_media_server.
# Only tracks that miss the sync_match_cache fast-path trigger a pool fetch
# for their artist — so warm-cache playlists pay zero pool cost. Misses
# for the same artist later in the playlist reuse the cached list and skip
# the per-variation SQL grid in check_track_exists. Empty dict (not None)
# to signal that pooling is enabled for this sync.
# Empty dict (not None) signals pooling is enabled for this sync;
# entries are filled lazily by `_find_track_in_media_server` so warm
# caches pay zero pool cost.
candidate_pool: Dict[str, list] = {}
# Use the same robust matching approach as "Download Missing Tracks"
@ -230,11 +243,8 @@ class PlaylistSyncService:
# Update progress for each track
progress_percent = 20 + (40 * (i + 1) / total_tracks) # 20-60% for matching
# Extract artist name from both string and dict formats
if track.artists:
first_artist = track.artists[0]
artist_name = first_artist if isinstance(first_artist, str) else (first_artist.get('name', 'Unknown') if isinstance(first_artist, dict) else str(first_artist))
current_track_name = f"{artist_name} - {track.name}"
current_track_name = f"{_artist_name(track.artists[0]) or 'Unknown'} - {track.name}"
else:
current_track_name = track.name
self._update_progress(playlist.name, "Matching tracks", current_track_name, progress_percent, 5, 2,
@ -478,32 +488,38 @@ class PlaylistSyncService:
self._cancelled = False
def _get_or_fetch_artist_candidates(self, candidate_pool: Optional[Dict[str, list]], db, artist_name: str, active_server) -> Optional[list]:
"""Lazy per-artist pool fetch. Only fires SQL when a track for this artist
actually missed the sync_match_cache fast-path — playlists where every
track is cached pay zero pool cost. Returns the candidate list (possibly
empty) on success; returns None when pooling is disabled so the caller
falls back to the legacy per-track SQL loop.
"""
"""Lazy per-artist pool fetch. Returns None when pooling is off so the
caller falls back to the per-track SQL loop; otherwise returns the
cached list (possibly empty), fetching on first miss."""
if candidate_pool is None:
return None
key = db._normalize_for_comparison(artist_name)
from core.text.normalize import normalize_for_comparison
key = normalize_for_comparison(artist_name)
if key in candidate_pool:
return candidate_pool[key]
try:
candidates = db.search_tracks(
artist=artist_name,
limit=10000,
# Fast path — indexed artist_id lookup. Hits idx_artists_name +
# idx_tracks_artist_id, returns in milliseconds for both hits and
# misses. Handles the 90%+ exact-name case.
candidates = db.get_artist_tracks_indexed(
artist_name,
server_source=active_server,
limit=_POOL_FETCH_LIMIT,
)
# Cache the empty result too — same key is asked once per artist this
# sync, then never again. Empty list still triggers the batched path
# in check_track_exists, which short-circuits without firing SQL.
# Slow-path fallback only when fast path found nothing. Preserves
# recall for diacritic variants and `tracks.track_artist` features
# (compilations / soundtracks where the per-track artist differs
# from the album artist).
if not candidates:
candidates = db.search_tracks(
artist=artist_name,
limit=_POOL_FETCH_LIMIT,
server_source=active_server,
)
candidate_pool[key] = candidates or []
return candidate_pool[key]
except Exception as fetch_err:
logger.debug(f"Candidate pool fetch failed for '{artist_name}': {fetch_err}")
# Don't cache the failure — let a later artist for the same key retry,
# and let this call's check_track_exists fall through to legacy SQL.
return None
async def _find_track_in_media_server(self, spotify_track: SpotifyTrack, candidate_pool: Optional[Dict[str, list]] = None) -> Tuple[Optional[TrackInfo], float]:
@ -568,14 +584,8 @@ class PlaylistSyncService:
if self._cancelled:
return None, 0.0
# Extract artist name from both string and dict formats
if isinstance(artist, str):
artist_name = artist
elif isinstance(artist, dict) and 'name' in artist:
artist_name = artist['name']
else:
artist_name = str(artist)
artist_name = _artist_name(artist)
# Use the improved database check_track_exists method with server awareness
try:
db = MusicDatabase()

127
tests/database/test_get_artist_tracks_indexed.py Normal file
View file

@ -0,0 +1,127 @@
"""Tests for `MusicDatabase.get_artist_tracks_indexed` — the indexed
two-step lookup that backs the sync candidate pool fast path."""
from __future__ import annotations
from database.music_database import MusicDatabase
def _seed(db: MusicDatabase, rows):
"""Insert (artist_name, album_title, track_title, server_source) tuples.
IDs are TEXT PRIMARY KEY in this schema so we hand-mint string IDs to
keep foreign-key wiring happy."""
conn = db._get_connection()
cursor = conn.cursor()
artist_ids: dict = {}
album_ids: dict = {}
track_counter = 0
for artist_name, album_title, track_title, server_source in rows:
if artist_name not in artist_ids:
aid = f"a-{len(artist_ids) + 1}"
cursor.execute(
"INSERT INTO artists (id, name, server_source) VALUES (?, ?, ?)",
(aid, artist_name, server_source),
)
artist_ids[artist_name] = aid
album_key = (artist_name, album_title, server_source)
if album_key not in album_ids:
alid = f"al-{len(album_ids) + 1}"
cursor.execute(
"INSERT INTO albums (id, artist_id, title, server_source) VALUES (?, ?, ?, ?)",
(alid, artist_ids[artist_name], album_title, server_source),
)
album_ids[album_key] = alid
track_counter += 1
cursor.execute(
"INSERT INTO tracks (id, album_id, artist_id, title, server_source) VALUES (?, ?, ?, ?, ?)",
(f"t-{track_counter}", album_ids[album_key], artist_ids[artist_name], track_title, server_source),
)
conn.commit()
def test_exact_name_match_returns_tracks(tmp_path):
db = MusicDatabase(str(tmp_path / "music.db"))
_seed(db, [
('Drake', 'For All The Dogs', 'First Person Shooter', 'plex'),
('Drake', 'For All The Dogs', 'Slime You Out', 'plex'),
('SZA', 'SOS', 'Kill Bill', 'plex'),
])
tracks = db.get_artist_tracks_indexed('Drake')
titles = sorted(t.title for t in tracks)
assert titles == ['First Person Shooter', 'Slime You Out']
def test_case_insensitive_fallback_finds_artist(tmp_path):
"""Exact match misses 'DRAKE' (case-sensitive index lookup), but the
fallback LOWER() comparison still finds the canonical 'Drake' row."""
db = MusicDatabase(str(tmp_path / "music.db"))
_seed(db, [
('Drake', 'FATD', 'IDGAF', 'plex'),
])
tracks = db.get_artist_tracks_indexed('DRAKE')
assert len(tracks) == 1
assert tracks[0].title == 'IDGAF'
def test_artist_absent_returns_empty_list(tmp_path):
"""Genuinely missing artists must fall straight through both steps
and return [] — that's what lets the caller skip the slow LIKE
fallback when the artist isn't in the library at all."""
db = MusicDatabase(str(tmp_path / "music.db"))
_seed(db, [
('Drake', 'FATD', 'IDGAF', 'plex'),
])
assert db.get_artist_tracks_indexed('Nonexistent Artist') == []
def test_empty_name_returns_empty(tmp_path):
db = MusicDatabase(str(tmp_path / "music.db"))
assert db.get_artist_tracks_indexed('') == []
def test_server_source_filter_excludes_other_servers(tmp_path):
"""The pool is per-server — Plex sync must not see Jellyfin tracks
even when the artist exists on both."""
db = MusicDatabase(str(tmp_path / "music.db"))
_seed(db, [
('Drake', 'Plex Album', 'Plex Track', 'plex'),
('Drake', 'Jellyfin Album', 'Jellyfin Track', 'jellyfin'),
])
plex_tracks = db.get_artist_tracks_indexed('Drake', server_source='plex')
jellyfin_tracks = db.get_artist_tracks_indexed('Drake', server_source='jellyfin')
assert [t.title for t in plex_tracks] == ['Plex Track']
assert [t.title for t in jellyfin_tracks] == ['Jellyfin Track']
def test_no_server_filter_returns_all_servers(tmp_path):
db = MusicDatabase(str(tmp_path / "music.db"))
_seed(db, [
('Drake', 'Plex Album', 'Plex Track', 'plex'),
('Drake', 'Jellyfin Album', 'Jellyfin Track', 'jellyfin'),
])
tracks = db.get_artist_tracks_indexed('Drake')
titles = sorted(t.title for t in tracks)
assert titles == ['Jellyfin Track', 'Plex Track']
def test_limit_caps_result_set(tmp_path):
db = MusicDatabase(str(tmp_path / "music.db"))
_seed(db, [('Drake', 'Album', f'Track {i}', 'plex') for i in range(10)])
tracks = db.get_artist_tracks_indexed('Drake', limit=3)
assert len(tracks) == 3
def test_returned_tracks_carry_artist_and_album_fields(tmp_path):
"""check_track_exists' batched path reads `artist_name` and
`album_title` off each track for confidence scoring — verify the
indexed query attaches them like search_tracks does."""
db = MusicDatabase(str(tmp_path / "music.db"))
_seed(db, [
('Drake', 'For All The Dogs', 'First Person Shooter', 'plex'),
])
tracks = db.get_artist_tracks_indexed('Drake')
assert len(tracks) == 1
t = tracks[0]
assert t.artist_name == 'Drake'
assert t.album_title == 'For All The Dogs'
assert t.title == 'First Person Shooter'

43
tests/sync/test_artist_name_extraction.py Normal file
View file

@ -0,0 +1,43 @@
"""Tests for `_artist_name` in services/sync_service.py — the helper
that pulls a string name out of Spotify's bare-string / dict / fallback
artist representations."""
from services.sync_service import _artist_name
def test_bare_string_returned_as_is():
assert _artist_name('Drake') == 'Drake'
def test_dict_with_name_field():
assert _artist_name({'name': 'Drake', 'id': '3TVXt'}) == 'Drake'
def test_dict_without_name_field_falls_back_to_str_repr():
"""Missing name field shouldn't crash — caller should still get a
string back, even if it's the awkward dict repr."""
out = _artist_name({'id': '3TVXt'})
assert isinstance(out, str)
assert out != ''
def test_dict_with_non_string_name_falls_back():
"""Defensive — if some endpoint ever returns {name: None} or a list,
the helper must not propagate the bad type."""
out = _artist_name({'name': None})
assert isinstance(out, str)
def test_none_returns_empty_string():
assert _artist_name(None) == ''
def test_unexpected_type_returns_string_repr():
"""A weird type (int, custom object) must coerce to a string instead
of raising — sync iterates a lot of inputs and one bad row shouldn't
crash the whole loop."""
assert _artist_name(12345) == '12345'
def test_empty_string_stays_empty():
assert _artist_name('') == ''

79
tests/sync/test_sync_candidate_pool.py
View file

@ -28,15 +28,19 @@ def _make_service() -> PlaylistSyncService:
)
def _make_db_stub(search_returns=None, raise_on_search=None) -> MagicMock:
def _make_db_stub(indexed_returns=None, search_returns=None, raise_on_search=None) -> MagicMock:
"""MusicDatabase stub mirroring the contract the helper relies on:
- _normalize_for_comparison returns a lower-cased key
- search_tracks returns a list (or raises)
- get_artist_tracks_indexed is the fast path (indexed artist_id lookup)
- search_tracks is the slow LIKE-based fallback for recall edge cases
Pool-key normalization runs through `core.text.normalize` directly,
not through the db, so no `_normalize_for_comparison` stub is needed.
"""
db = MagicMock()
db._normalize_for_comparison.side_effect = lambda s: s.lower().strip()
db.get_artist_tracks_indexed.return_value = indexed_returns if indexed_returns is not None else []
if raise_on_search is not None:
db.search_tracks.side_effect = raise_on_search
db.get_artist_tracks_indexed.side_effect = raise_on_search
else:
db.search_tracks.return_value = search_returns if search_returns is not None else []
return db
@ -54,21 +58,43 @@ def test_returns_none_when_pool_disabled():
result = svc._get_or_fetch_artist_candidates(None, db, 'Drake', 'plex')
assert result is None
db.search_tracks.assert_not_called()
db.get_artist_tracks_indexed.assert_not_called()
# ---------------------------------------------------------------------------
# Lazy population
# ---------------------------------------------------------------------------
def test_first_call_for_artist_runs_search_and_caches():
def test_indexed_fast_path_hits_skip_the_like_fallback():
"""When the indexed lookup finds tracks, the LIKE-based fallback must
NOT run — that's the whole perf point of the fast path."""
svc = _make_service()
pool: dict = {}
db = _make_db_stub(search_returns=['t1', 't2'])
db = _make_db_stub(indexed_returns=['t1', 't2'])
result = svc._get_or_fetch_artist_candidates(pool, db, 'Drake', 'plex')
assert result == ['t1', 't2']
assert pool == {'drake': ['t1', 't2']}
db.get_artist_tracks_indexed.assert_called_once_with(
'Drake', server_source='plex', limit=10000,
)
db.search_tracks.assert_not_called()
def test_like_fallback_runs_when_indexed_returns_empty():
"""Diacritics / featured-artist recall lives in the LIKE path. The
helper must fall through to search_tracks when the indexed lookup
finds nothing, otherwise sync regresses on those cases. Note that
the pool key is accent-folded (`Beyoncé` → `beyonce`) so library
spellings with/without diacritics share one entry."""
svc = _make_service()
pool: dict = {}
db = _make_db_stub(indexed_returns=[], search_returns=['feature-track'])
result = svc._get_or_fetch_artist_candidates(pool, db, 'Beyoncé', 'plex')
assert result == ['feature-track']
assert pool == {'beyonce': ['feature-track']}
db.get_artist_tracks_indexed.assert_called_once()
db.search_tracks.assert_called_once_with(
artist='Drake', limit=10000, server_source='plex',
artist='Beyoncé', limit=10000, server_source='plex',
)
@ -77,29 +103,31 @@ def test_second_call_for_same_artist_reuses_cache():
re-fetch — that's the whole perf point of the pool."""
svc = _make_service()
pool = {'drake': ['cached']}
db = _make_db_stub(search_returns=['fresh'])
db = _make_db_stub(indexed_returns=['fresh'], search_returns=['stale'])
result = svc._get_or_fetch_artist_candidates(pool, db, 'Drake', 'plex')
assert result == ['cached']
db.get_artist_tracks_indexed.assert_not_called()
db.search_tracks.assert_not_called()
def test_empty_result_is_still_cached():
"""Artist not in library → empty list cached. Next call short-circuits
via check_track_exists' batched path without firing SQL."""
def test_artist_absent_from_library_cached_as_empty_list():
"""Both paths return [] → cache [] so the next call short-circuits
via check_track_exists' batched path without firing SQL again."""
svc = _make_service()
pool: dict = {}
db = _make_db_stub(search_returns=[])
db = _make_db_stub(indexed_returns=[], search_returns=[])
result = svc._get_or_fetch_artist_candidates(pool, db, 'Obscure', 'plex')
assert result == []
assert pool == {'obscure': []}
def test_none_return_normalized_to_empty_list():
"""Defensive — if search_tracks ever returns None, helper must coerce
"""Defensive — if both paths ever return None, helper must coerce
to [] so the cached value is still a valid iterable for the matcher."""
svc = _make_service()
pool: dict = {}
db = _make_db_stub()
db.get_artist_tracks_indexed.return_value = None
db.search_tracks.return_value = None
result = svc._get_or_fetch_artist_candidates(pool, db, 'Anyone', 'plex')
assert result == []
@ -131,11 +159,13 @@ def test_pool_key_is_normalized_so_casing_variants_share_one_fetch():
a playlist that mixes casing would re-fetch the same artist twice."""
svc = _make_service()
pool: dict = {}
db = _make_db_stub(search_returns=['t'])
db = _make_db_stub(indexed_returns=['t'])
svc._get_or_fetch_artist_candidates(pool, db, 'Drake', 'plex')
db.get_artist_tracks_indexed.reset_mock()
db.search_tracks.reset_mock()
result = svc._get_or_fetch_artist_candidates(pool, db, 'DRAKE', 'plex')
assert result == ['t']
db.get_artist_tracks_indexed.assert_not_called()
db.search_tracks.assert_not_called()
@ -143,24 +173,35 @@ def test_different_artists_get_separate_pool_entries():
svc = _make_service()
pool: dict = {}
db = _make_db_stub()
db.search_tracks.side_effect = [['drake-track'], ['sza-track']]
db.get_artist_tracks_indexed.side_effect = [['drake-track'], ['sza-track']]
svc._get_or_fetch_artist_candidates(pool, db, 'Drake', 'plex')
svc._get_or_fetch_artist_candidates(pool, db, 'SZA', 'plex')
assert pool == {'drake': ['drake-track'], 'sza': ['sza-track']}
assert db.search_tracks.call_count == 2
assert db.get_artist_tracks_indexed.call_count == 2
# ---------------------------------------------------------------------------
# Server source plumbing
# ---------------------------------------------------------------------------
def test_active_server_is_passed_through_to_search_tracks():
def test_active_server_is_passed_through_to_indexed_path():
"""Misrouting server_source would make the pool include tracks from
the wrong server (e.g. Plex tracks in a Jellyfin sync) — verify it
survives the trip."""
survives the trip on the fast path."""
svc = _make_service()
pool: dict = {}
db = _make_db_stub(search_returns=['t'])
db = _make_db_stub(indexed_returns=['t'])
svc._get_or_fetch_artist_candidates(pool, db, 'Drake', 'jellyfin')
db.get_artist_tracks_indexed.assert_called_once_with(
'Drake', server_source='jellyfin', limit=10000,
)
def test_active_server_is_passed_through_to_like_fallback():
"""Same server_source check for the slow LIKE-based fallback path."""
svc = _make_service()
pool: dict = {}
db = _make_db_stub(indexed_returns=[], search_returns=['t'])
svc._get_or_fetch_artist_candidates(pool, db, 'Drake', 'jellyfin')
db.search_tracks.assert_called_once_with(
artist='Drake', limit=10000, server_source='jellyfin',

0
tests/text/__init__.py Normal file
View file

38
tests/text/test_normalize.py Normal file
View file

@ -0,0 +1,38 @@
"""Tests for `core.text.normalize.normalize_for_comparison`."""
from core.text.normalize import normalize_for_comparison
def test_empty_input_returns_empty_string():
assert normalize_for_comparison("") == ""
assert normalize_for_comparison(None) == "" # type: ignore[arg-type]
def test_lowercases_ascii():
assert normalize_for_comparison("Drake") == "drake"
assert normalize_for_comparison("DRAKE") == "drake"
def test_strips_surrounding_whitespace():
assert normalize_for_comparison(" Drake ") == "drake"
assert normalize_for_comparison("\tDrake\n") == "drake"
def test_folds_accents_to_ascii():
"""Diacritic-different spellings of the same artist must collapse to
one normalized key — otherwise the pool would re-fetch the same
artist when the playlist and library disagree on casing/accents."""
assert normalize_for_comparison("Beyoncé") == "beyonce"
assert normalize_for_comparison("Björk") == "bjork"
assert normalize_for_comparison("Subcarpaţi") == "subcarpati"
def test_combines_lowercase_and_accent_folding():
assert normalize_for_comparison("BEYONCÉ") == "beyonce"
def test_preserves_internal_whitespace():
"""Multi-word artist names must keep their internal spacing — only
leading/trailing whitespace is stripped."""
assert normalize_for_comparison("Bon Iver") == "bon iver"
assert normalize_for_comparison("Tame Impala") == "tame impala"