music-assistant-server

1"""Test Tidal Media Manager."""
2
3import json
4import pathlib
5from typing import Any
6from unittest.mock import Mock, patch
7
8from music_assistant_models.enums import MediaType
9from music_assistant_models.errors import RetriesExhausted
10from music_assistant_models.media_items import Playlist
11
12from music_assistant.providers.tidal.jsonapi import JsonApiDocument
13from music_assistant.providers.tidal.media import TidalMediaManager
14
15FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures" / "v2"
16
17
18def _load_raw(name: str) -> dict[str, Any]:
19    with open(FIXTURES_DIR / name) as f:
20        data: dict[str, Any] = json.load(f)
21        return data
22
23
24def _load_doc(name: str) -> JsonApiDocument:
25    return JsonApiDocument(_load_raw(name))
26
27
28async def test_search(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
29    """Test search reads all types from the official searchResults endpoint."""
30    provider_mock.api.get_jsonapi.return_value = _load_doc("search.json")
31
32    results = await media_manager.search(
33        "lukas graham",
34        [MediaType.ARTIST, MediaType.ALBUM, MediaType.TRACK, MediaType.PLAYLIST],
35        limit=5,
36    )
37
38    # 20 of each in the fixture, capped to the limit
39    assert len(results.tracks) == 5
40    assert len(results.albums) == 5
41    assert len(results.artists) == 5
42    assert len(results.playlists) == 5
43    assert results.tracks[0].name == "7 Years"
44    first_playlist = results.playlists[0]
45    assert isinstance(first_playlist, Playlist)
46    assert first_playlist.is_editable is False
47
48    provider_mock.api.get_jsonapi.assert_called_with(
49        "searchResults",
50        params={"filter[query]": "lukas graham"},
51        include=[
52            "tracks.artists",
53            "tracks.albums.coverArt",
54            "albums.coverArt",
55            "artists.profileArt",
56            "playlists.coverArt",
57        ],
58    )
59
60
61async def test_search_single_type_passes_query_verbatim(
62    media_manager: TidalMediaManager, provider_mock: Mock
63) -> None:
64    """Test search requests only the includes for the wanted types, query as a filter."""
65    provider_mock.api.get_jsonapi.return_value = _load_doc("search.json")
66
67    results = await media_manager.search("AC/DC", [MediaType.ARTIST])
68
69    assert results.tracks == []
70    provider_mock.api.get_jsonapi.assert_called_with(
71        "searchResults", params={"filter[query]": "AC/DC"}, include=["artists.profileArt"]
72    )
73
74
75async def test_search_no_types_short_circuits(
76    media_manager: TidalMediaManager, provider_mock: Mock
77) -> None:
78    """Test search with no supported media types does not hit the API."""
79    results = await media_manager.search("x", [])
80
81    assert results.tracks == []
82    provider_mock.api.get_jsonapi.assert_not_called()
83
84
85@patch("music_assistant.providers.tidal.media.parse_artist_v2")
86async def test_get_artist(
87    mock_parse_artist: Mock, media_manager: TidalMediaManager, provider_mock: Mock
88) -> None:
89    """Test get_artist uses the official JSON:API endpoint."""
90    doc = Mock()
91    provider_mock.api.get_jsonapi.return_value = doc
92    mock_parse_artist.return_value = Mock(item_id="1")
93
94    artist = await media_manager.get_artist("1")
95
96    assert artist.item_id == "1"
97    provider_mock.api.get_jsonapi.assert_called_with(
98        "artists/1", include=["profileArt", "biography"]
99    )
100    mock_parse_artist.assert_called_once_with(provider_mock, doc, doc.data)
101
102
103@patch("music_assistant.providers.tidal.media.parse_album_v2")
104async def test_get_album(
105    mock_parse_album: Mock, media_manager: TidalMediaManager, provider_mock: Mock
106) -> None:
107    """Test get_album uses the official JSON:API endpoint."""
108    doc = Mock()
109    provider_mock.api.get_jsonapi.return_value = doc
110    mock_parse_album.return_value = Mock(item_id="1")
111
112    album = await media_manager.get_album("1")
113
114    assert album.item_id == "1"
115    provider_mock.api.get_jsonapi.assert_called_with(
116        "albums/1", include=["artists", "coverArt", "genres"]
117    )
118    mock_parse_album.assert_called_once_with(provider_mock, doc, doc.data)
119
120
121@patch("music_assistant.providers.tidal.media.parse_track_v2")
122async def test_get_track(
123    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
124) -> None:
125    """Test get_track fetches the official track and applies unofficial lyrics."""
126    doc = Mock()
127    provider_mock.api.get_jsonapi.return_value = doc
128    track = Mock(item_id="1")
129    track.metadata = Mock()
130    mock_parse_track.return_value = track
131    provider_mock.api.get.return_value = {"lyrics": "Test Lyrics", "subtitles": "Synced"}
132
133    result = await media_manager.get_track("1")
134
135    assert result.item_id == "1"
136    provider_mock.api.get_jsonapi.assert_called_with(
137        "tracks/1", include=["artists", "albums", "albums.coverArt", "genres", "credits"]
138    )
139    provider_mock.api.get.assert_called_with("tracks/1/lyrics")
140    assert track.metadata.lyrics == "Test Lyrics"
141    assert track.metadata.lrc_lyrics == "Synced"
142
143
144@patch("music_assistant.providers.tidal.media.parse_track_v2")
145async def test_get_track_tolerates_lyrics_failure(
146    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
147) -> None:
148    """Test get_track still returns the track when the lyrics lookup fails."""
149    doc = Mock()
150    provider_mock.api.get_jsonapi.return_value = doc
151    track = Mock(item_id="1")
152    mock_parse_track.return_value = track
153    provider_mock.api.get.side_effect = RetriesExhausted("lyrics lookup failed")
154
155    result = await media_manager.get_track("1")
156
157    assert result.item_id == "1"
158    mock_parse_track.assert_called_once_with(provider_mock, doc, doc.data)
159
160
161async def test_get_album_tracks(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
162    """Test album tracks read from the official relationship endpoint with ordering."""
163    doc = _load_doc("album_items.json")
164
165    async def _pages(*_a: Any, **_k: Any) -> Any:
166        yield doc
167
168    provider_mock.api.paginate_jsonapi = _pages
169
170    tracks = await media_manager.get_album_tracks("58756127")
171
172    assert len(tracks) == 11
173    assert tracks[0].name == "7 Years"
174    assert tracks[0].track_number == 1
175    assert tracks[0].disc_number == 1
176    assert tracks[0].album is not None
177
178
179async def test_get_album_tracks_skips_videos(
180    media_manager: TidalMediaManager, provider_mock: Mock
181) -> None:
182    """Test a music video in the items relationship is not parsed as a track."""
183    raw = _load_raw("album_items.json")
184    raw["data"].insert(
185        0, {"id": "99999", "type": "videos", "meta": {"volumeNumber": 1, "trackNumber": 1}}
186    )
187    raw["included"].append(
188        {"id": "99999", "type": "videos", "attributes": {"title": "Bonus Video"}}
189    )
190    doc = JsonApiDocument(raw)
191
192    async def _pages(*_a: Any, **_k: Any) -> Any:
193        yield doc
194
195    provider_mock.api.paginate_jsonapi = _pages
196
197    tracks = await media_manager.get_album_tracks("58756127")
198
199    assert len(tracks) == 11
200    assert all(track.item_id != "99999" for track in tracks)
201
202
203async def test_get_artist_albums(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
204    """Test artist albums read from the official relationship endpoint."""
205    doc = _load_doc("artist_albums.json")
206
207    async def _pages(*_a: Any, **_k: Any) -> Any:
208        yield doc
209
210    provider_mock.api.paginate_jsonapi = _pages
211
212    albums = await media_manager.get_artist_albums("4184211")
213
214    assert len(albums) == 20
215    assert all(album.item_id for album in albums)
216
217
218async def test_get_artist_toptracks(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
219    """Test artist top tracks read from the official relationship endpoint."""
220    provider_mock.api.get_jsonapi.return_value = _load_doc("artist_toptracks.json")
221
222    tracks = await media_manager.get_artist_toptracks("4184211")
223
224    assert len(tracks) == 20
225    provider_mock.api.get_jsonapi.assert_called_with(
226        "artists/4184211/relationships/tracks",
227        params={"collapseBy": "FINGERPRINT"},
228        include=["tracks.artists", "tracks.albums.coverArt"],
229        replace_media="tracks",
230    )
231
232
233async def test_get_similar_tracks_respects_limit(
234    media_manager: TidalMediaManager, provider_mock: Mock
235) -> None:
236    """Test similar tracks are read from the official endpoint and capped to the limit."""
237    provider_mock.api.get_jsonapi.return_value = _load_doc("similar_tracks.json")
238
239    tracks = await media_manager.get_similar_tracks("58756128", limit=5)
240
241    assert len(tracks) == 5
242
243
244async def test_get_similar_artists(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
245    """Test similar artists read from the official relationship endpoint."""
246    provider_mock.api.get_jsonapi.return_value = _load_doc("similar_artists.json")
247
248    artists = await media_manager.get_similar_artists("4184211")
249
250    assert len(artists) == 20
251    provider_mock.api.get_jsonapi.assert_called_with(
252        "artists/4184211/relationships/similarArtists",
253        include=["similarArtists.profileArt"],
254    )
255
256
257@patch("music_assistant.providers.tidal.media.parse_playlist")
258async def test_get_playlist(
259    mock_parse_playlist: Mock, media_manager: TidalMediaManager, provider_mock: Mock
260) -> None:
261    """Test get_playlist."""
262    provider_mock.api.get.return_value = {"uuid": "1", "title": "Test Playlist"}
263    mock_parse_playlist.return_value = Mock(item_id="1")
264
265    playlist = await media_manager.get_playlist("1")
266
267    assert playlist.item_id == "1"
268    provider_mock.api.get.assert_called_with("playlists/1")
269    mock_parse_playlist.assert_called_once()
270
271
272@patch("music_assistant.providers.tidal.media.parse_track")
273async def test_get_playlist_tracks(
274    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
275) -> None:
276    """Test get_playlist_tracks."""
277    provider_mock.api.get.return_value = {"items": [{"id": 1}]}
278    mock_parse_track.return_value = Mock(item_id="1")
279
280    tracks = await media_manager.get_playlist_tracks("1")
281
282    assert len(tracks) == 1
283    assert tracks[0].item_id == "1"
284    provider_mock.api.get.assert_called_with(
285        "playlists/1/tracks",
286        params={"limit": 200, "offset": 0},
287    )
288
289
290async def test_get_playlist_favorite_tracks(
291    media_manager: TidalMediaManager, provider_mock: Mock
292) -> None:
293    """Test get_playlist returns the virtual favorite tracks playlist without an API call."""
294    playlist = await media_manager.get_playlist("favorite_tracks")
295
296    assert playlist.item_id == "favorite_tracks"
297    assert playlist.name == "Favorite Tracks"
298    assert not playlist.is_editable
299    provider_mock.api.get.assert_not_called()
300
301
302@patch("music_assistant.providers.tidal.media.parse_track_v2")
303async def test_get_playlist_tracks_favorite_tracks(
304    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
305) -> None:
306    """Test get_playlist_tracks walks the official collection with sequential positions."""
307    doc = JsonApiDocument(
308        {
309            "data": [
310                {"type": "tracks", "id": "1"},
311                {"type": "tracks", "id": "2"},
312            ],
313            "included": [
314                {"type": "tracks", "id": "1", "attributes": {}},
315                {"type": "tracks", "id": "2", "attributes": {}},
316            ],
317        }
318    )
319
320    async def _pages(*_a: Any, **_k: Any) -> Any:
321        yield doc
322
323    provider_mock.api.paginate_jsonapi = _pages
324    mock_parse_track.side_effect = [Mock(item_id="1"), Mock(item_id="2")]
325
326    tracks = await media_manager.get_playlist_tracks("favorite_tracks", page=0)
327
328    assert len(tracks) == 2
329    assert tracks[0].item_id == "1"
330    assert tracks[0].position == 1
331    assert tracks[1].item_id == "2"
332    assert tracks[1].position == 2
333
334
335@patch("music_assistant.providers.tidal.media.parse_track_v2")
336async def test_favorite_tracks_walk_feeds_churn_cache(
337    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
338) -> None:
339    """
340    Test the favourites walk hands each linkage item to note_replaced_track.
341
342    The read is made with replaceMedia, so Tidal already computed any stale->live
343    replacement pairs; discarding them would waste the largest healing source.
344    """
345    items = [{"type": "tracks", "id": "1"}, {"type": "tracks", "id": "2"}]
346    doc = JsonApiDocument(
347        {
348            "data": items,
349            "included": [
350                {"type": "tracks", "id": "1", "attributes": {}},
351                {"type": "tracks", "id": "2", "attributes": {}},
352            ],
353        }
354    )
355
356    async def _pages(*_a: Any, **_k: Any) -> Any:
357        yield doc
358
359    provider_mock.api.paginate_jsonapi = _pages
360    mock_parse_track.side_effect = [Mock(item_id="1"), Mock(item_id="2")]
361
362    await media_manager.get_playlist_tracks("favorite_tracks", page=0)
363
364    assert provider_mock.note_replaced_track.call_count == 2
365    provider_mock.note_replaced_track.assert_any_call(items[0])
366    provider_mock.note_replaced_track.assert_any_call(items[1])
367
368
369async def test_get_playlist_tracks_favorite_tracks_later_page(
370    media_manager: TidalMediaManager, provider_mock: Mock
371) -> None:
372    """Test get_playlist_tracks returns nothing for pages after the first, without a fetch."""
373    provider_mock.api.paginate_jsonapi = Mock()
374
375    tracks = await media_manager.get_playlist_tracks("favorite_tracks", page=1)
376
377    assert tracks == []
378    provider_mock.api.paginate_jsonapi.assert_not_called()
379