music-assistant-server

1"""Test Tidal Media Manager."""
2
3import json
4import pathlib
5from typing import Any
6from unittest.mock import Mock, patch
7
8import pytest
9from aiohttp.client_exceptions import ClientError
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import MediaNotFoundError, RetriesExhausted
12from music_assistant_models.media_items import Playlist
13
14from music_assistant.providers.tidal.jsonapi import JsonApiDocument
15from music_assistant.providers.tidal.media import TidalMediaManager
16
17FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures" / "v2"
18
19
20def _load_raw(name: str) -> dict[str, Any]:
21    with open(FIXTURES_DIR / name) as f:
22        data: dict[str, Any] = json.load(f)
23        return data
24
25
26def _load_doc(name: str) -> JsonApiDocument:
27    return JsonApiDocument(_load_raw(name))
28
29
30def _break_resource(raw: dict[str, Any], resource_id: str) -> dict[str, Any]:
31    """Blank the title of one included resource, which its parser cannot handle."""
32    for resource in raw["included"]:
33        if resource["id"] == resource_id:
34            attributes = resource["attributes"]
35            # artists carry a name where tracks and albums carry a title
36            attributes["name" if "name" in attributes else "title"] = None
37            return raw
38    raise AssertionError(f"resource {resource_id} not in fixture")
39
40
41async def test_search(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
42    """Test search reads all types from the official searchResults endpoint."""
43    provider_mock.api.get_jsonapi.return_value = _load_doc("search.json")
44
45    results = await media_manager.search(
46        "lukas graham",
47        [MediaType.ARTIST, MediaType.ALBUM, MediaType.TRACK, MediaType.PLAYLIST],
48        limit=5,
49    )
50
51    # 20 of each in the fixture, capped to the limit
52    assert len(results.tracks) == 5
53    assert len(results.albums) == 5
54    assert len(results.artists) == 5
55    assert len(results.playlists) == 5
56    assert results.tracks[0].name == "7 Years"
57    first_playlist = results.playlists[0]
58    assert isinstance(first_playlist, Playlist)
59    assert first_playlist.is_editable is False
60
61    provider_mock.api.get_jsonapi.assert_called_with(
62        "searchResults",
63        params={"filter[query]": "lukas graham"},
64        include=[
65            "tracks.artists",
66            "tracks.albums.coverArt",
67            "albums.coverArt",
68            "artists.profileArt",
69            "playlists.coverArt",
70        ],
71    )
72
73
74async def test_search_single_type_passes_query_verbatim(
75    media_manager: TidalMediaManager, provider_mock: Mock
76) -> None:
77    """Test search requests only the includes for the wanted types, query as a filter."""
78    provider_mock.api.get_jsonapi.return_value = _load_doc("search.json")
79
80    results = await media_manager.search("AC/DC", [MediaType.ARTIST])
81
82    assert results.tracks == []
83    provider_mock.api.get_jsonapi.assert_called_with(
84        "searchResults", params={"filter[query]": "AC/DC"}, include=["artists.profileArt"]
85    )
86
87
88async def test_search_no_types_short_circuits(
89    media_manager: TidalMediaManager, provider_mock: Mock
90) -> None:
91    """Test search with no supported media types does not hit the API."""
92    results = await media_manager.search("x", [])
93
94    assert results.tracks == []
95    provider_mock.api.get_jsonapi.assert_not_called()
96
97
98@patch("music_assistant.providers.tidal.media.parse_artist_v2")
99async def test_get_artist(
100    mock_parse_artist: Mock, media_manager: TidalMediaManager, provider_mock: Mock
101) -> None:
102    """Test get_artist uses the official JSON:API endpoint."""
103    doc = Mock()
104    provider_mock.api.get_jsonapi.return_value = doc
105    mock_parse_artist.return_value = Mock(item_id="1")
106
107    artist = await media_manager.get_artist("1")
108
109    assert artist.item_id == "1"
110    provider_mock.api.get_jsonapi.assert_called_with(
111        "artists/1", include=["profileArt", "biography"]
112    )
113    mock_parse_artist.assert_called_once_with(provider_mock, doc, doc.data)
114
115
116@patch("music_assistant.providers.tidal.media.parse_album_v2")
117async def test_get_album(
118    mock_parse_album: Mock, media_manager: TidalMediaManager, provider_mock: Mock
119) -> None:
120    """Test get_album uses the official JSON:API endpoint."""
121    doc = Mock()
122    provider_mock.api.get_jsonapi.return_value = doc
123    mock_parse_album.return_value = Mock(item_id="1")
124
125    album = await media_manager.get_album("1")
126
127    assert album.item_id == "1"
128    provider_mock.api.get_jsonapi.assert_called_with(
129        "albums/1", include=["artists", "coverArt", "genres"]
130    )
131    mock_parse_album.assert_called_once_with(provider_mock, doc, doc.data)
132
133
134@patch("music_assistant.providers.tidal.media.parse_track_v2")
135async def test_get_track(
136    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
137) -> None:
138    """Test get_track fetches the official track and applies unofficial lyrics."""
139    doc = Mock()
140    provider_mock.api.get_jsonapi.return_value = doc
141    track = Mock(item_id="1")
142    track.metadata = Mock()
143    mock_parse_track.return_value = track
144    provider_mock.api.get.return_value = {"lyrics": "Test Lyrics", "subtitles": "Synced"}
145
146    result = await media_manager.get_track("1")
147
148    assert result.item_id == "1"
149    provider_mock.api.get_jsonapi.assert_called_with(
150        "tracks/1", include=["artists", "albums", "albums.coverArt", "genres", "credits"]
151    )
152    provider_mock.api.get.assert_called_with("tracks/1/lyrics")
153    assert track.metadata.lyrics == "Test Lyrics"
154    assert track.metadata.lrc_lyrics == "Synced"
155
156
157@patch("music_assistant.providers.tidal.media.parse_track_v2")
158async def test_get_track_tolerates_lyrics_failure(
159    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
160) -> None:
161    """Test get_track still returns the track when the lyrics lookup fails."""
162    doc = Mock()
163    provider_mock.api.get_jsonapi.return_value = doc
164    track = Mock(item_id="1")
165    mock_parse_track.return_value = track
166    provider_mock.api.get.side_effect = RetriesExhausted("lyrics lookup failed")
167
168    result = await media_manager.get_track("1")
169
170    assert result.item_id == "1"
171    mock_parse_track.assert_called_once_with(provider_mock, doc, doc.data)
172
173
174async def test_get_album_tracks(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
175    """Test album tracks read from the official relationship endpoint with ordering."""
176    doc = _load_doc("album_items.json")
177
178    async def _pages(*_a: Any, **_k: Any) -> Any:
179        yield doc
180
181    provider_mock.api.paginate_jsonapi = _pages
182
183    tracks = await media_manager.get_album_tracks("58756127")
184
185    assert len(tracks) == 11
186    assert tracks[0].name == "7 Years"
187    assert tracks[0].track_number == 1
188    assert tracks[0].disc_number == 1
189    assert tracks[0].album is not None
190
191
192async def test_get_album_tracks_skips_videos(
193    media_manager: TidalMediaManager, provider_mock: Mock
194) -> None:
195    """Test a music video in the items relationship is not parsed as a track."""
196    raw = _load_raw("album_items.json")
197    raw["data"].insert(
198        0, {"id": "99999", "type": "videos", "meta": {"volumeNumber": 1, "trackNumber": 1}}
199    )
200    raw["included"].append(
201        {"id": "99999", "type": "videos", "attributes": {"title": "Bonus Video"}}
202    )
203    doc = JsonApiDocument(raw)
204
205    async def _pages(*_a: Any, **_k: Any) -> Any:
206        yield doc
207
208    provider_mock.api.paginate_jsonapi = _pages
209
210    tracks = await media_manager.get_album_tracks("58756127")
211
212    assert len(tracks) == 11
213    assert all(track.item_id != "99999" for track in tracks)
214
215
216async def test_get_album_tracks_skips_unparsable_track(
217    media_manager: TidalMediaManager, provider_mock: Mock
218) -> None:
219    """Test one unusable track does not take the rest of the album down with it."""
220    doc = JsonApiDocument(_break_resource(_load_raw("album_items.json"), "58756128"))
221
222    async def _pages(*_a: Any, **_k: Any) -> Any:
223        yield doc
224
225    provider_mock.api.paginate_jsonapi = _pages
226
227    tracks = await media_manager.get_album_tracks("58756127")
228
229    assert len(tracks) == 10
230    assert all(track.item_id != "58756128" for track in tracks)
231    provider_mock.logger.warning.assert_called_once()
232
233
234async def test_get_album_tracks_keeps_tracks_from_earlier_pages(
235    media_manager: TidalMediaManager, provider_mock: Mock
236) -> None:
237    """Test an unusable track on a later page does not discard the earlier pages."""
238    first_page = _load_doc("album_items.json")
239    second_page = JsonApiDocument(
240        {
241            "data": [
242                {"id": "998", "type": "tracks", "meta": {"volumeNumber": 1, "trackNumber": 12}},
243                {"id": "999", "type": "tracks", "meta": {"volumeNumber": 1, "trackNumber": 13}},
244            ],
245            "included": [
246                {"id": "998", "type": "tracks", "attributes": {"title": None}},
247                {
248                    "id": "999",
249                    "type": "tracks",
250                    "attributes": {"title": "Hidden Track", "duration": "PT3M0S"},
251                },
252            ],
253        }
254    )
255
256    async def _pages(*_a: Any, **_k: Any) -> Any:
257        yield first_page
258        yield second_page
259
260    provider_mock.api.paginate_jsonapi = _pages
261
262    tracks = await media_manager.get_album_tracks("58756127")
263
264    assert len(tracks) == 12
265    assert tracks[0].name == "7 Years"
266    assert tracks[-1].name == "Hidden Track"
267    assert tracks[-1].track_number == 13
268
269
270async def test_get_album_tracks_fetch_failure_propagates(
271    media_manager: TidalMediaManager, provider_mock: Mock
272) -> None:
273    """Test a failure to read a page is raised instead of returning a partial album."""
274
275    async def _pages(*_a: Any, **_k: Any) -> Any:
276        yield _load_doc("album_items.json")
277        raise ClientError("connection lost")
278
279    provider_mock.api.paginate_jsonapi = _pages
280
281    with pytest.raises(ClientError):
282        await media_manager.get_album_tracks("58756127")
283
284
285async def test_get_album_tracks_missing_album_still_not_found(
286    media_manager: TidalMediaManager, provider_mock: Mock
287) -> None:
288    """Test an album that is gone is still reported as not found."""
289
290    async def _pages(*_a: Any, **_k: Any) -> Any:
291        for _ in ():  # never yields: the api client raises while fetching the first page
292            yield
293        raise MediaNotFoundError("Item not found: albums/1/relationships/items")
294
295    provider_mock.api.paginate_jsonapi = _pages
296
297    with pytest.raises(MediaNotFoundError):
298        await media_manager.get_album_tracks("1")
299
300
301async def test_get_artist_albums(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
302    """Test artist albums read from the official relationship endpoint."""
303    doc = _load_doc("artist_albums.json")
304
305    async def _pages(*_a: Any, **_k: Any) -> Any:
306        yield doc
307
308    provider_mock.api.paginate_jsonapi = _pages
309
310    albums = await media_manager.get_artist_albums("4184211")
311
312    assert len(albums) == 20
313    assert all(album.item_id for album in albums)
314
315
316async def test_get_artist_toptracks(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
317    """Test artist top tracks read from the official relationship endpoint."""
318    provider_mock.api.get_jsonapi.return_value = _load_doc("artist_toptracks.json")
319
320    tracks = await media_manager.get_artist_toptracks("4184211")
321
322    assert len(tracks) == 20
323    provider_mock.api.get_jsonapi.assert_called_with(
324        "artists/4184211/relationships/tracks",
325        params={"collapseBy": "FINGERPRINT"},
326        include=["tracks.artists", "tracks.albums.coverArt"],
327        replace_media="tracks",
328    )
329
330
331async def test_get_artist_albums_skips_unparsable_album(
332    media_manager: TidalMediaManager, provider_mock: Mock
333) -> None:
334    """Test one unusable album does not empty the artist's discography."""
335    doc = JsonApiDocument(_break_resource(_load_raw("artist_albums.json"), "541774424"))
336
337    async def _pages(*_a: Any, **_k: Any) -> Any:
338        yield doc
339
340    provider_mock.api.paginate_jsonapi = _pages
341
342    albums = await media_manager.get_artist_albums("4184211")
343
344    assert len(albums) == 19
345    assert all(album.item_id != "541774424" for album in albums)
346
347
348async def test_get_artist_toptracks_skips_unparsable_track(
349    media_manager: TidalMediaManager, provider_mock: Mock
350) -> None:
351    """Test one unusable track does not empty the artist's top tracks."""
352    provider_mock.api.get_jsonapi.return_value = JsonApiDocument(
353        _break_resource(_load_raw("artist_toptracks.json"), "58503071")
354    )
355
356    tracks = await media_manager.get_artist_toptracks("4184211")
357
358    assert len(tracks) == 19
359    assert all(track.item_id != "58503071" for track in tracks)
360
361
362async def test_get_artist_toptracks_fetch_failure_propagates(
363    media_manager: TidalMediaManager, provider_mock: Mock
364) -> None:
365    """Test a failure to read the top tracks is raised instead of reported as empty."""
366    provider_mock.api.get_jsonapi.side_effect = ClientError("connection lost")
367
368    with pytest.raises(ClientError):
369        await media_manager.get_artist_toptracks("4184211")
370
371
372async def test_get_similar_tracks_respects_limit(
373    media_manager: TidalMediaManager, provider_mock: Mock
374) -> None:
375    """Test similar tracks are read from the official endpoint and capped to the limit."""
376    provider_mock.api.get_jsonapi.return_value = _load_doc("similar_tracks.json")
377
378    tracks = await media_manager.get_similar_tracks("58756128", limit=5)
379
380    assert len(tracks) == 5
381
382
383async def test_get_similar_tracks_fills_the_limit_around_an_unparsable_track(
384    media_manager: TidalMediaManager, provider_mock: Mock
385) -> None:
386    """Test the limit is still filled when one of the similar tracks cannot be parsed."""
387    raw = _load_raw("similar_tracks.json")
388    provider_mock.api.get_jsonapi.return_value = JsonApiDocument(
389        _break_resource(raw, raw["data"][0]["id"])
390    )
391
392    tracks = await media_manager.get_similar_tracks("58756128", limit=5)
393
394    assert len(tracks) == 5
395    assert all(track.item_id != raw["data"][0]["id"] for track in tracks)
396
397
398async def test_get_similar_artists(media_manager: TidalMediaManager, provider_mock: Mock) -> None:
399    """Test similar artists read from the official relationship endpoint."""
400    provider_mock.api.get_jsonapi.return_value = _load_doc("similar_artists.json")
401
402    artists = await media_manager.get_similar_artists("4184211")
403
404    assert len(artists) == 20
405    provider_mock.api.get_jsonapi.assert_called_with(
406        "artists/4184211/relationships/similarArtists",
407        include=["similarArtists.profileArt"],
408    )
409
410
411async def test_get_similar_artists_skips_unparsable_artist(
412    media_manager: TidalMediaManager, provider_mock: Mock
413) -> None:
414    """Test one unusable artist does not empty the similar artists."""
415    raw = _load_raw("similar_artists.json")
416    provider_mock.api.get_jsonapi.return_value = JsonApiDocument(
417        _break_resource(raw, raw["data"][0]["id"])
418    )
419
420    artists = await media_manager.get_similar_artists("4184211")
421
422    assert len(artists) == 19
423    assert all(artist.item_id != raw["data"][0]["id"] for artist in artists)
424
425
426async def test_search_skips_unparsable_track(
427    media_manager: TidalMediaManager, provider_mock: Mock
428) -> None:
429    """Test one unusable track does not drop the whole search result."""
430    raw = _load_raw("search.json")
431    broken_id = raw["data"][0]["relationships"]["tracks"]["data"][0]["id"]
432    provider_mock.api.get_jsonapi.return_value = JsonApiDocument(_break_resource(raw, broken_id))
433
434    results = await media_manager.search("lukas graham", [MediaType.TRACK], limit=5)
435
436    assert len(results.tracks) == 4
437    assert all(track.item_id != broken_id for track in results.tracks)
438
439
440@patch("music_assistant.providers.tidal.media.parse_track_v2")
441async def test_favorite_tracks_skips_unparsable_track(
442    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
443) -> None:
444    """Test one unusable favourite does not fail the whole favourites playlist."""
445    doc = JsonApiDocument(
446        {
447            "data": [
448                {"type": "tracks", "id": "1"},
449                {"type": "tracks", "id": "2"},
450                {"type": "tracks", "id": "3"},
451            ],
452            "included": [
453                {"type": "tracks", "id": "1", "attributes": {}},
454                {"type": "tracks", "id": "2", "attributes": {}},
455                {"type": "tracks", "id": "3", "attributes": {}},
456            ],
457        }
458    )
459
460    async def _pages(*_a: Any, **_k: Any) -> Any:
461        yield doc
462
463    provider_mock.api.paginate_jsonapi = _pages
464    mock_parse_track.side_effect = [Mock(item_id="1"), KeyError("id"), Mock(item_id="3")]
465
466    tracks = await media_manager.get_playlist_tracks("favorite_tracks", page=0)
467
468    assert [track.item_id for track in tracks] == ["1", "3"]
469    assert [track.position for track in tracks] == [1, 2]
470
471
472@patch("music_assistant.providers.tidal.media.parse_playlist")
473async def test_get_playlist(
474    mock_parse_playlist: Mock, media_manager: TidalMediaManager, provider_mock: Mock
475) -> None:
476    """Test get_playlist."""
477    provider_mock.api.get.return_value = {"uuid": "1", "title": "Test Playlist"}
478    mock_parse_playlist.return_value = Mock(item_id="1")
479
480    playlist = await media_manager.get_playlist("1")
481
482    assert playlist.item_id == "1"
483    provider_mock.api.get.assert_called_with("playlists/1")
484    mock_parse_playlist.assert_called_once()
485
486
487@patch("music_assistant.providers.tidal.media.parse_track")
488async def test_get_playlist_tracks(
489    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
490) -> None:
491    """Test get_playlist_tracks."""
492    provider_mock.api.get.return_value = {"items": [{"id": 1}]}
493    mock_parse_track.return_value = Mock(item_id="1")
494
495    tracks = await media_manager.get_playlist_tracks("1")
496
497    assert len(tracks) == 1
498    assert tracks[0].item_id == "1"
499    provider_mock.api.get.assert_called_with(
500        "playlists/1/tracks",
501        params={"limit": 200, "offset": 0},
502    )
503
504
505async def test_get_playlist_favorite_tracks(
506    media_manager: TidalMediaManager, provider_mock: Mock
507) -> None:
508    """Test get_playlist returns the virtual favorite tracks playlist without an API call."""
509    playlist = await media_manager.get_playlist("favorite_tracks")
510
511    assert playlist.item_id == "favorite_tracks"
512    assert playlist.name == "Favorite Tracks"
513    assert not playlist.is_editable
514    provider_mock.api.get.assert_not_called()
515
516
517@patch("music_assistant.providers.tidal.media.parse_track_v2")
518async def test_get_playlist_tracks_favorite_tracks(
519    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
520) -> None:
521    """Test get_playlist_tracks walks the official collection with sequential positions."""
522    doc = JsonApiDocument(
523        {
524            "data": [
525                {"type": "tracks", "id": "1"},
526                {"type": "tracks", "id": "2"},
527            ],
528            "included": [
529                {"type": "tracks", "id": "1", "attributes": {}},
530                {"type": "tracks", "id": "2", "attributes": {}},
531            ],
532        }
533    )
534
535    async def _pages(*_a: Any, **_k: Any) -> Any:
536        yield doc
537
538    provider_mock.api.paginate_jsonapi = _pages
539    mock_parse_track.side_effect = [Mock(item_id="1"), Mock(item_id="2")]
540
541    tracks = await media_manager.get_playlist_tracks("favorite_tracks", page=0)
542
543    assert len(tracks) == 2
544    assert tracks[0].item_id == "1"
545    assert tracks[0].position == 1
546    assert tracks[1].item_id == "2"
547    assert tracks[1].position == 2
548
549
550@patch("music_assistant.providers.tidal.media.parse_track_v2")
551async def test_favorite_tracks_walk_feeds_churn_cache(
552    mock_parse_track: Mock, media_manager: TidalMediaManager, provider_mock: Mock
553) -> None:
554    """
555    Test the favourites walk hands each linkage item to note_replaced_track.
556
557    The read is made with replaceMedia, so Tidal already computed any stale->live
558    replacement pairs; discarding them would waste the largest healing source.
559    """
560    items = [{"type": "tracks", "id": "1"}, {"type": "tracks", "id": "2"}]
561    doc = JsonApiDocument(
562        {
563            "data": items,
564            "included": [
565                {"type": "tracks", "id": "1", "attributes": {}},
566                {"type": "tracks", "id": "2", "attributes": {}},
567            ],
568        }
569    )
570
571    async def _pages(*_a: Any, **_k: Any) -> Any:
572        yield doc
573
574    provider_mock.api.paginate_jsonapi = _pages
575    mock_parse_track.side_effect = [Mock(item_id="1"), Mock(item_id="2")]
576
577    await media_manager.get_playlist_tracks("favorite_tracks", page=0)
578
579    assert provider_mock.note_replaced_track.call_count == 2
580    provider_mock.note_replaced_track.assert_any_call(items[0])
581    provider_mock.note_replaced_track.assert_any_call(items[1])
582
583
584async def test_get_playlist_tracks_favorite_tracks_later_page(
585    media_manager: TidalMediaManager, provider_mock: Mock
586) -> None:
587    """Test get_playlist_tracks returns nothing for pages after the first, without a fetch."""
588    provider_mock.api.paginate_jsonapi = Mock()
589
590    tracks = await media_manager.get_playlist_tracks("favorite_tracks", page=1)
591
592    assert tracks == []
593    provider_mock.api.paginate_jsonapi.assert_not_called()
594