/
/
/
1"""Tests for Soundcloud library listings and recommendations."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from typing import Any
7
8from music_assistant_models.enums import MediaType
9
10from music_assistant.models.music_provider import SyncRunState
11from music_assistant.providers.soundcloud import SoundcloudMusicProvider
12
13
14def _artist_obj(artist_id: int | None, username: str = "Some Artist") -> dict[str, Any]:
15 """Build a Soundcloud API user object."""
16 return {"id": artist_id, "username": username, "permalink": "some-artist"}
17
18
19def _track_obj(track_id: int, title: str = "Some Track") -> dict[str, Any]:
20 """Build a playable Soundcloud API track object."""
21 return {
22 "id": track_id,
23 "title": title,
24 "duration": 235818,
25 "permalink_url": f"https://soundcloud.com/artist/{track_id}",
26 "user": {"id": 1, "username": "Some Artist", "permalink": "some-artist"},
27 }
28
29
30async def test_library_artists_skip_unparsable(
31 provider: SoundcloudMusicProvider, sync_run: SyncRunState
32) -> None:
33 """An artist without a valid id is skipped, the rest is returned."""
34 provider._soundcloud.get_following.return_value = {
35 "collection": [_artist_obj(None), _artist_obj(7, "Real Artist")]
36 }
37
38 artists = [artist async for artist in provider.get_library_artists()]
39
40 provider._soundcloud.get_following.assert_awaited_once_with("1")
41 assert [artist.item_id for artist in artists] == ["7"]
42 assert artists[0].name == "Real Artist"
43 # the skipped artist has no id to report, so the run can not be a basis for deletions
44 assert sync_run.skipped_item_ids == {}
45 assert MediaType.ARTIST in sync_run.incomplete_media_types
46
47
48async def test_skipped_artist_is_reported_with_its_item_id(
49 provider: SoundcloudMusicProvider, sync_run: SyncRunState
50) -> None:
51 """
52 A skipped artist is reported under the id its provider mapping carries.
53
54 That id is what the deletion pass resolves the item by, so an artist Soundcloud still
55 follows is left alone instead of being removed from the library.
56 """
57 # an artist that carries an id, but no username to build the item from
58 provider._soundcloud.get_following.return_value = {
59 "collection": [{"id": 7, "permalink": "some-artist"}]
60 }
61
62 assert [artist async for artist in provider.get_library_artists()] == []
63
64 assert sync_run.skipped_item_ids == {MediaType.ARTIST: {"7"}}
65 assert not sync_run.incomplete_media_types
66
67
68async def test_library_playlists(provider: SoundcloudMusicProvider, sync_run: SyncRunState) -> None:
69 """Playlists are resolved to their full object before being parsed."""
70
71 async def _account_playlists() -> AsyncGenerator[dict[str, Any]]:
72 yield {"playlist": {"id": 10}}
73 yield {"not_a_playlist": {}} # unexpected shape, must be skipped
74
75 provider._soundcloud.get_account_playlists = _account_playlists
76 provider._soundcloud.get_playlist_details.return_value = {"id": 10, "title": "Summer Mix"}
77
78 playlists = [playlist async for playlist in provider.get_library_playlists()]
79
80 assert [playlist.item_id for playlist in playlists] == ["10"]
81 assert playlists[0].name == "Summer Mix"
82 # the unexpected entry holds no id at all, so the deletion pass is held back
83 assert MediaType.PLAYLIST in sync_run.incomplete_media_types
84
85
86async def test_library_playlists_skip_failing_details(
87 provider: SoundcloudMusicProvider, sync_run: SyncRunState
88) -> None:
89 """A playlist whose details cannot be parsed is skipped and reported under its id."""
90
91 async def _account_playlists() -> AsyncGenerator[dict[str, Any]]:
92 yield {"playlist": {"id": 10}}
93
94 provider._soundcloud.get_account_playlists = _account_playlists
95 provider._soundcloud.get_playlist_details.return_value = {"id": 10} # no title
96
97 playlists = [playlist async for playlist in provider.get_library_playlists()]
98
99 assert playlists == []
100 assert sync_run.skipped_item_ids == {MediaType.PLAYLIST: {"10"}}
101 assert not sync_run.incomplete_media_types
102
103
104async def test_library_tracks_skip_unparsable(
105 provider: SoundcloudMusicProvider, sync_run: SyncRunState
106) -> None:
107 """A track that cannot be parsed is skipped without failing the sync."""
108
109 async def _liked_tracks(_user_id: str) -> AsyncGenerator[dict[str, Any]]:
110 yield {"id": 1} # missing title/duration
111 yield _track_obj(2)
112
113 provider._soundcloud.get_track_details_liked = _liked_tracks
114
115 tracks = [track async for track in provider.get_library_tracks()]
116
117 assert [track.item_id for track in tracks] == ["2"]
118 assert sync_run.skipped_item_ids == {MediaType.TRACK: {"1"}}
119 assert not sync_run.incomplete_media_types
120
121
122async def test_recommendation_payload(provider: SoundcloudMusicProvider) -> None:
123 """The payload holds the mixed selections, keeping only system playlists."""
124 provider._soundcloud.get_mixed_selection.return_value = {
125 "collection": [
126 {
127 "id": "mixed-1",
128 "title": "Made for you",
129 "items": {
130 "collection": [
131 {"id": 10, "title": "Summer Mix", "kind": "system-playlist"},
132 {"id": 11, "title": "Not a playlist", "kind": "track"},
133 ]
134 },
135 }
136 ]
137 }
138
139 folders = await provider._fetch_recommendation_payload()
140
141 assert [folder.name for folder in folders] == ["Made for you"]
142 assert [item.item_id for item in folders[0].items] == ["10"]
143 # the feed row is filled on demand, so building the payload must not fetch it
144 provider._soundcloud.get_subscribe_feed.assert_not_awaited()
145
146
147async def test_recommendation_payload_with_empty_selection(
148 provider: SoundcloudMusicProvider,
149) -> None:
150 """A mixed selection without items yields an empty folder instead of failing."""
151 provider._soundcloud.get_mixed_selection.return_value = {
152 "collection": [{"id": "mixed-1", "title": "Made for you"}]
153 }
154
155 folders = await provider._fetch_recommendation_payload()
156
157 assert [folder.name for folder in folders] == ["Made for you"]
158 assert folders[0].items == []
159
160
161async def test_subscribed_feed_tracks(provider: SoundcloudMusicProvider) -> None:
162 """Reposts count as tracks in the feed, other item types are ignored."""
163 provider._soundcloud.get_subscribe_feed.return_value = {
164 "collection": [
165 {"type": "track", "track": _track_obj(1)},
166 {"type": "track-repost", "track": _track_obj(2)},
167 {"type": "playlist", "playlist": {"id": 99}},
168 ]
169 }
170
171 feed_tracks: Any = SoundcloudMusicProvider._get_subscribed_feed_tracks.__wrapped__ # type: ignore[attr-defined]
172 tracks = await feed_tracks(provider)
173
174 assert [track.item_id for track in tracks] == ["1", "2"]
175
176
177async def test_subscribed_feed_skips_unparsable_track(provider: SoundcloudMusicProvider) -> None:
178 """An unplayable track in the feed does not discard the rest of the row."""
179 provider._soundcloud.get_subscribe_feed.return_value = {
180 "collection": [
181 {"type": "track", "track": {"id": 1}}, # missing title/duration
182 {"type": "track", "track": _track_obj(2)},
183 ]
184 }
185
186 feed_tracks: Any = SoundcloudMusicProvider._get_subscribed_feed_tracks.__wrapped__ # type: ignore[attr-defined]
187 tracks = await feed_tracks(provider)
188
189 assert [track.item_id for track in tracks] == ["2"]
190
191
192async def test_subscribed_feed_without_results(provider: SoundcloudMusicProvider) -> None:
193 """An empty subscribed feed yields no tracks."""
194 provider._soundcloud.get_subscribe_feed.return_value = {}
195
196 feed_tracks: Any = SoundcloudMusicProvider._get_subscribed_feed_tracks.__wrapped__ # type: ignore[attr-defined]
197 tracks = await feed_tracks(provider)
198
199 assert tracks == []
200