Expand wishlist test coverage
- add direct service and presence coverage - pin resolver, processing, route, and payload edge cases - keep wishlist package extraction safe for future refactors
This commit is contained in:
Antti Kettunen
parent
f5226bd5b5
commit
a7c1bb96a1
7 changed files with 753 additions and 2 deletions
|
|
@ -128,6 +128,8 @@ def _build_runtime(
|
|||
progress_calls=None,
|
||||
guard_events=None,
|
||||
batch_map=None,
|
||||
guard_acquired=True,
|
||||
is_actually_processing=False,
|
||||
):
|
||||
if progress_calls is None:
|
||||
progress_calls = []
|
||||
|
|
@ -146,7 +148,7 @@ def _build_runtime(
|
|||
def guard():
|
||||
guard_events.append("enter")
|
||||
try:
|
||||
yield True
|
||||
yield guard_acquired
|
||||
finally:
|
||||
guard_events.append("exit")
|
||||
|
||||
|
|
@ -159,7 +161,6 @@ def _build_runtime(
|
|||
|
||||
runtime = WishlistAutoProcessingRuntime(
|
||||
processing_guard=guard,
|
||||
is_actually_processing=lambda: False,
|
||||
app_context_factory=app_context,
|
||||
get_wishlist_service=lambda: wishlist_service,
|
||||
get_profiles_database=lambda: profiles_db,
|
||||
|
|
@ -174,6 +175,7 @@ def _build_runtime(
|
|||
get_active_server=lambda: active_server,
|
||||
logger=logger,
|
||||
current_time_fn=lambda: 123.0,
|
||||
is_actually_processing=lambda: is_actually_processing,
|
||||
profile_id=1,
|
||||
)
|
||||
return runtime, wishlist_service, profiles_db, music_db, executor, logger, progress_calls, guard_events
|
||||
|
|
@ -231,3 +233,94 @@ def test_process_wishlist_automatically_creates_batch_for_matching_tracks():
|
|||
assert any(kwargs.get("progress") == 50 for _args, kwargs in progress_calls)
|
||||
assert guard_events == ["enter", "exit"]
|
||||
assert any("Starting automatic wishlist batch" in msg for msg in logger.info_messages)
|
||||
|
||||
|
||||
def test_process_wishlist_automatically_returns_early_when_already_processing():
|
||||
runtime, _service, _profiles_db, music_db, executor, logger, progress_calls, guard_events = _build_runtime(
|
||||
tracks=[
|
||||
{
|
||||
"name": "Album Track",
|
||||
"artists": [{"name": "Artist A"}],
|
||||
"spotify_data": {"album": {"album_type": "album"}},
|
||||
}
|
||||
],
|
||||
cycle_value="albums",
|
||||
count=1,
|
||||
is_actually_processing=True,
|
||||
)
|
||||
|
||||
process_wishlist_automatically(runtime, automation_id="auto-3")
|
||||
|
||||
assert executor.submissions == []
|
||||
assert music_db.duplicate_removals == []
|
||||
assert progress_calls == []
|
||||
assert guard_events == []
|
||||
assert any("Already processing" in msg for msg in logger.info_messages)
|
||||
|
||||
|
||||
def test_process_wishlist_automatically_returns_early_when_guard_not_acquired():
|
||||
runtime, _service, _profiles_db, music_db, executor, logger, progress_calls, guard_events = _build_runtime(
|
||||
tracks=[
|
||||
{
|
||||
"name": "Album Track",
|
||||
"artists": [{"name": "Artist A"}],
|
||||
"spotify_data": {"album": {"album_type": "album"}},
|
||||
}
|
||||
],
|
||||
cycle_value="albums",
|
||||
count=1,
|
||||
guard_acquired=False,
|
||||
)
|
||||
|
||||
process_wishlist_automatically(runtime, automation_id="auto-4")
|
||||
|
||||
assert executor.submissions == []
|
||||
assert music_db.duplicate_removals == []
|
||||
assert progress_calls == []
|
||||
assert guard_events == ["enter", "exit"]
|
||||
assert any("race condition check" in msg for msg in logger.info_messages)
|
||||
|
||||
|
||||
def test_process_wishlist_automatically_returns_early_when_no_tracks_are_available():
|
||||
runtime, _service, _profiles_db, music_db, executor, logger, progress_calls, guard_events = _build_runtime(
|
||||
tracks=[],
|
||||
cycle_value="albums",
|
||||
count=0,
|
||||
)
|
||||
|
||||
process_wishlist_automatically(runtime, automation_id="auto-5")
|
||||
|
||||
assert executor.submissions == []
|
||||
assert music_db.duplicate_removals == []
|
||||
assert guard_events == ["enter", "exit"]
|
||||
assert [kwargs.get("progress") for _args, kwargs in progress_calls if "progress" in kwargs] == [10]
|
||||
assert any("No tracks in wishlist for auto-processing" in msg for msg in logger.warning_messages)
|
||||
|
||||
|
||||
def test_process_wishlist_automatically_skips_when_wishlist_batch_is_already_active():
|
||||
batch_map = {
|
||||
"batch-1": {
|
||||
"playlist_id": "wishlist",
|
||||
"phase": "analysis",
|
||||
}
|
||||
}
|
||||
runtime, _service, _profiles_db, music_db, executor, logger, progress_calls, guard_events = _build_runtime(
|
||||
tracks=[
|
||||
{
|
||||
"name": "Album Track",
|
||||
"artists": [{"name": "Artist A"}],
|
||||
"spotify_data": {"album": {"album_type": "album"}},
|
||||
}
|
||||
],
|
||||
cycle_value="albums",
|
||||
count=1,
|
||||
batch_map=batch_map,
|
||||
)
|
||||
|
||||
process_wishlist_automatically(runtime, automation_id="auto-6")
|
||||
|
||||
assert executor.submissions == []
|
||||
assert music_db.duplicate_removals == []
|
||||
assert guard_events == ["enter", "exit"]
|
||||
assert [kwargs.get("progress") for _args, kwargs in progress_calls if "progress" in kwargs] == [10]
|
||||
assert any("already active in another batch" in msg for msg in logger.info_messages)
|
||||
|
|
|
|||
|
|
@ -56,3 +56,37 @@ def test_ensure_spotify_track_format_builds_webui_shape():
|
|||
assert out["album"]["album_type"] == "album"
|
||||
assert out["album"]["total_tracks"] == 0
|
||||
assert out["source"] == "webui_modal"
|
||||
|
||||
|
||||
def test_extract_spotify_track_from_modal_info_converts_trackresult_like_object():
|
||||
track_info = {
|
||||
"spotify_track": SimpleNamespace(
|
||||
title="Song Two",
|
||||
artist="Artist Two",
|
||||
album="Album Two",
|
||||
)
|
||||
}
|
||||
|
||||
out = payloads.extract_spotify_track_from_modal_info(track_info)
|
||||
|
||||
assert out["source"] == "trackresult"
|
||||
assert out["name"] == "Song Two"
|
||||
assert out["artists"] == [{"name": "Artist Two"}]
|
||||
assert out["album"]["name"] == "Album Two"
|
||||
|
||||
|
||||
def test_extract_spotify_track_from_modal_info_reconstructs_from_slskd_result():
|
||||
track_info = {
|
||||
"slskd_result": SimpleNamespace(
|
||||
title="Song Three",
|
||||
artist="Artist Three",
|
||||
album="Album Three",
|
||||
)
|
||||
}
|
||||
|
||||
out = payloads.extract_spotify_track_from_modal_info(track_info)
|
||||
|
||||
assert out["reconstructed"] is True
|
||||
assert out["name"] == "Song Three"
|
||||
assert out["artists"] == [{"name": "Artist Three"}]
|
||||
assert out["album"]["name"] == "Album Three"
|
||||
|
|
|
|||
56
tests/wishlist/test_presence.py
Normal file
56
tests/wishlist/test_presence.py
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
import json
|
||||
|
||||
from core.wishlist import presence
|
||||
|
||||
|
||||
class _FakeCursor:
|
||||
def __init__(self, profile_rows=None, legacy_rows=None, fail_profile_query=False):
|
||||
self.profile_rows = list(profile_rows or [])
|
||||
self.legacy_rows = list(legacy_rows or [])
|
||||
self.fail_profile_query = fail_profile_query
|
||||
self.calls = []
|
||||
self._last_sql = ""
|
||||
|
||||
def execute(self, sql, params=None):
|
||||
self.calls.append((sql, params))
|
||||
self._last_sql = sql
|
||||
if self.fail_profile_query and "WHERE profile_id = ?" in sql:
|
||||
raise RuntimeError("profile_id column missing")
|
||||
|
||||
def fetchall(self):
|
||||
if "WHERE profile_id = ?" in self._last_sql and not self.fail_profile_query:
|
||||
return list(self.profile_rows)
|
||||
return list(self.legacy_rows)
|
||||
|
||||
|
||||
def test_load_wishlist_keys_uses_profile_specific_query():
|
||||
cursor = _FakeCursor(
|
||||
profile_rows=[
|
||||
(json.dumps({"name": "Song One", "artists": [{"name": "Artist One"}]}),),
|
||||
("not-json",),
|
||||
]
|
||||
)
|
||||
|
||||
keys = presence.load_wishlist_keys(cursor, profile_id=7)
|
||||
|
||||
assert keys == {"song one|||artist one"}
|
||||
assert cursor.calls == [
|
||||
("SELECT spotify_data FROM wishlist_tracks WHERE profile_id = ?", (7,)),
|
||||
]
|
||||
|
||||
|
||||
def test_load_wishlist_keys_falls_back_to_legacy_schema():
|
||||
cursor = _FakeCursor(
|
||||
fail_profile_query=True,
|
||||
legacy_rows=[
|
||||
(json.dumps({"name": "Song Two", "artists": [{"name": "Artist Two"}]}),),
|
||||
],
|
||||
)
|
||||
|
||||
keys = presence.load_wishlist_keys(cursor, profile_id=3)
|
||||
|
||||
assert keys == {"song two|||artist two"}
|
||||
assert cursor.calls == [
|
||||
("SELECT spotify_data FROM wishlist_tracks WHERE profile_id = ?", (3,)),
|
||||
("SELECT spotify_data FROM wishlist_tracks", None),
|
||||
]
|
||||
|
|
@ -223,3 +223,112 @@ def test_automatic_wishlist_cleanup_after_db_update_removes_library_matches():
|
|||
|
||||
assert removed == 1
|
||||
assert wishlist_service.removed == [("sp-1", True, None, 1)]
|
||||
|
||||
|
||||
class _CleanupProfilesDatabase:
|
||||
def get_all_profiles(self):
|
||||
return [{"id": 1}]
|
||||
|
||||
|
||||
class _CleanupWishlistService:
|
||||
def __init__(self, tracks):
|
||||
self._tracks = list(tracks)
|
||||
self.removed = []
|
||||
|
||||
def get_wishlist_tracks_for_download(self, profile_id=1):
|
||||
return list(self._tracks)
|
||||
|
||||
def mark_track_download_result(self, spotify_track_id, success, error_message=None, profile_id=1):
|
||||
self.removed.append((spotify_track_id, success, error_message, profile_id))
|
||||
return True
|
||||
|
||||
|
||||
class _CleanupMusicDatabase:
|
||||
def __init__(self):
|
||||
self.track_checks = []
|
||||
|
||||
def check_track_exists(self, track_name, artist_name, confidence_threshold=0.7, server_source=None, album=None):
|
||||
self.track_checks.append((track_name, artist_name, server_source, album))
|
||||
if track_name == "Owned Song" and artist_name == "Artist A":
|
||||
return {"id": "db-track"}, 0.9
|
||||
if track_name == "Broken Song":
|
||||
raise RuntimeError("boom")
|
||||
return None, 0.0
|
||||
|
||||
|
||||
def test_remove_tracks_already_in_library_skips_enhance_tracks_and_handles_errors():
|
||||
wishlist_service = _CleanupWishlistService(
|
||||
[
|
||||
{
|
||||
"name": "Owned Song",
|
||||
"artists": ["Artist A"],
|
||||
"spotify_track_id": "sp-1",
|
||||
"album": {"name": "Album A"},
|
||||
},
|
||||
{
|
||||
"name": "Enhance Song",
|
||||
"artists": [{"name": "Artist B"}],
|
||||
"spotify_track_id": "sp-2",
|
||||
"source_type": "enhance",
|
||||
"album": {"name": "Album B"},
|
||||
},
|
||||
{
|
||||
"name": "Broken Song",
|
||||
"artists": [{"name": "Artist C"}],
|
||||
"spotify_track_id": "sp-3",
|
||||
"album": {"name": "Album C"},
|
||||
},
|
||||
]
|
||||
)
|
||||
music_db = _CleanupMusicDatabase()
|
||||
|
||||
removed = processing.remove_tracks_already_in_library(
|
||||
wishlist_service,
|
||||
_CleanupProfilesDatabase(),
|
||||
music_db,
|
||||
"navidrome",
|
||||
logger=_FakeLogger(),
|
||||
skip_track_fn=lambda track: track.get("source_type") == "enhance",
|
||||
)
|
||||
|
||||
assert removed == 1
|
||||
assert wishlist_service.removed == [("sp-1", True, None, 1)]
|
||||
assert music_db.track_checks == [
|
||||
("Owned Song", "Artist A", "navidrome", "Album A"),
|
||||
("Broken Song", "Artist C", "navidrome", "Album C"),
|
||||
]
|
||||
|
||||
|
||||
def test_finalize_auto_wishlist_completion_with_no_tracks_added_still_resets_state():
|
||||
db = _FakeDB()
|
||||
automation_engine = _FakeAutomationEngine()
|
||||
resets = []
|
||||
activities = []
|
||||
summary = {"tracks_added": 0, "total_failed": 5, "errors": 0}
|
||||
|
||||
result = processing.finalize_auto_wishlist_completion(
|
||||
"batch-2",
|
||||
summary,
|
||||
download_batches={"batch-2": {"current_cycle": "singles"}},
|
||||
tasks_lock=_FakeLock(),
|
||||
reset_processing_state=lambda: resets.append(True),
|
||||
add_activity_item=lambda *args: activities.append(args),
|
||||
automation_engine=automation_engine,
|
||||
db_factory=lambda: db,
|
||||
logger=_FakeLogger(),
|
||||
)
|
||||
|
||||
assert result is summary
|
||||
assert resets == [True]
|
||||
assert activities == []
|
||||
assert automation_engine.events == [
|
||||
(
|
||||
"wishlist_processing_completed",
|
||||
{
|
||||
"tracks_processed": "5",
|
||||
"tracks_found": "0",
|
||||
"tracks_failed": "5",
|
||||
},
|
||||
)
|
||||
]
|
||||
assert db.connection.committed is True
|
||||
|
|
|
|||
|
|
@ -49,6 +49,70 @@ def test_check_and_remove_from_wishlist_uses_search_result_fallback():
|
|||
assert wishlist_service.removed == [("sp-track-1", True, None, 1)]
|
||||
|
||||
|
||||
def test_check_and_remove_from_wishlist_uses_spotify_source_id():
|
||||
fake_db = SimpleNamespace(get_all_profiles=lambda: [{"id": 1}])
|
||||
wishlist_service = _FakeWishlistService(
|
||||
[
|
||||
{
|
||||
"wishlist_id": 11,
|
||||
"spotify_track_id": "sp-track-1",
|
||||
"id": "sp-track-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
context = {
|
||||
"source": "spotify",
|
||||
"track_info": {
|
||||
"id": "sp-track-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
},
|
||||
"search_result": {},
|
||||
"original_search_result": {},
|
||||
}
|
||||
|
||||
resolution.check_and_remove_from_wishlist(
|
||||
context,
|
||||
wishlist_service=wishlist_service,
|
||||
database=fake_db,
|
||||
)
|
||||
|
||||
assert wishlist_service.removed == [("sp-track-1", True, None, 1)]
|
||||
|
||||
|
||||
def test_check_and_remove_from_wishlist_uses_wishlist_id_lookup():
|
||||
fake_db = SimpleNamespace(get_all_profiles=lambda: [{"id": 1}])
|
||||
wishlist_service = _FakeWishlistService(
|
||||
[
|
||||
{
|
||||
"wishlist_id": 22,
|
||||
"spotify_track_id": "sp-track-2",
|
||||
"id": "sp-track-2",
|
||||
"name": "Song Two",
|
||||
"artists": [{"name": "Artist Two"}],
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
context = {
|
||||
"source": "manual",
|
||||
"track_info": {"wishlist_id": 22},
|
||||
"search_result": {},
|
||||
"original_search_result": {},
|
||||
}
|
||||
|
||||
resolution.check_and_remove_from_wishlist(
|
||||
context,
|
||||
wishlist_service=wishlist_service,
|
||||
database=fake_db,
|
||||
)
|
||||
|
||||
assert wishlist_service.removed == [("sp-track-2", True, None, 1)]
|
||||
|
||||
|
||||
def test_check_and_remove_track_from_wishlist_by_metadata_uses_fuzzy_match():
|
||||
fake_db = SimpleNamespace(get_all_profiles=lambda: [{"id": 1}])
|
||||
wishlist_service = _FakeWishlistService(
|
||||
|
|
@ -77,3 +141,41 @@ def test_check_and_remove_track_from_wishlist_by_metadata_uses_fuzzy_match():
|
|||
|
||||
assert removed is True
|
||||
assert wishlist_service.removed == [("sp-track-2", True, None, 1)]
|
||||
|
||||
|
||||
def test_check_and_remove_track_from_wishlist_by_metadata_uses_direct_id_match():
|
||||
fake_db = SimpleNamespace(get_all_profiles=lambda: [{"id": 1}])
|
||||
wishlist_service = _FakeWishlistService([])
|
||||
|
||||
track_data = {
|
||||
"name": "Song Three",
|
||||
"id": "sp-track-3",
|
||||
"artists": [{"name": "Artist Three"}],
|
||||
}
|
||||
|
||||
removed = resolution.check_and_remove_track_from_wishlist_by_metadata(
|
||||
track_data,
|
||||
wishlist_service=wishlist_service,
|
||||
database=fake_db,
|
||||
)
|
||||
|
||||
assert removed is True
|
||||
assert wishlist_service.removed == [("sp-track-3", True, None, 1)]
|
||||
|
||||
|
||||
def test_check_and_remove_track_from_wishlist_by_metadata_returns_false_when_no_match():
|
||||
fake_db = SimpleNamespace(get_all_profiles=lambda: [{"id": 1}])
|
||||
wishlist_service = _FakeWishlistService([])
|
||||
|
||||
removed = resolution.check_and_remove_track_from_wishlist_by_metadata(
|
||||
{
|
||||
"name": "Missing Song",
|
||||
"id": "",
|
||||
"artists": [{"name": "Missing Artist"}],
|
||||
},
|
||||
wishlist_service=wishlist_service,
|
||||
database=fake_db,
|
||||
)
|
||||
|
||||
assert removed is False
|
||||
assert wishlist_service.removed == []
|
||||
|
|
|
|||
|
|
@ -485,3 +485,76 @@ def test_add_album_track_to_wishlist_builds_spotify_payload_and_merges_context()
|
|||
]
|
||||
assert add_call["spotify_track_data"]["duration_ms"] == 1234
|
||||
assert add_call["spotify_track_data"]["explicit"] is True
|
||||
|
||||
|
||||
def test_set_wishlist_cycle_rejects_invalid_cycle():
|
||||
runtime, _service, _db, _logger, _activity_calls = _build_runtime()
|
||||
|
||||
payload, status = set_wishlist_cycle(runtime, "mixes")
|
||||
|
||||
assert status == 400
|
||||
assert payload == {"error": "Invalid cycle. Must be 'albums' or 'singles'"}
|
||||
|
||||
|
||||
def test_remove_album_from_wishlist_matches_album_id():
|
||||
tracks = [
|
||||
{
|
||||
"wishlist_id": 1,
|
||||
"spotify_track_id": "track-1",
|
||||
"id": "track-1",
|
||||
"spotify_data": {
|
||||
"album": {"id": "album-1", "name": "Complete Album"},
|
||||
"artists": [{"name": "Artist One"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"wishlist_id": 2,
|
||||
"spotify_track_id": "track-2",
|
||||
"id": "track-2",
|
||||
"spotify_data": {
|
||||
"album": {"id": "album-2", "name": "Other Album"},
|
||||
"artists": [{"name": "Artist Two"}],
|
||||
},
|
||||
},
|
||||
]
|
||||
runtime, service, _db, _logger, _activity_calls = _build_runtime(tracks=tracks)
|
||||
|
||||
payload, status = remove_album_from_wishlist(runtime, album_id="album-1")
|
||||
|
||||
assert status == 200
|
||||
assert payload == {
|
||||
"success": True,
|
||||
"message": "Removed 1 track(s) from wishlist",
|
||||
"removed_count": 1,
|
||||
}
|
||||
assert service.removed == [("track-1", 1)]
|
||||
|
||||
|
||||
def test_remove_album_from_wishlist_requires_lookup_fields():
|
||||
runtime, _service, _db, _logger, _activity_calls = _build_runtime()
|
||||
|
||||
payload, status = remove_album_from_wishlist(runtime)
|
||||
|
||||
assert status == 400
|
||||
assert payload == {"success": False, "error": "No album_id or album_name provided"}
|
||||
|
||||
|
||||
def test_remove_batch_from_wishlist_rejects_invalid_payload():
|
||||
runtime, _service, _db, _logger, _activity_calls = _build_runtime()
|
||||
|
||||
payload, status = remove_batch_from_wishlist(runtime, "track-1")
|
||||
|
||||
assert status == 400
|
||||
assert payload == {"success": False, "error": "Missing or invalid spotify_track_ids"}
|
||||
|
||||
|
||||
def test_add_album_track_to_wishlist_requires_required_fields():
|
||||
runtime, _service, _db, _logger, _activity_calls = _build_runtime()
|
||||
|
||||
payload, status = add_album_track_to_wishlist(runtime, track=None, artist=None, album=None)
|
||||
|
||||
assert status == 400
|
||||
assert payload == {
|
||||
"success": False,
|
||||
"error": "Missing required fields: track, artist, album",
|
||||
}
|
||||
|
|
|
|||
284
tests/wishlist/test_service.py
Normal file
284
tests/wishlist/test_service.py
Normal file
|
|
@ -0,0 +1,284 @@
|
|||
from types import SimpleNamespace
|
||||
|
||||
from core.wishlist.service import WishlistService
|
||||
|
||||
|
||||
class _FakeWishlistDatabase:
|
||||
def __init__(self, tracks=None, count=None):
|
||||
self.tracks = list(tracks or [])
|
||||
self.count = len(self.tracks) if count is None else count
|
||||
self.add_calls = []
|
||||
self.track_queries = []
|
||||
|
||||
def add_to_wishlist(self, **kwargs):
|
||||
self.add_calls.append(kwargs)
|
||||
return True
|
||||
|
||||
def get_wishlist_tracks(self, limit=None, profile_id=1):
|
||||
self.track_queries.append(("get_wishlist_tracks", limit, profile_id))
|
||||
tracks = list(self.tracks)
|
||||
return tracks if limit is None else tracks[:limit]
|
||||
|
||||
def get_wishlist_count(self, profile_id=1):
|
||||
self.track_queries.append(("get_wishlist_count", profile_id))
|
||||
return self.count
|
||||
|
||||
|
||||
def _build_service(fake_db):
|
||||
service = WishlistService(database_path="test.db")
|
||||
service._database = fake_db
|
||||
return service
|
||||
|
||||
|
||||
def test_add_failed_track_from_modal_normalizes_candidates_and_forwards_payload():
|
||||
fake_db = _FakeWishlistDatabase()
|
||||
service = _build_service(fake_db)
|
||||
|
||||
track_info = {
|
||||
"failure_reason": "Download cancelled",
|
||||
"download_index": 4,
|
||||
"table_index": 4,
|
||||
"candidates": [
|
||||
SimpleNamespace(title="Candidate Song", artist="Candidate Artist", filename="track.flac"),
|
||||
{"title": "kept"},
|
||||
],
|
||||
"spotify_track": {
|
||||
"id": "sp-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
"album": {"name": "Album One"},
|
||||
},
|
||||
}
|
||||
source_context = {"playlist_name": "Playlist One"}
|
||||
|
||||
result = service.add_failed_track_from_modal(
|
||||
track_info,
|
||||
source_type="playlist",
|
||||
source_context=source_context,
|
||||
profile_id=3,
|
||||
)
|
||||
|
||||
assert result is True
|
||||
assert fake_db.add_calls[0]["spotify_track_data"]["id"] == "sp-1"
|
||||
assert fake_db.add_calls[0]["failure_reason"] == "Download cancelled"
|
||||
assert fake_db.add_calls[0]["source_type"] == "playlist"
|
||||
assert fake_db.add_calls[0]["profile_id"] == 3
|
||||
assert fake_db.add_calls[0]["source_info"]["playlist_name"] == "Playlist One"
|
||||
assert fake_db.add_calls[0]["source_info"]["original_modal_data"] == {
|
||||
"download_index": 4,
|
||||
"table_index": 4,
|
||||
"candidates": [
|
||||
{
|
||||
"title": "Candidate Song",
|
||||
"artist": "Candidate Artist",
|
||||
"filename": "track.flac",
|
||||
},
|
||||
{"title": "kept"},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def test_add_failed_track_from_modal_returns_false_when_no_spotify_track_found():
|
||||
fake_db = _FakeWishlistDatabase()
|
||||
service = _build_service(fake_db)
|
||||
|
||||
result = service.add_failed_track_from_modal({"track_info": {}}, profile_id=1)
|
||||
|
||||
assert result is False
|
||||
assert fake_db.add_calls == []
|
||||
|
||||
|
||||
def test_get_wishlist_tracks_for_download_formats_modal_shape():
|
||||
fake_db = _FakeWishlistDatabase(
|
||||
tracks=[
|
||||
{
|
||||
"id": "wl-1",
|
||||
"spotify_track_id": "sp-1",
|
||||
"spotify_data": {
|
||||
"id": "sp-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
"album": {"name": "Album One"},
|
||||
"duration_ms": 321,
|
||||
"preview_url": "https://example.test/preview",
|
||||
"external_urls": {"spotify": "https://open.spotify.com/track/sp-1"},
|
||||
"popularity": 88,
|
||||
"track_number": 7,
|
||||
"disc_number": 2,
|
||||
},
|
||||
"failure_reason": "Download failed",
|
||||
"retry_count": 2,
|
||||
"date_added": "2024-01-01",
|
||||
"last_attempted": "2024-01-02",
|
||||
"source_type": "playlist",
|
||||
"source_info": {"playlist_name": "Playlist One"},
|
||||
}
|
||||
]
|
||||
)
|
||||
service = _build_service(fake_db)
|
||||
|
||||
formatted_tracks = service.get_wishlist_tracks_for_download(limit=1, profile_id=7)
|
||||
|
||||
assert fake_db.track_queries == [("get_wishlist_tracks", 1, 7)]
|
||||
assert formatted_tracks == [
|
||||
{
|
||||
"wishlist_id": "wl-1",
|
||||
"spotify_track_id": "sp-1",
|
||||
"spotify_data": {
|
||||
"id": "sp-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
"album": {"name": "Album One"},
|
||||
"duration_ms": 321,
|
||||
"preview_url": "https://example.test/preview",
|
||||
"external_urls": {"spotify": "https://open.spotify.com/track/sp-1"},
|
||||
"popularity": 88,
|
||||
"track_number": 7,
|
||||
"disc_number": 2,
|
||||
},
|
||||
"failure_reason": "Download failed",
|
||||
"retry_count": 2,
|
||||
"date_added": "2024-01-01",
|
||||
"last_attempted": "2024-01-02",
|
||||
"source_type": "playlist",
|
||||
"source_info": {"playlist_name": "Playlist One"},
|
||||
"id": "sp-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
"album": {"name": "Album One"},
|
||||
"duration_ms": 321,
|
||||
"preview_url": "https://example.test/preview",
|
||||
"external_urls": {"spotify": "https://open.spotify.com/track/sp-1"},
|
||||
"popularity": 88,
|
||||
"track_number": 7,
|
||||
"disc_number": 2,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_check_track_in_wishlist_and_find_matching_wishlist_track_handle_id_variants():
|
||||
fake_db = _FakeWishlistDatabase(
|
||||
tracks=[
|
||||
{
|
||||
"id": "wl-1",
|
||||
"spotify_track_id": "sp-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
"spotify_data": {
|
||||
"id": "sp-1",
|
||||
"name": "Song One",
|
||||
"artists": [{"name": "Artist One"}],
|
||||
"album": {"name": "Album One"},
|
||||
},
|
||||
"failure_reason": "Download failed",
|
||||
"retry_count": 1,
|
||||
"date_added": "2024-01-01",
|
||||
"last_attempted": None,
|
||||
"source_type": "playlist",
|
||||
"source_info": {},
|
||||
},
|
||||
{
|
||||
"id": "sp-2",
|
||||
"spotify_track_id": "sp-2",
|
||||
"name": "Song Two",
|
||||
"artists": ["Artist Two"],
|
||||
"spotify_data": {
|
||||
"id": "sp-2",
|
||||
"name": "Song Two",
|
||||
"artists": [{"name": "Artist Two"}],
|
||||
"album": {"name": "Album Two"},
|
||||
},
|
||||
"failure_reason": "Download failed",
|
||||
"retry_count": 1,
|
||||
"date_added": "2024-01-02",
|
||||
"last_attempted": None,
|
||||
"source_type": "manual",
|
||||
"source_info": {},
|
||||
},
|
||||
]
|
||||
)
|
||||
service = _build_service(fake_db)
|
||||
formatted_tracks = service.get_wishlist_tracks_for_download()
|
||||
|
||||
assert service.check_track_in_wishlist("sp-1") is True
|
||||
assert service.check_track_in_wishlist("sp-2") is True
|
||||
assert service.check_track_in_wishlist("missing") is False
|
||||
assert service.find_matching_wishlist_track("Song One", "Artist One") == formatted_tracks[0]
|
||||
assert service.find_matching_wishlist_track("Song Two", "Artist Two") == formatted_tracks[1]
|
||||
assert service.find_matching_wishlist_track("Missing", "Nobody") is None
|
||||
|
||||
|
||||
def test_get_wishlist_summary_returns_empty_summary_when_count_is_zero():
|
||||
fake_db = _FakeWishlistDatabase(tracks=[], count=0)
|
||||
service = _build_service(fake_db)
|
||||
|
||||
assert service.get_wishlist_summary(profile_id=2) == {
|
||||
"total_tracks": 0,
|
||||
"by_source_type": {},
|
||||
"recent_failures": [],
|
||||
}
|
||||
|
||||
|
||||
def test_get_wishlist_summary_groups_by_source_type_and_limits_recent_failures():
|
||||
fake_db = _FakeWishlistDatabase(
|
||||
tracks=[
|
||||
{
|
||||
"source_type": "playlist",
|
||||
"failure_reason": "f1",
|
||||
"retry_count": 1,
|
||||
"date_added": "2024-01-01",
|
||||
"spotify_data": {"name": "Song A", "artists": [{"name": "Artist A"}]},
|
||||
},
|
||||
{
|
||||
"source_type": "playlist",
|
||||
"failure_reason": "f2",
|
||||
"retry_count": 2,
|
||||
"date_added": "2024-01-02",
|
||||
"spotify_data": {"name": "Song B", "artists": ["Artist B"]},
|
||||
},
|
||||
{
|
||||
"source_type": "album",
|
||||
"failure_reason": "f3",
|
||||
"retry_count": 3,
|
||||
"date_added": "2024-01-03",
|
||||
"spotify_data": {"name": "Song C", "artists": [{"name": "Artist C"}]},
|
||||
},
|
||||
{
|
||||
"source_type": "manual",
|
||||
"failure_reason": "f4",
|
||||
"retry_count": 4,
|
||||
"date_added": "2024-01-04",
|
||||
"spotify_data": {"name": "Song D", "artists": [{"name": "Artist D"}]},
|
||||
},
|
||||
{
|
||||
"source_type": "manual",
|
||||
"failure_reason": "f5",
|
||||
"retry_count": 5,
|
||||
"date_added": "2024-01-05",
|
||||
"spotify_data": {"name": "Song E", "artists": [{"name": "Artist E"}]},
|
||||
},
|
||||
{
|
||||
"source_type": "manual",
|
||||
"failure_reason": "f6",
|
||||
"retry_count": 6,
|
||||
"date_added": "2024-01-06",
|
||||
"spotify_data": {"name": "Song F", "artists": [{"name": "Artist F"}]},
|
||||
},
|
||||
]
|
||||
)
|
||||
service = _build_service(fake_db)
|
||||
|
||||
summary = service.get_wishlist_summary(profile_id=4)
|
||||
|
||||
assert summary["total_tracks"] == 6
|
||||
assert summary["by_source_type"] == {"playlist": 2, "album": 1, "manual": 3}
|
||||
assert len(summary["recent_failures"]) == 5
|
||||
assert summary["recent_failures"][0] == {
|
||||
"name": "Song A",
|
||||
"artist": "Artist A",
|
||||
"failure_reason": "f1",
|
||||
"retry_count": 1,
|
||||
"date_added": "2024-01-01",
|
||||
}
|
||||
assert summary["recent_failures"][1]["artist"] == "Artist B"
|
||||
assert summary["recent_failures"][-1]["name"] == "Song E"
|
||||
Loading…
Reference in a new issue