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

Merge pull request #330 from JohnBaumb/fix/performance-and-db-optimization

Browse source 
Performance and database optimization - reduce redundant queries and unbounded responses
This commit is contained in:
BoulderBadgeDad 2026-04-19 17:38:20 -07:00 committed by GitHub
commit be2d0569da
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 2054 additions and 162 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api/__init__.py
View file

@ -43,6 +43,7 @@ def create_api_blueprint():
from .listenbrainz import register_routes as reg_listenbrainz
from .cache import register_routes as reg_cache
from .request import register_routes as reg_request
from .request import start_cleanup_thread as _start_request_cleanup
# ---- rate-limit only /api/v1 routes (not the whole app) ----
limiter.limit("60 per minute")(bp)
@ -62,6 +63,11 @@ def create_api_blueprint():
reg_cache(bp)
reg_request(bp)
# Start the periodic cleanup timer for in-memory request tracking so
# idle periods don't leave stale entries in memory. Idempotent across
# calls; safe with multi-blueprint registration.
_start_request_cleanup()
# ---- error handlers (scoped to this Blueprint) ----
@bp.errorhandler(400)
def _bad_request(e):

33
api/auth.py
View file

@ -4,8 +4,9 @@ API key authentication for the SoulSync public API.
import hashlib
import secrets
import threading
import uuid
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from functools import wraps
from flask import request, current_app
@ -13,6 +14,27 @@ from flask import request, current_app
from .helpers import api_error
# Throttle persistence of `last_used_at` so every authenticated request
# does not rewrite the full app config. Maps key_hash -> last-persisted datetime.
_USAGE_WRITE_INTERVAL = timedelta(minutes=15)
_last_persisted_usage: dict[str, datetime] = {}
_usage_lock = threading.Lock()
def _should_persist_usage(key_hash: str, now: datetime) -> bool:
"""Return True if `last_used_at` for the given key should be written to disk.
Thread-safe: tracks the last write per key hash in memory and only returns
True once per `_USAGE_WRITE_INTERVAL`.
"""
with _usage_lock:
previous = _last_persisted_usage.get(key_hash)
if previous is None or (now - previous) >= _USAGE_WRITE_INTERVAL:
_last_persisted_usage[key_hash] = now
return True
return False
def generate_api_key(label=""):
"""Generate a new API key.
@ -67,9 +89,12 @@ def require_api_key(f):
if not matched:
return api_error("INVALID_KEY", "Invalid API key.", 403)
# Update last-used timestamp (best-effort)
matched["last_used_at"] = datetime.now(timezone.utc).isoformat()
config_mgr.set("api_keys", stored_keys)
# Update last-used timestamp (best-effort, throttled to avoid rewriting
# the full app config on every authenticated request).
now = datetime.now(timezone.utc)
matched["last_used_at"] = now.isoformat()
if _should_persist_usage(key_hash, now):
config_mgr.set("api_keys", stored_keys)
return f(*args, **kwargs)

64
api/downloads.py
View file

@ -40,16 +40,68 @@ def register_routes(bp):
@bp.route("/downloads", methods=["GET"])
@require_api_key
def list_downloads():
"""List active and recent download tasks."""
"""List download tasks with optional filtering and pagination.
Query params:
status: comma-separated statuses to include (e.g. "downloading,queued").
Default includes all.
limit: max tasks to return (default 100, max 500).
offset: skip the first N tasks (default 0).
Response includes `total` (post-filter count) so clients can paginate
without fetching everything. Tasks are sorted by `status_change_time`
descending so newest/in-flight tasks appear first.
"""
try:
from web_server import download_tasks, tasks_lock
tasks = []
with tasks_lock:
for task_id, task in download_tasks.items():
tasks.append(_serialize_download(task_id, task))
# Parse pagination params
try:
limit = int(request.args.get("limit", 100))
except (TypeError, ValueError):
limit = 100
try:
offset = int(request.args.get("offset", 0))
except (TypeError, ValueError):
offset = 0
# Clamp to sensible bounds
limit = max(1, min(limit, 500))
offset = max(0, offset)
return api_success({"downloads": tasks})
status_param = request.args.get("status", "").strip()
status_filter = (
{s.strip() for s in status_param.split(",") if s.strip()}
if status_param
else None
)
# Snapshot under the lock, then sort/slice outside.
with tasks_lock:
snapshot = list(download_tasks.items())
if status_filter:
snapshot = [
(tid, t) for tid, t in snapshot
if (t.get("status") or "") in status_filter
]
# Sort newest-first by status_change_time; fall back to string id
# so ordering is stable when timestamps are missing or tied.
snapshot.sort(
key=lambda item: (item[1].get("status_change_time") or "", item[0]),
reverse=True,
)
total = len(snapshot)
page = snapshot[offset:offset + limit]
tasks = [_serialize_download(tid, t) for tid, t in page]
return api_success({
"downloads": tasks,
"total": total,
"limit": limit,
"offset": offset,
})
except ImportError:
return api_error("NOT_AVAILABLE", "Download tracking not available.", 501)
except Exception as e:

11
api/library.py
View file

@ -186,15 +186,8 @@ def register_routes(bp):
try:
db = get_database()
tracks = db.search_tracks(title=title, artist=artist, limit=limit)
if not tracks:
return api_success({"tracks": []})
# Re-query by IDs to get full row data
track_ids = [t.id for t in tracks]
full_tracks = db.api_get_tracks_by_ids(track_ids)
return api_success({"tracks": [serialize_track(t, fields) for t in full_tracks]})
tracks = db.api_search_tracks(title=title, artist=artist, limit=limit)
return api_success({"tracks": [serialize_track(t, fields) for t in tracks]})
except Exception as e:
return api_error("LIBRARY_ERROR", str(e), 500)

61
api/request.py
View file

@ -23,6 +23,16 @@ _requests_lock = threading.Lock()
# Max age before auto-cleanup
_MAX_REQUEST_AGE = timedelta(hours=1)
# How often the background cleanup timer runs. Short enough to keep memory
# bounded during idle periods, long enough that slow-polling external clients
# still see their request for close to the TTL.
_CLEANUP_INTERVAL_SECONDS = 300 # 5 minutes
# Guards for the singleton background cleanup thread.
_cleanup_thread: "threading.Thread | None" = None
_cleanup_stop_event = threading.Event()
_cleanup_thread_lock = threading.Lock()
def _cleanup_old_requests():
"""Remove requests older than 1 hour to prevent unbounded growth."""
@ -32,6 +42,57 @@ def _cleanup_old_requests():
if r.get('created_at', datetime.now()) < cutoff]
for rid in expired:
del _pending_requests[rid]
return len(expired) if expired else 0
def _cleanup_loop():
"""Background thread: periodically evict expired requests."""
while not _cleanup_stop_event.is_set():
# wait() returns True if the event was set (shutdown), False on timeout
if _cleanup_stop_event.wait(timeout=_CLEANUP_INTERVAL_SECONDS):
return
try:
removed = _cleanup_old_requests()
if removed:
logger.debug(f"Request cleanup: evicted {removed} stale entries")
except Exception as e:
logger.warning(f"Request cleanup loop error: {e}")
def start_cleanup_thread() -> bool:
"""Start the background cleanup timer once per process.
Returns True if a new thread was started, False if one was already
running. Safe to call multiple times; callers in multi-worker setups
should still gate on worker identity if they want exactly one thread
across the entire deployment.
"""
global _cleanup_thread
with _cleanup_thread_lock:
if _cleanup_thread is not None and _cleanup_thread.is_alive():
return False
_cleanup_stop_event.clear()
_cleanup_thread = threading.Thread(
target=_cleanup_loop,
name="api-request-cleanup",
daemon=True,
)
_cleanup_thread.start()
logger.info("Started api/request cleanup timer (interval=%ss)" % _CLEANUP_INTERVAL_SECONDS)
return True
def stop_cleanup_thread(timeout: float = 2.0) -> None:
"""Signal the cleanup thread to exit. Used in tests and shutdown paths."""
global _cleanup_thread
with _cleanup_thread_lock:
thread = _cleanup_thread
_cleanup_stop_event.set()
if thread is not None and thread.is_alive():
thread.join(timeout=timeout)
with _cleanup_thread_lock:
_cleanup_thread = None
_cleanup_stop_event.clear()
def _run_search_and_download(request_id, query, notify_url):

34
api/wishlist.py
View file

@ -19,21 +19,19 @@ def register_routes(bp):
fields = parse_fields(request)
profile_id = parse_profile_id(request)
category_filter = category if category in ("singles", "albums") else None
try:
from database.music_database import get_database
db = get_database()
raw_tracks = db.get_wishlist_tracks(profile_id=profile_id)
# Category filter
if category in ("singles", "albums"):
raw_tracks = [
t for t in raw_tracks
if _track_category(t) == category
]
total = len(raw_tracks)
start = (page - 1) * limit
tracks = raw_tracks[start:start + limit]
offset = (page - 1) * limit
tracks = db.get_wishlist_tracks(
profile_id=profile_id,
category=category_filter,
limit=limit,
offset=offset,
)
total = db.get_wishlist_count(profile_id=profile_id, category=category_filter)
return api_success(
{"tracks": [serialize_wishlist_track(t, fields) for t in tracks]},
@ -104,15 +102,3 @@ def register_routes(bp):
return api_error("NOT_AVAILABLE", "Wishlist processing function not available.", 501)
except Exception as e:
return api_error("WISHLIST_ERROR", str(e), 500)
def _track_category(track):
"""Determine if a wishlist track is a single or album track."""
album_type = ""
if isinstance(track, dict):
sd = track.get("spotify_data", {})
if isinstance(sd, dict):
album = sd.get("album", {})
if isinstance(album, dict):
album_type = album.get("album_type", "")
return "albums" if album_type == "album" else "singles"

20
core/lastfm_worker.py
View file

@ -307,7 +307,12 @@ class LastFMWorker:
logger.error(f"Error updating item status: {e2}")
def _get_existing_id(self, entity_type: str, entity_id: int) -> Optional[str]:
"""Check if an entity already has a lastfm_id (e.g. from manual match)."""
"""Check if an entity already has a lastfm_url (e.g. from manual match).
The Last.fm schema uses `lastfm_url` (not `lastfm_id`) on artists, albums,
and tracks. This helper always returned None before because it queried a
non-existent column and silently caught the exception.
"""
table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'}
table = table_map.get(entity_type)
if not table:
@ -316,7 +321,7 @@ class LastFMWorker:
try:
conn = self.db._get_connection()
cursor = conn.cursor()
cursor.execute(f"SELECT lastfm_id FROM {table} WHERE id = ?", (entity_id,))
cursor.execute(f"SELECT lastfm_url FROM {table} WHERE id = ?", (entity_id,))
row = cursor.fetchone()
return row[0] if row and row[0] else None
except Exception:
@ -329,7 +334,10 @@ class LastFMWorker:
"""Process an artist: get full info from Last.fm"""
existing_id = self._get_existing_id('artist', artist_id)
if existing_id:
logger.debug(f"Preserving existing Last.fm ID for artist '{artist_name}': {existing_id}")
# Row already has a Last.fm URL but status is NULL (legacy/manual match).
# Mark as matched so the worker does not re-select it forever.
logger.debug(f"Preserving existing Last.fm URL for artist '{artist_name}': {existing_id}")
self._mark_status('artist', artist_id, 'matched')
return
# Use get_artist_info for detailed data (includes stats, bio, tags, similar)
@ -353,7 +361,8 @@ class LastFMWorker:
"""Process an album: get full info from Last.fm"""
existing_id = self._get_existing_id('album', album_id)
if existing_id:
logger.debug(f"Preserving existing Last.fm ID for album '{album_name}': {existing_id}")
logger.debug(f"Preserving existing Last.fm URL for album '{album_name}': {existing_id}")
self._mark_status('album', album_id, 'matched')
return
result = self.client.get_album_info(artist_name, album_name)
@ -376,7 +385,8 @@ class LastFMWorker:
"""Process a track: get full info from Last.fm"""
existing_id = self._get_existing_id('track', track_id)
if existing_id:
logger.debug(f"Preserving existing Last.fm ID for track '{track_name}': {existing_id}")
logger.debug(f"Preserving existing Last.fm URL for track '{track_name}': {existing_id}")
self._mark_status('track', track_id, 'matched')
return
result = self.client.get_track_info(artist_name, track_name)

267
core/listening_stats_worker.py
View file

@ -180,12 +180,18 @@ class ListeningStatsWorker:
'played_at': entry.get('played_at'),
'duration_ms': entry.get('duration_ms', 0),
'server_source': active_server,
'db_track_id': self._resolve_db_track_id(
entry.get('track_title', ''),
entry.get('artist', '')
),
# db_track_id filled in below by a single batched lookup
'db_track_id': None,
})
# Batch-resolve track IDs for all events at once (was N+1 before).
id_map = self._resolve_db_track_ids_batch(events)
for ev in events:
title_l = (ev.get('title') or '').strip().lower()
artist_l = (ev.get('artist') or '').strip().lower()
if title_l:
ev['db_track_id'] = id_map.get((title_l, artist_l))
inserted = self.db.insert_listening_events(events)
self.stats['events_added'] += inserted
logger.info(f"Inserted {inserted} new listening events (of {len(events)} total)")
@ -269,59 +275,129 @@ class ListeningStatsWorker:
logger.error(f"Failed to build stats cache: {e}")
def _enrich_stats_items(self, cache):
"""Add image URLs, IDs, and Last.fm data to cached stats items."""
"""Add image URLs, IDs, and Last.fm data to cached stats items.
Previously ran one SELECT per artist / album / track entry. Now each
of the three lists is resolved with a single batched IN query so
cache rebuilds scale with the number of result sets, not with the
number of items in them.
"""
top_artists = cache.get('top_artists') or []
top_albums = cache.get('top_albums') or []
top_tracks = cache.get('top_tracks') or []
if not (top_artists or top_albums or top_tracks):
return
conn = None
try:
conn = self.db._get_connection()
cursor = conn.cursor()
for artist in (cache.get('top_artists') or []):
try:
cursor.execute("""
SELECT thumb_url, id, lastfm_listeners, lastfm_playcount, soul_id
FROM artists WHERE LOWER(name) = LOWER(?) LIMIT 1
""", (artist['name'],))
r = cursor.fetchone()
if r:
artist['image_url'] = r[0] or None
artist['id'] = r[1]
artist['global_listeners'] = r[2]
artist['global_playcount'] = r[3]
artist['soul_id'] = r[4]
except Exception:
pass
# ---- top_artists: match by LOWER(name) ----
if top_artists:
names = [a.get('name') or '' for a in top_artists]
unique_names = {n.lower() for n in names if n}
artist_rows = {}
if unique_names:
name_list = list(unique_names)
chunk = 500
for i in range(0, len(name_list), chunk):
sub = name_list[i:i + chunk]
placeholders = ','.join(['?'] * len(sub))
cursor.execute(
f"""
SELECT LOWER(name), thumb_url, id, lastfm_listeners,
lastfm_playcount, soul_id
FROM artists
WHERE LOWER(name) IN ({placeholders})
""",
sub,
)
for row in cursor.fetchall():
# Keep first match per lowered name (LIMIT 1 equiv).
artist_rows.setdefault(row[0], row)
for album in (cache.get('top_albums') or []):
try:
cursor.execute("""
SELECT al.thumb_url, al.id, al.artist_id FROM albums al
WHERE LOWER(al.title) = LOWER(?) LIMIT 1
""", (album['name'],))
r = cursor.fetchone()
for artist in top_artists:
key = (artist.get('name') or '').lower()
r = artist_rows.get(key)
if r:
album['image_url'] = r[0] or None
album['id'] = r[1]
album['artist_id'] = r[2]
except Exception:
pass
artist['image_url'] = r[1] or None
artist['id'] = r[2]
artist['global_listeners'] = r[3]
artist['global_playcount'] = r[4]
artist['soul_id'] = r[5]
for track in (cache.get('top_tracks') or []):
try:
cursor.execute("""
SELECT al.thumb_url, t.id, t.artist_id FROM tracks t
JOIN albums al ON al.id = t.album_id
JOIN artists ar ON ar.id = t.artist_id
WHERE LOWER(t.title) = LOWER(?) AND LOWER(ar.name) = LOWER(?) LIMIT 1
""", (track['name'], track.get('artist', '')))
r = cursor.fetchone()
# ---- top_albums: match by LOWER(title) ----
if top_albums:
titles = [a.get('name') or '' for a in top_albums]
unique_titles = {t.lower() for t in titles if t}
album_rows = {}
if unique_titles:
title_list = list(unique_titles)
chunk = 500
for i in range(0, len(title_list), chunk):
sub = title_list[i:i + chunk]
placeholders = ','.join(['?'] * len(sub))
cursor.execute(
f"""
SELECT LOWER(title), thumb_url, id, artist_id
FROM albums
WHERE LOWER(title) IN ({placeholders})
""",
sub,
)
for row in cursor.fetchall():
album_rows.setdefault(row[0], row)
for album in top_albums:
key = (album.get('name') or '').lower()
r = album_rows.get(key)
if r:
track['image_url'] = r[0] or None
track['id'] = r[1]
track['artist_id'] = r[2]
except Exception:
pass
except Exception:
pass
album['image_url'] = r[1] or None
album['id'] = r[2]
album['artist_id'] = r[3]
# ---- top_tracks: match by (LOWER(title), LOWER(artist name)) ----
if top_tracks:
pairs = set()
for t in top_tracks:
name = (t.get('name') or '').lower()
artist = (t.get('artist') or '').lower()
if name:
pairs.add((name, artist))
track_rows = {}
if pairs:
pair_list = list(pairs)
chunk = 500
for i in range(0, len(pair_list), chunk):
sub = pair_list[i:i + chunk]
placeholders = ','.join(['(?,?)'] * len(sub))
flat = [v for pair in sub for v in pair]
cursor.execute(
f"""
SELECT LOWER(t.title), LOWER(ar.name),
al.thumb_url, t.id, t.artist_id
FROM tracks t
JOIN albums al ON al.id = t.album_id
JOIN artists ar ON ar.id = t.artist_id
WHERE (LOWER(t.title), LOWER(ar.name)) IN ({placeholders})
""",
flat,
)
for row in cursor.fetchall():
track_rows.setdefault((row[0], row[1]), row)
for track in top_tracks:
key = ((track.get('name') or '').lower(),
(track.get('artist') or '').lower())
r = track_rows.get(key)
if r:
track['image_url'] = r[2] or None
track['id'] = r[3]
track['artist_id'] = r[4]
except Exception as e:
logger.error(f"Error enriching stats items: {e}")
finally:
if conn:
conn.close()
@ -454,30 +530,95 @@ class ListeningStatsWorker:
if conn:
conn.close()
def _resolve_db_track_ids_batch(self, events):
"""Batch-resolve DB track IDs for a list of history events.
Returns a dict ``{(title_lower, artist_lower): track_id}`` so callers
can look up without another DB round-trip. Replaces the former N+1
pattern of one SELECT per event (500 events = 500 queries).
Uses row-value IN with chunking (500 pairs = 1000 variables, well
under SQLite's default limit). Case-insensitive matching is preserved.
"""
pairs = set()
for ev in events:
title = (ev.get('title') or '').strip()
artist = (ev.get('artist') or '').strip()
if title:
pairs.add((title.lower(), artist.lower()))
result = {}
if not pairs:
return result
pair_list = list(pairs)
chunk_size = 500
conn = None
try:
conn = self.db._get_connection()
cursor = conn.cursor()
for i in range(0, len(pair_list), chunk_size):
chunk = pair_list[i:i + chunk_size]
placeholders = ','.join(['(?,?)'] * len(chunk))
flat_args = [v for pair in chunk for v in pair]
cursor.execute(
f"""
SELECT LOWER(t.title), LOWER(ar.name), t.id
FROM tracks t
JOIN artists ar ON ar.id = t.artist_id
WHERE (LOWER(t.title), LOWER(ar.name)) IN ({placeholders})
""",
flat_args,
)
for title_l, artist_l, tid in cursor.fetchall():
# Keep first match per pair to match the LIMIT 1 semantics
# of the original per-event query.
result.setdefault((title_l, artist_l), tid)
except Exception as e:
logger.error(f"Error batch-resolving track IDs: {e}")
finally:
if conn:
conn.close()
return result
def _map_play_counts_to_db(self, server_counts, server_source):
"""Map server track IDs to DB track IDs for play count updates.
Looks up tracks by matching the server's track ID stored in
the tracks table (from library sync).
Looks up which server IDs exist in the tracks table. Replaces a
previous N+1 pattern of one SELECT per server ID with a single
batched IN query (chunked for safety).
"""
if not server_counts:
return []
conn = None
try:
conn = self.db._get_connection()
cursor = conn.cursor()
# Build a lookup of server_id → db_track_id
# The tracks table stores server IDs as the primary 'id' column
updates = []
for server_id, play_count in server_counts.items():
cursor.execute("SELECT id FROM tracks WHERE id = ?", (server_id,))
row = cursor.fetchone()
if row:
updates.append({
'db_track_id': row[0],
'play_count': play_count,
'last_played': None, # Could be fetched separately
})
return updates
ids = list(server_counts.keys())
existing = set()
chunk_size = 500
for i in range(0, len(ids), chunk_size):
chunk = ids[i:i + chunk_size]
placeholders = ','.join(['?'] * len(chunk))
cursor.execute(
f"SELECT id FROM tracks WHERE id IN ({placeholders})",
chunk,
)
existing.update(r[0] for r in cursor.fetchall())
return [
{
'db_track_id': server_id,
'play_count': play_count,
'last_played': None, # Could be fetched separately
}
for server_id, play_count in server_counts.items()
if server_id in existing
]
except Exception as e:
logger.error(f"Error mapping play counts: {e}")
return []

28
core/metadata_cache.py
View file

@ -326,20 +326,28 @@ class MetadataCache:
""", (row['id'],))
conn.commit()
# Resolve entity IDs to full data
# Resolve entity IDs to full data via a single batched query
# (chunked to stay below SQLite's default variable limit).
result_ids = json.loads(row['result_ids'])
if not result_ids:
return []
results = []
for eid in result_ids:
cursor.execute("""
SELECT raw_json FROM metadata_cache_entities
WHERE source = ? AND entity_type = ? AND entity_id = ?
""", (source, search_type, eid))
erow = cursor.fetchone()
if erow:
results.append(json.loads(erow['raw_json']))
raw_by_id: Dict[str, dict] = {}
for i in range(0, len(result_ids), 500):
chunk = result_ids[i:i + 500]
placeholders = ','.join('?' * len(chunk))
cursor.execute(f"""
SELECT entity_id, raw_json FROM metadata_cache_entities
WHERE source = ? AND entity_type = ? AND entity_id IN ({placeholders})
""", [source, search_type, *chunk])
for erow in cursor.fetchall():
try:
raw_by_id[erow['entity_id']] = json.loads(erow['raw_json'])
except (ValueError, TypeError):
continue
# Preserve the original result_ids ordering.
results = [raw_by_id[eid] for eid in result_ids if eid in raw_by_id]
# Only return if we found all (or most) entries — partial results are unreliable
if len(results) >= len(result_ids) * 0.8:

32
core/musicbrainz_worker.py
View file

@ -254,16 +254,27 @@ class MusicBrainzWorker:
conn.close()
def _get_existing_id(self, entity_type: str, entity_id: int) -> Optional[str]:
"""Check if an entity already has a musicbrainz_id (e.g. from manual match)."""
table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'}
table = table_map.get(entity_type)
if not table:
"""Check if an entity already has a MusicBrainz ID (e.g. from manual match).
MusicBrainz ID columns differ per entity type: artists use `musicbrainz_id`,
albums use `musicbrainz_release_id`, and tracks use `musicbrainz_recording_id`.
Before this fix, all three were queried as `musicbrainz_id`, so the
existing-ID check silently failed for albums and tracks.
"""
table_config = {
'artist': ('artists', 'musicbrainz_id'),
'album': ('albums', 'musicbrainz_release_id'),
'track': ('tracks', 'musicbrainz_recording_id'),
}
cfg = table_config.get(entity_type)
if not cfg:
return None
table, column = cfg
conn = None
try:
conn = self.db._get_connection()
cursor = conn.cursor()
cursor.execute(f"SELECT musicbrainz_id FROM {table} WHERE id = ?", (entity_id,))
cursor.execute(f"SELECT {column} FROM {table} WHERE id = ?", (entity_id,))
row = cursor.fetchone()
return row[0] if row and row[0] else None
except Exception:
@ -285,6 +296,17 @@ class MusicBrainzWorker:
existing_id = self._get_existing_id(item_type, item_id)
if existing_id:
logger.debug(f"Preserving existing MusicBrainz ID for {item_type} '{item_name}': {existing_id}")
# Mark as matched so this row is not re-selected forever when
# match_status is NULL but the MBID is already populated.
try:
if item_type == 'artist':
self.mb_service.update_artist_mbid(item_id, existing_id, 'matched')
elif item_type == 'album':
self.mb_service.update_album_mbid(item_id, existing_id, 'matched')
elif item_type == 'track':
self.mb_service.update_track_mbid(item_id, existing_id, 'matched')
except Exception as mark_err:
logger.error(f"Error marking {item_type} #{item_id} matched: {mark_err}")
return
if item_type == 'artist':

4
core/qobuz_worker.py
View file

@ -388,6 +388,8 @@ class QobuzWorker:
existing_id = self._get_existing_id('artist', artist_id)
if existing_id:
logger.debug(f"Preserving existing Qobuz ID for artist '{artist_name}': {existing_id}")
# Mark as matched so this row is not re-selected on every loop.
self._mark_status('artist', artist_id, 'matched')
return
result = self.client.search_artist(artist_name)
@ -429,6 +431,7 @@ class QobuzWorker:
existing_id = self._get_existing_id('album', album_id)
if existing_id:
logger.debug(f"Preserving existing Qobuz ID for album '{album_name}': {existing_id}")
self._mark_status('album', album_id, 'matched')
return
result = self.client.search_album(artist_name, album_name)
@ -484,6 +487,7 @@ class QobuzWorker:
existing_id = self._get_existing_id('track', track_id)
if existing_id:
logger.debug(f"Preserving existing Qobuz ID for track '{track_name}': {existing_id}")
self._mark_status('track', track_id, 'matched')
return
result = self.client.search_track(artist_name, track_name)

4
core/tidal_worker.py
View file

@ -401,6 +401,8 @@ class TidalWorker:
existing_id = self._get_existing_id('artist', artist_id)
if existing_id:
logger.debug(f"Preserving existing Tidal ID for artist '{artist_name}': {existing_id}")
# Mark as matched so this row is not re-selected on every loop.
self._mark_status('artist', artist_id, 'matched')
return
result = self.client.search_artist(artist_name)
@ -438,6 +440,7 @@ class TidalWorker:
existing_id = self._get_existing_id('album', album_id)
if existing_id:
logger.debug(f"Preserving existing Tidal ID for album '{album_name}': {existing_id}")
self._mark_status('album', album_id, 'matched')
return
result = self.client.search_album(artist_name, album_name)
@ -486,6 +489,7 @@ class TidalWorker:
existing_id = self._get_existing_id('track', track_id)
if existing_id:
logger.debug(f"Preserving existing Tidal ID for track '{track_name}': {existing_id}")
self._mark_status('track', track_id, 'matched')
return
result = self.client.search_track(artist_name, track_name)

185
database/music_database.py
View file

@ -409,6 +409,12 @@ class MusicDatabase:
# Add Discogs enrichment columns (migration)
self._add_discogs_columns(cursor)
# Backfill match_status for rows that already have an external ID but
# NULL status. Prevents enrichment workers from re-processing these
# rows forever. Must run AFTER all *_match_status columns have been
# created by the migrations above.
self._backfill_match_status_for_existing_ids(cursor)
# Bubble snapshots table for persisting UI state across page refreshes
cursor.execute("""
CREATE TABLE IF NOT EXISTS bubble_snapshots (
@ -1922,6 +1928,56 @@ class MusicDatabase:
except Exception as e:
logger.error(f"Error adding Discogs columns: {e}")
def _backfill_match_status_for_existing_ids(self, cursor):
"""Set `<provider>_match_status = 'matched'` for rows that already have a
populated external ID but NULL match_status.
Prevents enrichment workers from re-selecting the same rows forever when
the ID was populated outside the worker (file tags, manual match,
pre-migration legacy data) without a corresponding status update.
Only runs columns that actually exist, so pre-migration databases are
handled safely. UPDATE statements are cheap no-ops when nothing matches.
"""
# (table, id_column, status_column)
targets = [
('artists', 'lastfm_url', 'lastfm_match_status'),
('albums', 'lastfm_url', 'lastfm_match_status'),
('tracks', 'lastfm_url', 'lastfm_match_status'),
('artists', 'musicbrainz_id', 'musicbrainz_match_status'),
('albums', 'musicbrainz_release_id', 'musicbrainz_match_status'),
('tracks', 'musicbrainz_recording_id', 'musicbrainz_match_status'),
('artists', 'tidal_id', 'tidal_match_status'),
('albums', 'tidal_id', 'tidal_match_status'),
('tracks', 'tidal_id', 'tidal_match_status'),
('artists', 'qobuz_id', 'qobuz_match_status'),
('albums', 'qobuz_id', 'qobuz_match_status'),
('tracks', 'qobuz_id', 'qobuz_match_status'),
]
total_backfilled = 0
for table, id_col, status_col in targets:
try:
cursor.execute(f"PRAGMA table_info({table})")
cols = {row[1] for row in cursor.fetchall()}
if id_col not in cols or status_col not in cols:
continue
cursor.execute(
f"UPDATE {table} SET {status_col} = 'matched' "
f"WHERE {status_col} IS NULL AND {id_col} IS NOT NULL AND {id_col} != ''"
)
if cursor.rowcount and cursor.rowcount > 0:
total_backfilled += cursor.rowcount
logger.info(
f"Backfilled {cursor.rowcount} rows in {table}.{status_col} "
f"where {id_col} was already set."
)
except Exception as e:
logger.error(f"Error backfilling {table}.{status_col}: {e}")
if total_backfilled == 0:
logger.debug("Match-status backfill: no rows needed updating.")
def _add_deezer_columns(self, cursor):
"""Add Deezer tracking + generic metadata columns for enrichment (artists, albums, tracks)"""
try:
@ -5063,12 +5119,41 @@ class MusicDatabase:
except Exception as e:
logger.error(f"Error searching tracks with title='{title}', artist='{artist}': {e}")
return []
def api_search_tracks(self, title: str = "", artist: str = "", limit: int = 50,
server_source: Optional[str] = None) -> List[Dict[str, Any]]:
"""Search tracks and return full dict rows (all track columns plus artist_name,
album_title, album_thumb_url). Avoids the double-query pattern of calling
search_tracks() followed by api_get_tracks_by_ids().
"""
try:
if not title and not artist:
return []
conn = self._get_connection()
cursor = conn.cursor()
basic_rows = self._search_tracks_basic_rows(cursor, title, artist, limit, server_source)
if basic_rows:
return [dict(r) for r in basic_rows]
fuzzy_rows = self._search_tracks_fuzzy_rows(cursor, title, artist, limit, server_source)
return [dict(r) for r in fuzzy_rows]
except Exception as e:
logger.error(f"API: Error searching tracks with title='{title}', artist='{artist}': {e}")
return []
def _search_tracks_basic(self, cursor, title: str, artist: str, limit: int, server_source: str = None) -> List[DatabaseTrack]:
"""Basic SQL LIKE search - fastest method"""
rows = self._search_tracks_basic_rows(cursor, title, artist, limit, server_source)
return self._rows_to_tracks(rows)
def _search_tracks_basic_rows(self, cursor, title: str, artist: str, limit: int,
server_source: Optional[str] = None):
"""Basic SQL LIKE search returning raw rows (shared by DatabaseTrack and dict-returning callers)."""
where_conditions = []
params = []
if title:
where_conditions.append("unidecode_lower(tracks.title) LIKE ?")
params.append(f"%{self._normalize_for_comparison(title)}%")
@ -5083,13 +5168,13 @@ class MusicDatabase:
if server_source:
where_conditions.append("tracks.server_source = ?")
params.append(server_source)
if not where_conditions:
return []
where_clause = " AND ".join(where_conditions)
params.append(limit)
cursor.execute(f"""
SELECT tracks.*, artists.name as artist_name, albums.title as album_title, albums.thumb_url as album_thumb_url
FROM tracks
@ -5100,45 +5185,47 @@ class MusicDatabase:
LIMIT ?
""", params)
return self._rows_to_tracks(cursor.fetchall())
return cursor.fetchall()
def _search_tracks_fuzzy_fallback(self, cursor, title: str, artist: str, limit: int, server_source: str = None) -> List[DatabaseTrack]:
"""Broadest fuzzy search - partial word matching"""
rows = self._search_tracks_fuzzy_rows(cursor, title, artist, limit, server_source)
return self._rows_to_tracks(rows)
def _search_tracks_fuzzy_rows(self, cursor, title: str, artist: str, limit: int,
server_source: Optional[str] = None):
"""Broadest fuzzy search returning raw rows (shared by DatabaseTrack and dict-returning callers)."""
# Get broader results by searching for individual words
search_terms = []
if title:
# Split title into words and search for each (normalized for diacritics)
title_words = [w.strip() for w in self._normalize_for_comparison(title).split() if len(w.strip()) >= 3]
search_terms.extend(title_words)
if artist:
# Split artist into words and search for each (normalized for diacritics)
artist_words = [w.strip() for w in self._normalize_for_comparison(artist).split() if len(w.strip()) >= 3]
search_terms.extend(artist_words)
if not search_terms:
return []
# Build a query that searches for any of the words
like_conditions = []
params = []
for term in search_terms[:5]: # Limit to 5 terms to avoid too broad search
for term in search_terms[:5]:
like_conditions.append("(unidecode_lower(tracks.title) LIKE ? OR unidecode_lower(artists.name) LIKE ? OR unidecode_lower(COALESCE(tracks.track_artist, '')) LIKE ?)")
params.extend([f"%{term}%", f"%{term}%", f"%{term}%"])
if not like_conditions:
return []
# Build WHERE clause with optional server filter
where_parts = [f"({' OR '.join(like_conditions)})"]
if server_source:
where_parts.append("tracks.server_source = ?")
params.append(server_source) # Append after LIKE params, before LIMIT
params.append(server_source)
where_clause = " AND ".join(where_parts)
params.append(limit * 3) # Get more results for scoring
params.append(limit * 3)
cursor.execute(f"""
SELECT tracks.*, artists.name as artist_name, albums.title as album_title, albums.thumb_url as album_thumb_url
FROM tracks
@ -5154,23 +5241,19 @@ class MusicDatabase:
# Score and filter results
scored_results = []
for row in rows:
# Simple scoring based on how many search terms match
score = 0
db_title_lower = self._normalize_for_comparison(row['title'])
db_artist_lower = self._normalize_for_comparison(row['artist_name'])
for term in search_terms:
if term in db_title_lower or term in db_artist_lower:
score += 1
if score > 0:
scored_results.append((score, row))
# Sort by score and take top results
scored_results.sort(key=lambda x: x[0], reverse=True)
top_rows = [row for score, row in scored_results[:limit]]
return self._rows_to_tracks(top_rows)
return [row for score, row in scored_results[:limit]]
def _rows_to_tracks(self, rows) -> List[DatabaseTrack]:
"""Convert database rows to DatabaseTrack objects"""
@ -6632,8 +6715,14 @@ class MusicDatabase:
logger.error(f"Error removing track from wishlist: {e}")
return False
def get_wishlist_tracks(self, limit: Optional[int] = None, profile_id: int = 1) -> List[Dict[str, Any]]:
"""Get all tracks in the wishlist for the given profile, ordered by date added (oldest first for retry priority)"""
def get_wishlist_tracks(self, limit: Optional[int] = None, profile_id: int = 1,
offset: int = 0, category: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get tracks in the wishlist for the given profile, ordered by date added
(oldest first for retry priority).
Supports SQL-level pagination via limit/offset and optional category
filtering (singles vs albums) pushed down to SQL using json_extract.
"""
try:
with self._get_connection() as conn:
cursor = conn.cursor()
@ -6643,12 +6732,26 @@ class MusicDatabase:
last_attempted, date_added, source_type, source_info
FROM wishlist_tracks
WHERE profile_id = ?
ORDER BY date_added
"""
params = [profile_id]
params: List[Any] = [profile_id]
if category == "albums":
query += " AND json_extract(spotify_data, '$.album.album_type') = 'album'"
elif category == "singles":
query += (
" AND (json_extract(spotify_data, '$.album.album_type') IS NULL"
" OR json_extract(spotify_data, '$.album.album_type') != 'album')"
)
query += " ORDER BY date_added"
if limit:
query += f" LIMIT {limit}"
query += " LIMIT ?"
params.append(int(limit))
if offset:
query += " OFFSET ?"
params.append(int(offset))
cursor.execute(query, params)
rows = cursor.fetchall()
@ -6679,7 +6782,7 @@ class MusicDatabase:
except Exception as e:
logger.error(f"Error getting wishlist tracks: {e}")
return []
def update_wishlist_retry(self, spotify_track_id: str, success: bool, error_message: str = None, profile_id: int = 1) -> bool:
"""Update retry count and status for a wishlist track"""
try:
@ -6706,12 +6809,22 @@ class MusicDatabase:
logger.error(f"Error updating wishlist retry status: {e}")
return False
def get_wishlist_count(self, profile_id: int = 1) -> int:
"""Get the total number of tracks in the wishlist for the given profile"""
def get_wishlist_count(self, profile_id: int = 1, category: Optional[str] = None) -> int:
"""Get the total number of tracks in the wishlist for the given profile,
optionally filtered by category ('singles' or 'albums')."""
try:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM wishlist_tracks WHERE profile_id = ?", (profile_id,))
query = "SELECT COUNT(*) FROM wishlist_tracks WHERE profile_id = ?"
params: List[Any] = [profile_id]
if category == "albums":
query += " AND json_extract(spotify_data, '$.album.album_type') = 'album'"
elif category == "singles":
query += (
" AND (json_extract(spotify_data, '$.album.album_type') IS NULL"
" OR json_extract(spotify_data, '$.album.album_type') != 'album')"
)
cursor.execute(query, params)
result = cursor.fetchone()
return result[0] if result else 0
except Exception as e:

110
tests/test_auth_usage_throttle.py Normal file
View file

@ -0,0 +1,110 @@
"""Unit tests for the auth last_used_at write throttle.
Fix 1.2: every authenticated API request previously called
`config_mgr.set("api_keys", ...)`, which rewrites the entire app config
blob to SQLite. Writes are now throttled per key hash.
"""
import sys
import threading
import types
from datetime import datetime, timedelta, timezone
import pytest
# api/__init__.py eagerly imports flask_limiter. Tests only need the auth
# module, so stub flask_limiter before importing the api package.
def _install_flask_limiter_stub():
if "flask_limiter" in sys.modules:
return
stub = types.ModuleType("flask_limiter")
class _Limiter:
def __init__(self, *args, **kwargs):
pass
def limit(self, *args, **kwargs):
def decorator(target):
return target
return decorator
def init_app(self, app):
pass
stub.Limiter = _Limiter
sys.modules["flask_limiter"] = stub
util_stub = types.ModuleType("flask_limiter.util")
util_stub.get_remote_address = lambda: "127.0.0.1"
sys.modules["flask_limiter.util"] = util_stub
_install_flask_limiter_stub()
from api import auth # noqa: E402
@pytest.fixture(autouse=True)
def _reset_usage_cache():
"""Ensure a clean throttle cache for each test."""
with auth._usage_lock:
auth._last_persisted_usage.clear()
yield
with auth._usage_lock:
auth._last_persisted_usage.clear()
def test_first_call_persists():
now = datetime.now(timezone.utc)
assert auth._should_persist_usage("hash-a", now) is True
def test_second_call_within_interval_does_not_persist():
start = datetime.now(timezone.utc)
assert auth._should_persist_usage("hash-a", start) is True
# 5 minutes later, still inside the 15-minute window
assert auth._should_persist_usage("hash-a", start + timedelta(minutes=5)) is False
def test_call_after_interval_persists_again():
start = datetime.now(timezone.utc)
assert auth._should_persist_usage("hash-a", start) is True
later = start + auth._USAGE_WRITE_INTERVAL
assert auth._should_persist_usage("hash-a", later) is True
def test_different_keys_have_independent_throttles():
now = datetime.now(timezone.utc)
assert auth._should_persist_usage("hash-a", now) is True
assert auth._should_persist_usage("hash-b", now) is True
# Both keys should now be throttled for the next 15 minutes
assert auth._should_persist_usage("hash-a", now + timedelta(minutes=1)) is False
assert auth._should_persist_usage("hash-b", now + timedelta(minutes=1)) is False
def test_concurrent_access_is_thread_safe():
"""Many threads racing on the same key should only produce one persist per window."""
now = datetime.now(timezone.utc)
results: list[bool] = []
results_lock = threading.Lock()
def worker():
decision = auth._should_persist_usage("hash-shared", now)
with results_lock:
results.append(decision)
threads = [threading.Thread(target=worker) for _ in range(20)]
for t in threads:
t.start()
for t in threads:
t.join()
# Exactly one thread should have won the race and persisted.
assert results.count(True) == 1
assert results.count(False) == 19
def test_usage_interval_matches_spec():
"""The throttle window should be 15 minutes (documented contract)."""
assert auth._USAGE_WRITE_INTERVAL == timedelta(minutes=15)

174
tests/test_downloads_pagination.py Normal file
View file

@ -0,0 +1,174 @@
"""Unit tests for downloads endpoint pagination.
Fix 4.1: `GET /api/v1/downloads` previously returned every task in the
in-memory `download_tasks` dict on every call. With many downloads this
produces a huge payload. The endpoint now supports `limit`, `offset`,
and `status` query params and includes a `total` count.
"""
import sys
import threading
import types
from unittest.mock import patch
import pytest
# api/__init__.py eagerly imports flask_limiter. Stub before import.
def _install_flask_limiter_stub():
if "flask_limiter" in sys.modules:
return
stub = types.ModuleType("flask_limiter")
class _Limiter:
def __init__(self, *args, **kwargs):
pass
def limit(self, *args, **kwargs):
def decorator(target):
return target
return decorator
def init_app(self, app):
pass
stub.Limiter = _Limiter
sys.modules["flask_limiter"] = stub
util_stub = types.ModuleType("flask_limiter.util")
util_stub.get_remote_address = lambda: "127.0.0.1"
sys.modules["flask_limiter.util"] = util_stub
_install_flask_limiter_stub()
from flask import Flask, Blueprint # noqa: E402
from api import downloads as downloads_mod # noqa: E402
def _make_task(status="downloading", when=None):
return {
"status": status,
"track_name": f"Track {when}",
"artist_name": "Artist",
"album_name": "Album",
"username": "user",
"filename": "file.mp3",
"progress": 0,
"size": 1000,
"status_change_time": when or "2026-01-01T00:00:00",
}
def _make_app_with_tasks(tasks_dict):
"""Create a minimal Flask app with the downloads blueprint mounted and
a fake `web_server` module exposing the given download_tasks dict."""
fake_ws = types.ModuleType("web_server")
fake_ws.download_tasks = tasks_dict
fake_ws.tasks_lock = threading.RLock()
sys.modules["web_server"] = fake_ws
# Bypass API key auth for tests.
def _passthrough(f):
return f
app = Flask(__name__)
bp = Blueprint("v1", __name__, url_prefix="/api/v1")
with patch.object(downloads_mod, "require_api_key", _passthrough):
# downloads.register_routes was already imported with the real
# decorator bound, but register_routes runs fresh decorators at
# call time against the passed blueprint.
downloads_mod.register_routes(bp)
app.register_blueprint(bp)
return app
@pytest.fixture
def client():
tasks = {
f"task-{i:03d}": _make_task(
status="downloading" if i % 2 == 0 else "queued",
when=f"2026-01-{i+1:02d}T00:00:00",
)
for i in range(25)
}
app = _make_app_with_tasks(tasks)
with app.test_client() as c:
yield c
def test_default_limit_applied(client):
# 25 tasks, default limit is 100 -> all fit in one page.
resp = client.get("/api/v1/downloads")
assert resp.status_code == 200
data = resp.get_json()["data"]
assert data["total"] == 25
assert data["limit"] == 100
assert data["offset"] == 0
assert len(data["downloads"]) == 25
def test_limit_and_offset_slice_correctly(client):
resp = client.get("/api/v1/downloads?limit=5&offset=0")
data = resp.get_json()["data"]
assert len(data["downloads"]) == 5
assert data["total"] == 25
resp2 = client.get("/api/v1/downloads?limit=5&offset=20")
data2 = resp2.get_json()["data"]
assert len(data2["downloads"]) == 5
# Pages should not overlap.
page1_ids = {t["id"] for t in data["downloads"]}
page5_ids = {t["id"] for t in data2["downloads"]}
assert page1_ids.isdisjoint(page5_ids)
def test_status_filter_single(client):
resp = client.get("/api/v1/downloads?status=downloading&limit=100")
data = resp.get_json()["data"]
# 13 even-indexed tasks (0,2,...,24)
assert data["total"] == 13
for t in data["downloads"]:
assert t["status"] == "downloading"
def test_status_filter_multiple_comma_separated(client):
resp = client.get("/api/v1/downloads?status=downloading,queued&limit=100")
data = resp.get_json()["data"]
assert data["total"] == 25
def test_status_filter_no_match_returns_empty(client):
resp = client.get("/api/v1/downloads?status=nonexistent_status")
data = resp.get_json()["data"]
assert data["total"] == 0
assert data["downloads"] == []
def test_limit_is_clamped_to_max(client):
resp = client.get("/api/v1/downloads?limit=99999")
data = resp.get_json()["data"]
assert data["limit"] == 500
def test_negative_offset_is_normalized(client):
resp = client.get("/api/v1/downloads?offset=-5")
data = resp.get_json()["data"]
assert data["offset"] == 0
def test_invalid_limit_falls_back_to_default(client):
resp = client.get("/api/v1/downloads?limit=not_a_number")
data = resp.get_json()["data"]
assert data["limit"] == 100
def test_tasks_sorted_newest_first(client):
resp = client.get("/api/v1/downloads?limit=3&offset=0")
data = resp.get_json()["data"]
times = [t["status_change_time"] for t in data["downloads"]]
# Most recent (2026-01-25) should come first.
assert times == sorted(times, reverse=True)

266
tests/test_listening_stats_batch_queries.py Normal file
View file

@ -0,0 +1,266 @@
"""Tests for batched queries in listening stats worker (fix 2.1).
Before this fix the worker ran one SELECT per item for:
- resolving db_track_id when inserting history events
- mapping server play-count IDs to existing DB track IDs
- enriching top_artists / top_albums / top_tracks in the stats cache
Each pattern was N+1 on the DB. The fix replaces them with single
batched IN queries (chunked for safety).
"""
from __future__ import annotations
import pytest
from database.music_database import MusicDatabase
from core.listening_stats_worker import ListeningStatsWorker
class _FakeConfigManager:
def get(self, key, default=None):
return default
@pytest.fixture
def db(tmp_path):
return MusicDatabase(str(tmp_path / "music.db"))
@pytest.fixture
def worker(db):
return ListeningStatsWorker(db, _FakeConfigManager())
def _install_query_counter(db):
"""Replace db._get_connection with a wrapper that counts execute() calls.
Returns the counter dict (has key 'n') and a restore callback.
"""
original_get_connection = db._get_connection
counter = {"n": 0}
class _CursorProxy:
def __init__(self, real_cursor):
self._real = real_cursor
def execute(self, sql, params=()):
counter["n"] += 1
return self._real.execute(sql, params)
def __getattr__(self, name):
return getattr(self._real, name)
def __iter__(self):
return iter(self._real)
class _ConnProxy:
def __init__(self, real_conn):
self._real = real_conn
def cursor(self, *a, **k):
return _CursorProxy(self._real.cursor(*a, **k))
def __enter__(self):
self._real.__enter__()
return self
def __exit__(self, *exc):
return self._real.__exit__(*exc)
def __getattr__(self, name):
return getattr(self._real, name)
def wrapped():
return _ConnProxy(original_get_connection())
db._get_connection = wrapped
def restore():
db._get_connection = original_get_connection
return counter, restore
def _insert_track(db, track_id, title, artist_id, artist_name, album_id, album_title):
with db._get_connection() as conn:
conn.execute(
"INSERT OR IGNORE INTO artists (id, name) VALUES (?, ?)",
(artist_id, artist_name),
)
conn.execute(
"INSERT OR IGNORE INTO albums (id, title, artist_id, thumb_url) VALUES (?, ?, ?, ?)",
(album_id, album_title, artist_id, f"http://img/{album_id}.jpg"),
)
conn.execute(
"""INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path)
VALUES (?, ?, ?, ?, 1, 180, ?)""",
(track_id, album_id, artist_id, title, f"/music/{title}.mp3"),
)
conn.commit()
# ---------------------------------------------------------------------------
# _resolve_db_track_ids_batch
# ---------------------------------------------------------------------------
class TestResolveDbTrackIdsBatch:
def test_batch_returns_same_ids_as_per_event_lookup(self, db, worker):
_insert_track(db, "t1", "Alpha", "a1", "Band One", "al1", "X")
_insert_track(db, "t2", "Bravo", "a1", "Band One", "al1", "X")
_insert_track(db, "t3", "Alpha", "a2", "Band Two", "al2", "Y")
events = [
{"title": "Alpha", "artist": "Band One"},
{"title": "Bravo", "artist": "Band One"},
{"title": "Alpha", "artist": "Band Two"},
{"title": "Nonexistent", "artist": "Nobody"},
]
id_map = worker._resolve_db_track_ids_batch(events)
assert id_map[("alpha", "band one")] == "t1"
assert id_map[("bravo", "band one")] == "t2"
assert id_map[("alpha", "band two")] == "t3"
assert ("nonexistent", "nobody") not in id_map
def test_is_case_insensitive(self, db, worker):
_insert_track(db, "t1", "Great Song", "a1", "Some Band", "al1", "X")
id_map = worker._resolve_db_track_ids_batch(
[{"title": "GREAT SONG", "artist": "SOME BAND"}]
)
assert id_map[("great song", "some band")] == "t1"
def test_empty_list_returns_empty_dict(self, worker):
assert worker._resolve_db_track_ids_batch([]) == {}
def test_events_without_title_are_skipped(self, db, worker):
_insert_track(db, "t1", "Song", "a1", "Band", "al1", "X")
id_map = worker._resolve_db_track_ids_batch(
[{"title": "", "artist": "Band"}, {"title": "Song", "artist": "Band"}]
)
assert id_map == {("song", "band"): "t1"}
def test_runs_single_query_regardless_of_event_count(self, db, worker):
"""The whole point: 50 events must not trigger 50 queries."""
for i in range(50):
_insert_track(db, f"t{i}", f"Song {i}", "a1", "Band", "al1", "Album")
counter, restore = _install_query_counter(db)
try:
events = [{"title": f"Song {i}", "artist": "Band"} for i in range(50)]
id_map = worker._resolve_db_track_ids_batch(events)
finally:
restore()
assert len(id_map) == 50
# One batched query (everything fits in one chunk).
assert counter["n"] == 1
# ---------------------------------------------------------------------------
# _map_play_counts_to_db
# ---------------------------------------------------------------------------
class TestMapPlayCountsToDb:
def test_returns_updates_only_for_existing_ids(self, db, worker):
_insert_track(db, "t1", "A", "a1", "Band", "al1", "Album")
_insert_track(db, "t2", "B", "a1", "Band", "al1", "Album")
server_counts = {"t1": 5, "t2": 3, "ghost": 99}
updates = worker._map_play_counts_to_db(server_counts, "plex")
ids = {u["db_track_id"]: u["play_count"] for u in updates}
assert ids == {"t1": 5, "t2": 3}
def test_empty_input_returns_empty_list(self, worker):
assert worker._map_play_counts_to_db({}, "plex") == []
def test_runs_single_query_regardless_of_count_size(self, db, worker):
for i in range(30):
_insert_track(db, f"t{i}", f"S{i}", "a1", "Band", "al1", "Album")
counter, restore = _install_query_counter(db)
try:
server_counts = {f"t{i}": i for i in range(30)}
updates = worker._map_play_counts_to_db(server_counts, "plex")
finally:
restore()
assert len(updates) == 30
assert counter["n"] == 1
# ---------------------------------------------------------------------------
# _enrich_stats_items
# ---------------------------------------------------------------------------
class TestEnrichStatsItems:
def test_enriches_artists_albums_and_tracks(self, db, worker):
_insert_track(db, "t1", "Alpha", "a1", "Band One", "al1", "First Album")
_insert_track(db, "t2", "Bravo", "a2", "Band Two", "al2", "Second Album")
cache = {
"top_artists": [{"name": "Band One"}, {"name": "Band Two"}],
"top_albums": [{"name": "First Album"}, {"name": "Second Album"}],
"top_tracks": [
{"name": "Alpha", "artist": "Band One"},
{"name": "Bravo", "artist": "Band Two"},
],
}
worker._enrich_stats_items(cache)
by_name = {a["name"]: a for a in cache["top_artists"]}
assert by_name["Band One"]["id"] == "a1"
assert by_name["Band Two"]["id"] == "a2"
album_by_name = {a["name"]: a for a in cache["top_albums"]}
assert album_by_name["First Album"]["id"] == "al1"
assert album_by_name["First Album"]["image_url"] == "http://img/al1.jpg"
track_by_name = {t["name"]: t for t in cache["top_tracks"]}
assert track_by_name["Alpha"]["id"] == "t1"
assert track_by_name["Bravo"]["id"] == "t2"
assert track_by_name["Alpha"]["artist_id"] == "a1"
def test_unknown_entries_left_untouched(self, db, worker):
_insert_track(db, "t1", "Real", "a1", "Real Band", "al1", "Real Album")
cache = {
"top_artists": [{"name": "Unknown Band"}],
"top_albums": [{"name": "Unknown Album"}],
"top_tracks": [{"name": "Unknown", "artist": "Nobody"}],
}
worker._enrich_stats_items(cache)
assert "id" not in cache["top_artists"][0]
assert "id" not in cache["top_albums"][0]
assert "id" not in cache["top_tracks"][0]
def test_empty_cache_is_safe(self, worker):
worker._enrich_stats_items({}) # must not raise
worker._enrich_stats_items({"top_artists": [], "top_albums": [], "top_tracks": []})
def test_runs_one_query_per_section(self, db, worker):
for i in range(20):
_insert_track(db, f"t{i}", f"Song {i}", f"a{i}", f"Band {i}",
f"al{i}", f"Album {i}")
cache = {
"top_artists": [{"name": f"Band {i}"} for i in range(20)],
"top_albums": [{"name": f"Album {i}"} for i in range(20)],
"top_tracks": [
{"name": f"Song {i}", "artist": f"Band {i}"} for i in range(20)
],
}
counter, restore = _install_query_counter(db)
try:
worker._enrich_stats_items(cache)
finally:
restore()
# 3 batched queries total (artists + albums + tracks), not 60.
assert counter["n"] == 3

217
tests/test_metadata_cache_batch_lookup.py Normal file
View file

@ -0,0 +1,217 @@
"""Unit tests for batched metadata cache entity lookups.
Fix 1.3: `MetadataCache.get_search_results` previously resolved cached
entity IDs one-by-one, producing N extra SELECT queries per cached
search. The resolution now runs as a single batched `IN` query (chunked
to stay below SQLite's variable limit) and preserves the original
`result_ids` ordering.
"""
import json
import sqlite3
from datetime import datetime, timedelta
from unittest.mock import MagicMock
import pytest
from core.metadata_cache import MetadataCache
@pytest.fixture
def cache_with_db(tmp_path):
"""MetadataCache wired to a temporary SQLite DB with the required tables."""
db_path = tmp_path / "cache.db"
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
conn.executescript(
"""
CREATE TABLE metadata_cache_searches (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
search_type TEXT NOT NULL,
query_normalized TEXT NOT NULL,
search_limit INTEGER NOT NULL,
result_ids TEXT NOT NULL,
created_at TEXT NOT NULL,
last_accessed_at TEXT,
access_count INTEGER DEFAULT 0
);
CREATE TABLE metadata_cache_entities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
entity_type TEXT NOT NULL,
entity_id TEXT NOT NULL,
raw_json TEXT NOT NULL
);
"""
)
# Fake MusicDatabase that yields this sqlite connection.
fake_db = MagicMock()
def _get_connection():
c = sqlite3.connect(str(db_path))
c.row_factory = sqlite3.Row
return c
fake_db._get_connection.side_effect = _get_connection
cache = MetadataCache()
cache._get_db = lambda: fake_db # type: ignore[method-assign]
return cache, conn
def _insert_search(conn, source, search_type, query, limit, result_ids, created_at=None):
created_at = created_at or datetime.now().isoformat()
conn.execute(
"""INSERT INTO metadata_cache_searches
(source, search_type, query_normalized, search_limit, result_ids, created_at)
VALUES (?, ?, ?, ?, ?, ?)""",
(source, search_type, query.lower(), limit, json.dumps(result_ids), created_at),
)
conn.commit()
def _insert_entity(conn, source, entity_type, entity_id, payload):
conn.execute(
"""INSERT INTO metadata_cache_entities
(source, entity_type, entity_id, raw_json) VALUES (?, ?, ?, ?)""",
(source, entity_type, entity_id, json.dumps(payload)),
)
conn.commit()
def test_returns_results_in_original_order(cache_with_db):
cache, conn = cache_with_db
ids = ["z", "a", "m", "b"]
_insert_search(conn, "spotify", "track", "hello", 50, ids)
for eid in ids:
_insert_entity(conn, "spotify", "track", eid, {"id": eid, "name": f"track {eid}"})
results = cache.get_search_results("spotify", "track", "hello", 50)
assert results is not None
assert [r["id"] for r in results] == ids
def test_missing_entities_below_threshold_returns_none(cache_with_db):
cache, conn = cache_with_db
ids = [f"id{i}" for i in range(10)]
_insert_search(conn, "spotify", "track", "partial", 50, ids)
# Only insert 5/10 entities — below the 80 percent threshold.
for eid in ids[:5]:
_insert_entity(conn, "spotify", "track", eid, {"id": eid})
assert cache.get_search_results("spotify", "track", "partial", 50) is None
def test_missing_entities_at_threshold_returns_partial(cache_with_db):
cache, conn = cache_with_db
ids = [f"id{i}" for i in range(10)]
_insert_search(conn, "spotify", "track", "threshold", 50, ids)
# Insert 8/10 = exactly 80 percent — should return the 8 found.
for eid in ids[:8]:
_insert_entity(conn, "spotify", "track", eid, {"id": eid})
results = cache.get_search_results("spotify", "track", "threshold", 50)
assert results is not None
assert len(results) == 8
assert [r["id"] for r in results] == ids[:8]
def test_empty_result_ids_returns_empty_list(cache_with_db):
cache, conn = cache_with_db
_insert_search(conn, "spotify", "track", "empty", 50, [])
assert cache.get_search_results("spotify", "track", "empty", 50) == []
def test_expired_search_returns_none(cache_with_db):
cache, conn = cache_with_db
old = (datetime.now() - timedelta(days=10)).isoformat()
_insert_search(conn, "spotify", "track", "stale", 50, ["a"], created_at=old)
_insert_entity(conn, "spotify", "track", "a", {"id": "a"})
assert cache.get_search_results("spotify", "track", "stale", 50) is None
def test_cache_miss_on_unknown_query(cache_with_db):
cache, _ = cache_with_db
assert cache.get_search_results("spotify", "track", "nothing cached", 50) is None
def test_batch_lookup_uses_single_round_trip(tmp_path):
"""Sanity check that resolution does not issue one SELECT per entity_id."""
db_path = tmp_path / "cache2.db"
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
conn.executescript(
"""
CREATE TABLE metadata_cache_searches (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT, search_type TEXT, query_normalized TEXT,
search_limit INTEGER, result_ids TEXT, created_at TEXT,
last_accessed_at TEXT, access_count INTEGER DEFAULT 0
);
CREATE TABLE metadata_cache_entities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT, entity_type TEXT, entity_id TEXT, raw_json TEXT
);
"""
)
ids = [f"e{i}" for i in range(50)]
_insert_search(conn, "spotify", "track", "bulk", 50, ids)
for eid in ids:
_insert_entity(conn, "spotify", "track", eid, {"id": eid})
raw_selects = {"n": 0}
class CountingConnection:
def __init__(self, inner):
self._inner = inner
def cursor(self):
inner_cursor = self._inner.cursor()
class CountingCursor:
def __init__(self, c):
self._c = c
def execute(self, sql, params=()):
if "raw_json FROM metadata_cache_entities" in sql:
raw_selects["n"] += 1
return self._c.execute(sql, params)
def fetchone(self):
return self._c.fetchone()
def fetchall(self):
return self._c.fetchall()
return CountingCursor(inner_cursor)
def commit(self):
return self._inner.commit()
def close(self):
return self._inner.close()
def __getattr__(self, name):
return getattr(self._inner, name)
def _connect():
c = sqlite3.connect(str(db_path))
c.row_factory = sqlite3.Row
return CountingConnection(c)
fake_db = MagicMock()
fake_db._get_connection.side_effect = _connect
cache = MetadataCache()
cache._get_db = lambda: fake_db # type: ignore[method-assign]
results = cache.get_search_results("spotify", "track", "bulk", 50)
assert results is not None and len(results) == 50
# With batching, 50 entities should resolve in a single SELECT, not 50.
assert raw_selects["n"] == 1

137
tests/test_request_cleanup_timer.py Normal file
View file

@ -0,0 +1,137 @@
"""Tests for api/request.py periodic cleanup timer (fix 4.2).
Before this fix, `_cleanup_old_requests()` was only invoked on
`create_request`. During idle periods stale entries lingered for the
full uptime of the server. A background timer now runs every
_CLEANUP_INTERVAL_SECONDS and evicts anything older than _MAX_REQUEST_AGE.
"""
import sys
import threading
import time
import types
from datetime import datetime, timedelta
import pytest
# api/__init__.py imports flask_limiter at module load. Stub it.
def _install_flask_limiter_stub():
if "flask_limiter" in sys.modules:
return
stub = types.ModuleType("flask_limiter")
class _Limiter:
def __init__(self, *args, **kwargs):
pass
def limit(self, *args, **kwargs):
def decorator(target):
return target
return decorator
def init_app(self, app):
pass
stub.Limiter = _Limiter
sys.modules["flask_limiter"] = stub
util_stub = types.ModuleType("flask_limiter.util")
util_stub.get_remote_address = lambda: "127.0.0.1"
sys.modules["flask_limiter.util"] = util_stub
_install_flask_limiter_stub()
from api import request as request_mod # noqa: E402
@pytest.fixture(autouse=True)
def _clean_state():
# Ensure no leftover thread from a previous test.
request_mod.stop_cleanup_thread(timeout=1.0)
with request_mod._requests_lock:
request_mod._pending_requests.clear()
yield
request_mod.stop_cleanup_thread(timeout=1.0)
with request_mod._requests_lock:
request_mod._pending_requests.clear()
def _add_request(request_id, age_minutes):
with request_mod._requests_lock:
request_mod._pending_requests[request_id] = {
"request_id": request_id,
"query": "q",
"status": "queued",
"created_at": datetime.now() - timedelta(minutes=age_minutes),
"completed_at": None,
"download_id": None,
"error": None,
}
class TestCleanupOldRequests:
def test_evicts_entries_older_than_ttl(self):
_add_request("old-1", age_minutes=120) # > 60 min TTL
_add_request("old-2", age_minutes=61)
_add_request("fresh", age_minutes=5)
removed = request_mod._cleanup_old_requests()
assert removed == 2
with request_mod._requests_lock:
ids = set(request_mod._pending_requests.keys())
assert ids == {"fresh"}
def test_returns_zero_when_nothing_to_evict(self):
_add_request("fresh", age_minutes=5)
assert request_mod._cleanup_old_requests() == 0
def test_empty_map_is_safe(self):
assert request_mod._cleanup_old_requests() == 0
class TestCleanupThreadLifecycle:
def test_start_returns_true_first_time_false_after(self):
assert request_mod.start_cleanup_thread() is True
# Second call in the same process should not start a new thread.
assert request_mod.start_cleanup_thread() is False
def test_stop_joins_thread(self):
request_mod.start_cleanup_thread()
assert request_mod._cleanup_thread is not None
assert request_mod._cleanup_thread.is_alive()
request_mod.stop_cleanup_thread(timeout=2.0)
assert request_mod._cleanup_thread is None
def test_thread_evicts_on_wakeup(self, monkeypatch):
# Force a tiny interval so the test doesn't wait 5 minutes.
monkeypatch.setattr(request_mod, "_CLEANUP_INTERVAL_SECONDS", 0.05)
_add_request("old", age_minutes=120)
request_mod.start_cleanup_thread()
# Give the loop time to wake up and run at least once.
deadline = time.time() + 2.0
while time.time() < deadline:
with request_mod._requests_lock:
remaining = set(request_mod._pending_requests.keys())
if "old" not in remaining:
break
time.sleep(0.05)
with request_mod._requests_lock:
assert "old" not in request_mod._pending_requests
def test_stop_signals_thread_to_exit_promptly(self, monkeypatch):
# With a huge interval, stop must still make the thread exit via the
# stop event, not wait for the next timeout.
monkeypatch.setattr(request_mod, "_CLEANUP_INTERVAL_SECONDS", 30.0)
request_mod.start_cleanup_thread()
thread = request_mod._cleanup_thread
assert thread is not None
request_mod.stop_cleanup_thread(timeout=2.0)
assert not thread.is_alive()

99
tests/test_track_search_single_query.py Normal file
View file

@ -0,0 +1,99 @@
"""Unit tests for api_search_tracks (fix 2.2).
The track search endpoint previously called search_tracks() then
api_get_tracks_by_ids() to re-hydrate the same rows. api_search_tracks
now returns the full dict rows in a single query.
"""
import pytest
from database.music_database import MusicDatabase
@pytest.fixture
def db(tmp_path):
return MusicDatabase(str(tmp_path / "music.db"))
def _insert_track(db, track_id, title, artist_id, artist_name, album_id, album_title):
with db._get_connection() as conn:
conn.execute(
"INSERT OR IGNORE INTO artists (id, name) VALUES (?, ?)",
(artist_id, artist_name),
)
conn.execute(
"INSERT OR IGNORE INTO albums (id, title, artist_id) VALUES (?, ?, ?)",
(album_id, album_title, artist_id),
)
conn.execute(
"""INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path)
VALUES (?, ?, ?, ?, 1, 180, ?)""",
(track_id, album_id, artist_id, title, f"/music/{title}.mp3"),
)
conn.commit()
def test_api_search_tracks_returns_dict_rows_with_full_columns(db):
_insert_track(db, 100, "Great Song", 1, "Band One", 10, "Album X")
results = db.api_search_tracks(title="great")
assert len(results) == 1
row = results[0]
assert isinstance(row, dict)
assert str(row["id"]) == "100"
assert row["title"] == "Great Song"
assert row["artist_name"] == "Band One"
assert row["album_title"] == "Album X"
# file_path is a tracks.* column that must be present
assert row["file_path"] == "/music/Great Song.mp3"
def test_api_search_tracks_empty_query_returns_empty(db):
_insert_track(db, 1, "Track", 1, "Artist", 1, "Album")
assert db.api_search_tracks() == []
def test_api_search_tracks_no_matches_returns_empty(db):
_insert_track(db, 1, "Alpha", 1, "Artist A", 1, "Album A")
assert db.api_search_tracks(title="nonexistent") == []
def test_api_search_tracks_matches_by_artist(db):
_insert_track(db, 1, "Song A", 1, "Mystery Band", 1, "Album A")
_insert_track(db, 2, "Song B", 2, "Other Artist", 2, "Album B")
results = db.api_search_tracks(artist="mystery")
assert len(results) == 1
assert str(results[0]["id"]) == "1"
def test_api_search_tracks_respects_limit(db):
for i in range(10):
_insert_track(db, i + 1, f"Common Title {i}", 1, "Artist", 1, "Album")
results = db.api_search_tracks(title="common", limit=3)
assert len(results) == 3
def test_api_search_tracks_falls_back_to_fuzzy(db):
"""When basic LIKE returns nothing, fuzzy fallback should still find matches."""
_insert_track(db, 1, "Bohemian Rhapsody", 1, "Queen", 1, "A Night at the Opera")
# Basic search "bohemian" matches directly; the fuzzy path exists for
# cases where basic fails. Verify at least that the method surfaces a
# result for a direct query.
results = db.api_search_tracks(title="bohemian rhapsody")
assert any(str(r["id"]) == "1" for r in results)
def test_search_tracks_still_returns_database_track_objects(db):
"""Existing callers rely on the DatabaseTrack return shape of search_tracks()."""
_insert_track(db, 1, "Legacy Track", 1, "Legacy Artist", 1, "Legacy Album")
results = db.search_tracks(title="legacy")
assert len(results) == 1
# DatabaseTrack exposes attributes, not dict keys.
assert str(results[0].id) == "1"
assert results[0].title == "Legacy Track"
assert results[0].artist_name == "Legacy Artist"

131
tests/test_wishlist_pagination.py Normal file
View file

@ -0,0 +1,131 @@
"""Unit tests for SQL-level wishlist pagination and category filtering.
Fix 1.4: the API endpoint previously loaded the entire wishlist, filtered
by category in Python, then sliced for the requested page. Pagination and
category filtering are now pushed to SQL via LIMIT/OFFSET and json_extract.
"""
import json
import pytest
from database.music_database import MusicDatabase
@pytest.fixture
def db(tmp_path):
return MusicDatabase(str(tmp_path / "music.db"))
def _insert(db, spotify_track_id, name, album_type, profile_id=1, date_added=None):
spotify_data = {
"id": spotify_track_id,
"name": name,
"artists": [{"name": "Test Artist"}],
"album": {"name": f"Album for {name}", "album_type": album_type, "images": []},
}
date_added = date_added or f"2026-01-{len(spotify_track_id):02d}T00:00:00"
with db._get_connection() as conn:
conn.execute(
"""INSERT INTO wishlist_tracks
(spotify_track_id, spotify_data, failure_reason, retry_count,
date_added, source_type, source_info, profile_id)
VALUES (?, ?, '', 0, ?, 'test', '{}', ?)""",
(spotify_track_id, json.dumps(spotify_data), date_added, profile_id),
)
conn.commit()
def test_pagination_limit_offset(db):
for i in range(5):
_insert(db, f"t{i}", f"Track {i}", "album", date_added=f"2026-01-0{i+1}T00:00:00")
page1 = db.get_wishlist_tracks(limit=2, offset=0)
page2 = db.get_wishlist_tracks(limit=2, offset=2)
page3 = db.get_wishlist_tracks(limit=2, offset=4)
assert [t["spotify_track_id"] for t in page1] == ["t0", "t1"]
assert [t["spotify_track_id"] for t in page2] == ["t2", "t3"]
assert [t["spotify_track_id"] for t in page3] == ["t4"]
def test_category_filter_albums(db):
_insert(db, "a1", "Album Track 1", "album", date_added="2026-01-01T00:00:00")
_insert(db, "s1", "Single 1", "single", date_added="2026-01-02T00:00:00")
_insert(db, "a2", "Album Track 2", "album", date_added="2026-01-03T00:00:00")
_insert(db, "e1", "EP Track", "ep", date_added="2026-01-04T00:00:00")
albums = db.get_wishlist_tracks(category="albums")
assert sorted(t["spotify_track_id"] for t in albums) == ["a1", "a2"]
def test_category_filter_singles_includes_non_album_types(db):
_insert(db, "a1", "Album", "album", date_added="2026-01-01T00:00:00")
_insert(db, "s1", "Single", "single", date_added="2026-01-02T00:00:00")
_insert(db, "e1", "EP", "ep", date_added="2026-01-03T00:00:00")
_insert(db, "c1", "Compilation", "compilation", date_added="2026-01-04T00:00:00")
singles = db.get_wishlist_tracks(category="singles")
assert sorted(t["spotify_track_id"] for t in singles) == ["c1", "e1", "s1"]
def test_category_filter_singles_includes_missing_album_type(db):
# Manually insert a row whose album has no album_type (malformed)
with db._get_connection() as conn:
conn.execute(
"""INSERT INTO wishlist_tracks
(spotify_track_id, spotify_data, failure_reason, retry_count,
date_added, source_type, source_info, profile_id)
VALUES ('x1', ?, '', 0, '2026-01-01', 'test', '{}', 1)""",
(json.dumps({"id": "x1", "name": "X", "album": {"name": "A"}}),),
)
conn.commit()
singles = db.get_wishlist_tracks(category="singles")
assert [t["spotify_track_id"] for t in singles] == ["x1"]
def test_get_wishlist_count_no_filter(db):
for i in range(7):
_insert(db, f"t{i}", f"Track {i}", "album", date_added=f"2026-01-0{i+1}T00:00:00")
assert db.get_wishlist_count() == 7
def test_get_wishlist_count_with_category_filter(db):
_insert(db, "a1", "A1", "album", date_added="2026-01-01T00:00:00")
_insert(db, "a2", "A2", "album", date_added="2026-01-02T00:00:00")
_insert(db, "s1", "S1", "single", date_added="2026-01-03T00:00:00")
assert db.get_wishlist_count(category="albums") == 2
assert db.get_wishlist_count(category="singles") == 1
assert db.get_wishlist_count() == 3
def test_profile_isolation(db):
_insert(db, "p1-a", "A", "album", profile_id=1, date_added="2026-01-01T00:00:00")
_insert(db, "p1-b", "B", "album", profile_id=1, date_added="2026-01-02T00:00:00")
_insert(db, "p2-a", "C", "album", profile_id=2, date_added="2026-01-03T00:00:00")
assert db.get_wishlist_count(profile_id=1) == 2
assert db.get_wishlist_count(profile_id=2) == 1
p1 = db.get_wishlist_tracks(profile_id=1)
assert sorted(t["spotify_track_id"] for t in p1) == ["p1-a", "p1-b"]
def test_backward_compat_no_args_returns_all(db):
"""Existing callers (wishlist_service) pass no limit/offset — must still work."""
for i in range(3):
_insert(db, f"t{i}", f"Track {i}", "album", date_added=f"2026-01-0{i+1}T00:00:00")
rows = db.get_wishlist_tracks()
assert len(rows) == 3
def test_ordering_by_date_added(db):
_insert(db, "newer", "Newer", "album", date_added="2026-02-01T00:00:00")
_insert(db, "older", "Older", "album", date_added="2026-01-01T00:00:00")
rows = db.get_wishlist_tracks()
assert [r["spotify_track_id"] for r in rows] == ["older", "newer"]

333
tests/test_worker_existing_id_marks_matched.py Normal file
View file

@ -0,0 +1,333 @@
"""Tests for Fix 1.1: worker re-processing loop.
Before this fix:
* `musicbrainz_worker._get_existing_id` always queried `musicbrainz_id` even
for `albums`/`tracks` (which use `musicbrainz_release_id` /
`musicbrainz_recording_id`), so the existence check silently failed and
every row was re-processed on every loop.
* `lastfm_worker._get_existing_id` queried a non-existent `lastfm_id`
column (the real column is `lastfm_url`), with the same effect.
* Even when workers did find an existing external ID, they returned
without setting `<provider>_match_status`, so the row stayed NULL and
the next worker loop re-selected it forever.
This test module covers:
1. The backfill migration that retroactively sets match_status='matched'
for rows that already have an external ID populated.
2. `_get_existing_id` returns the correct column per entity type for
MusicBrainz and Last.fm.
3. Each worker's `_process_*` short-circuit path sets match_status to
'matched' when an existing external ID is found (lastfm, tidal,
qobuz, musicbrainz).
"""
from __future__ import annotations
import sys
import types
from unittest.mock import MagicMock, patch
import pytest
from database.music_database import MusicDatabase
# ---------------------------------------------------------------------------
# Minimal stubs for optional deps some workers import at module load.
# ---------------------------------------------------------------------------
def _ensure_stub_module(name: str, attrs: dict | None = None) -> None:
if name in sys.modules:
return
mod = types.ModuleType(name)
for k, v in (attrs or {}).items():
setattr(mod, k, v)
sys.modules[name] = mod
# TidalClient / QobuzClient live in core.* and are safe to import but require
# config_manager. We patch the classes at instantiation time instead.
@pytest.fixture
def db(tmp_path):
return MusicDatabase(str(tmp_path / "music.db"))
# ---------------------------------------------------------------------------
# Backfill migration
# ---------------------------------------------------------------------------
class TestBackfillMigration:
def test_lastfm_url_set_but_status_null_gets_matched(self, db):
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, lastfm_url) VALUES (?, ?, ?)",
("a1", "Artist A", "https://last.fm/music/Artist%20A"),
)
cur.execute(
"INSERT INTO artists (id, name, lastfm_url, lastfm_match_status) VALUES (?, ?, ?, ?)",
("a2", "Artist B", "https://last.fm/music/B", "matched"),
)
cur.execute(
"INSERT INTO artists (id, name) VALUES (?, ?)",
("a3", "Artist C"), # no url, status stays NULL
)
conn.commit()
# Run backfill a second time (first already ran during __init__)
db._backfill_match_status_for_existing_ids(cur)
conn.commit()
rows = cur.execute(
"SELECT name, lastfm_match_status FROM artists ORDER BY name"
).fetchall()
by_name = {r[0]: r[1] for r in rows}
assert by_name["Artist A"] == "matched"
assert by_name["Artist B"] == "matched" # untouched
assert by_name["Artist C"] is None # no id => no backfill
def test_musicbrainz_release_id_on_albums(self, db):
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name) VALUES (?, ?)",
("art1", "A"),
)
cur.execute(
"INSERT INTO albums (id, artist_id, title, musicbrainz_release_id) "
"VALUES (?, ?, ?, ?)",
("alb1", "art1", "Album X", "mb-release-uuid"),
)
conn.commit()
db._backfill_match_status_for_existing_ids(cur)
conn.commit()
status = cur.execute(
"SELECT musicbrainz_match_status FROM albums WHERE title = 'Album X'"
).fetchone()[0]
assert status == "matched"
def test_musicbrainz_recording_id_on_tracks(self, db):
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute("INSERT INTO artists (id, name) VALUES ('art2', 'A')")
cur.execute(
"INSERT INTO albums (id, artist_id, title) VALUES (?, ?, 'Alb')",
("alb2", "art2"),
)
cur.execute(
"INSERT INTO tracks (id, artist_id, album_id, title, musicbrainz_recording_id) "
"VALUES (?, ?, ?, ?, ?)",
("trk1", "art2", "alb2", "T1", "mb-rec-uuid"),
)
conn.commit()
db._backfill_match_status_for_existing_ids(cur)
conn.commit()
status = cur.execute(
"SELECT musicbrainz_match_status FROM tracks WHERE title = 'T1'"
).fetchone()[0]
assert status == "matched"
def test_tidal_and_qobuz_ids_backfilled(self, db):
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, tidal_id, qobuz_id) VALUES (?, ?, ?, ?)",
("art3", "A", "tidal123", "qobuz456"),
)
conn.commit()
db._backfill_match_status_for_existing_ids(cur)
conn.commit()
row = cur.execute(
"SELECT tidal_match_status, qobuz_match_status FROM artists WHERE id = 'art3'"
).fetchone()
assert tuple(row) == ("matched", "matched")
def test_empty_string_id_is_not_backfilled(self, db):
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, tidal_id) VALUES (?, ?, ?)",
("art4", "Empty", ""),
)
conn.commit()
db._backfill_match_status_for_existing_ids(cur)
conn.commit()
status = cur.execute(
"SELECT tidal_match_status FROM artists WHERE id = 'art4'"
).fetchone()[0]
assert status is None
# ---------------------------------------------------------------------------
# _get_existing_id column-mapping correctness
# ---------------------------------------------------------------------------
class TestGetExistingIdColumnMapping:
def _insert_tree(self, db):
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, lastfm_url, musicbrainz_id) "
"VALUES (?, ?, ?, ?)",
("art_x", "A", "https://last.fm/a", "mb-artist"),
)
cur.execute(
"INSERT INTO albums (id, artist_id, title, lastfm_url, musicbrainz_release_id) "
"VALUES (?, ?, ?, ?, ?)",
("alb_x", "art_x", "Album", "https://last.fm/album", "mb-release"),
)
cur.execute(
"INSERT INTO tracks (id, artist_id, album_id, title, lastfm_url, musicbrainz_recording_id) "
"VALUES (?, ?, ?, ?, ?, ?)",
("trk_x", "art_x", "alb_x", "Track", "https://last.fm/track", "mb-rec"),
)
conn.commit()
return "art_x", "alb_x", "trk_x"
def test_lastfm_worker_reads_lastfm_url_for_all_entity_types(self, db):
# Import lazily so test collection doesn't fail if config_manager is unavailable.
from core import lastfm_worker as lw
artist_id, album_id, track_id = self._insert_tree(db)
with patch.object(lw.LastFMWorker, "_init_client", return_value=None):
worker = lw.LastFMWorker(db)
assert worker._get_existing_id("artist", artist_id) == "https://last.fm/a"
assert worker._get_existing_id("album", album_id) == "https://last.fm/album"
assert worker._get_existing_id("track", track_id) == "https://last.fm/track"
def test_musicbrainz_worker_reads_correct_column_per_entity(self, db):
from core import musicbrainz_worker as mbw
artist_id, album_id, track_id = self._insert_tree(db)
with patch.object(mbw, "MusicBrainzService", return_value=MagicMock()):
worker = mbw.MusicBrainzWorker(db)
assert worker._get_existing_id("artist", artist_id) == "mb-artist"
assert worker._get_existing_id("album", album_id) == "mb-release"
assert worker._get_existing_id("track", track_id) == "mb-rec"
# ---------------------------------------------------------------------------
# Worker _process_* short-circuit marks status='matched'
# ---------------------------------------------------------------------------
def _read_status(db, table: str, column: str, row_id: int):
with db._get_connection() as conn:
row = conn.execute(
f"SELECT {column} FROM {table} WHERE id = ?", (row_id,)
).fetchone()
return row[0] if row else None
class TestLastFMWorkerMarksMatched:
def test_existing_url_triggers_matched_status(self, db):
from core import lastfm_worker as lw
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, lastfm_url) VALUES (?, ?, ?)",
("art_lf", "A", "https://last.fm/a"),
)
artist_id = "art_lf"
# Explicitly null out status to simulate legacy row
cur.execute(
"UPDATE artists SET lastfm_match_status = NULL WHERE id = ?",
(artist_id,),
)
conn.commit()
with patch.object(lw.LastFMWorker, "_init_client", return_value=None):
worker = lw.LastFMWorker(db)
worker.client = MagicMock()
worker._process_artist(artist_id, "A")
# Client must NOT be called because we short-circuited.
worker.client.get_artist_info.assert_not_called()
assert _read_status(db, "artists", "lastfm_match_status", artist_id) == "matched"
class TestTidalWorkerMarksMatched:
def test_existing_tidal_id_triggers_matched_status(self, db):
from core import tidal_worker as tw
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, tidal_id) VALUES (?, ?, ?)",
("art_td", "A", "tidal-123"),
)
artist_id = "art_td"
cur.execute(
"UPDATE artists SET tidal_match_status = NULL WHERE id = ?",
(artist_id,),
)
conn.commit()
fake_client = MagicMock()
worker = tw.TidalWorker(db, client=fake_client)
worker._process_artist(artist_id, "A")
fake_client.search_artist.assert_not_called()
assert _read_status(db, "artists", "tidal_match_status", artist_id) == "matched"
class TestQobuzWorkerMarksMatched:
def test_existing_qobuz_id_triggers_matched_status(self, db):
from core import qobuz_worker as qw
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, qobuz_id) VALUES (?, ?, ?)",
("art_qz", "A", "qobuz-xyz"),
)
artist_id = "art_qz"
cur.execute(
"UPDATE artists SET qobuz_match_status = NULL WHERE id = ?",
(artist_id,),
)
conn.commit()
fake_client = MagicMock()
worker = qw.QobuzWorker(db, client=fake_client)
worker._process_artist(artist_id, "A")
fake_client.search_artist.assert_not_called()
assert _read_status(db, "artists", "qobuz_match_status", artist_id) == "matched"
class TestMusicBrainzWorkerMarksMatched:
def test_existing_mbid_triggers_matched_status_via_service(self, db):
from core import musicbrainz_worker as mbw
with db._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO artists (id, name, musicbrainz_id) VALUES (?, ?, ?)",
("art_mb", "A", "mb-uuid"),
)
artist_id = "art_mb"
conn.commit()
fake_service = MagicMock()
with patch.object(mbw, "MusicBrainzService", return_value=fake_service):
worker = mbw.MusicBrainzWorker(db)
# mb_service on the instance is the MagicMock
worker._process_item({"type": "artist", "id": artist_id, "name": "A"})
fake_service.update_artist_mbid.assert_called_once_with(
artist_id, "mb-uuid", "matched"
)