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