/
/
/
1"""Tests for the Spotify library tracks listing."""
2
3from collections.abc import AsyncGenerator
4from typing import Any
5from unittest.mock import MagicMock
6
7from music_assistant.providers.spotify.provider import SpotifyProvider
8
9
10def _make_provider(instance_id: str = "spotify--test") -> SpotifyProvider:
11 """Return a Spotify provider with only what the library listing needs."""
12 provider = object.__new__(SpotifyProvider)
13 provider.config = MagicMock(instance_id=instance_id)
14 provider.manifest = MagicMock(domain="spotify")
15 provider.logger = MagicMock()
16 provider.mass = MagicMock()
17 return provider
18
19
20def _saved_track(track_id: str) -> dict[str, Any]:
21 """Return a saved track entry as returned by the me/tracks endpoint."""
22 return {
23 "added_at": "2026-01-01T00:00:00Z",
24 "track": {
25 "id": track_id,
26 "name": track_id,
27 "duration_ms": 180000,
28 "external_urls": {"spotify": f"https://open.spotify.com/track/{track_id}"},
29 "is_local": False,
30 "is_playable": True,
31 "explicit": False,
32 },
33 }
34
35
36async def test_library_tracks_skips_entry_without_track() -> None:
37 """
38 A saved track entry that carries no track at all is skipped.
39
40 Spotify returns such an entry for a track that is no longer available, which would
41 otherwise abort the whole track sync.
42 """
43 provider = _make_provider()
44 # a valid track after the empty ones, so stopping at an empty entry fails this test
45 items: list[Any] = [
46 _saved_track("track1"),
47 {"added_at": "2026-01-01T00:00:00Z", "track": None},
48 None,
49 _saved_track("track2"),
50 ]
51
52 async def _iter_items() -> AsyncGenerator[Any]:
53 for item in items:
54 yield item
55
56 provider._get_all_items = MagicMock(return_value=_iter_items()) # type: ignore[method-assign]
57
58 tracks = [track async for track in provider.get_library_tracks()]
59
60 assert [track.item_id for track in tracks] == ["track1", "track2"]
61