/
/
/
1"""Tests for skipping DRM protected Soundcloud tracks."""
2
3from __future__ import annotations
4
5import logging
6from collections.abc import AsyncGenerator
7from typing import Any
8
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import MediaNotFoundError
12
13from music_assistant.models.music_provider import SyncRunState
14from music_assistant.providers.soundcloud import (
15 DrmProtectedTrackError,
16 SoundcloudMusicProvider,
17 _is_drm_protected,
18)
19
20# transcodings as returned by the Soundcloud API for DRM protected content: the encrypted
21# HLS variants are the only playable ones, while the mp3 entries are advertised but 404
22DRM_TRANSCODINGS = [
23 {
24 "preset": "aac_160k",
25 "url": "https://api/aac_cbc",
26 "format": {"protocol": "cbc-encrypted-hls"},
27 },
28 {
29 "preset": "aac_160k",
30 "url": "https://api/aac_ctr",
31 "format": {"protocol": "ctr-encrypted-hls"},
32 },
33 {"preset": "mp3_1_0", "url": "https://api/mp3_hls", "format": {"protocol": "hls"}},
34 {"preset": "mp3_1_0", "url": "https://api/mp3_prog", "format": {"protocol": "progressive"}},
35]
36PLAIN_TRANSCODINGS = [
37 {"preset": "mp3_0_1", "url": "https://api/mp3_hls", "format": {"protocol": "hls"}},
38 {"preset": "mp3_0_1", "url": "https://api/mp3_prog", "format": {"protocol": "progressive"}},
39]
40
41
42def _track_obj(track_id: int, title: str, transcodings: list[dict[str, Any]]) -> dict[str, Any]:
43 """Build a Soundcloud API track object with the given transcodings."""
44 return {
45 "id": track_id,
46 "title": title,
47 "duration": 235818,
48 "full_duration": 235771,
49 "kind": "track",
50 "permalink_url": f"https://soundcloud.com/artist/{track_id}",
51 "policy": "MONETIZE",
52 "monetization_model": "AD_SUPPORTED",
53 "track_authorization": "auth-token",
54 "user": {"id": 1, "username": "Some Artist", "permalink": "some-artist"},
55 "media": {"transcodings": transcodings},
56 }
57
58
59def test_is_drm_protected_detects_encrypted_transcodings() -> None:
60 """Encrypted HLS transcodings mark a track as DRM protected."""
61 assert _is_drm_protected(_track_obj(1, "Danceteria", DRM_TRANSCODINGS)) is True
62
63
64def test_is_drm_protected_allows_plain_transcodings() -> None:
65 """A track with only plain transcodings is not DRM protected."""
66 assert _is_drm_protected(_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)) is False
67
68
69def test_is_drm_protected_allows_partial_track_object() -> None:
70 """A partial track object without media details is not treated as DRM protected."""
71 assert _is_drm_protected({"id": 3, "kind": "track"}) is False
72
73
74async def test_parse_track_rejects_drm_track(provider: SoundcloudMusicProvider) -> None:
75 """Parsing a DRM protected track raises an error the listing paths already skip."""
76 with pytest.raises(DrmProtectedTrackError):
77 await provider._parse_track(_track_obj(1, "Danceteria", DRM_TRANSCODINGS))
78 assert issubclass(DrmProtectedTrackError, MediaNotFoundError)
79
80
81async def test_parse_track_accepts_plain_track(provider: SoundcloudMusicProvider) -> None:
82 """Parsing a track without DRM returns a Track."""
83 track = await provider._parse_track(_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS))
84 assert track.item_id == "2"
85 assert track.name == "Lofi Beat"
86
87
88async def test_library_tracks_skip_drm_and_log_count(
89 provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
90) -> None:
91 """DRM protected tracks are not imported and the number skipped is logged."""
92
93 async def _liked_tracks(_user_id: str) -> AsyncGenerator[dict[str, Any]]:
94 yield _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
95 yield _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)
96 yield _track_obj(3, "The Fate of Ophelia", DRM_TRANSCODINGS)
97
98 provider._soundcloud.get_track_details_liked = _liked_tracks
99
100 with caplog.at_level(logging.INFO):
101 tracks = [track async for track in provider.get_library_tracks()]
102
103 assert [track.item_id for track in tracks] == ["2"]
104 assert "2" in caplog.text
105 assert "DRM" in caplog.text
106
107
108async def test_library_drm_tracks_are_not_reported_as_sync_failures(
109 provider: SoundcloudMusicProvider, sync_run: SyncRunState
110) -> None:
111 """
112 A DRM protected track is left out without counting as a failed sync item.
113
114 Soundcloud never allows those to be imported, so they are a permanent and expected
115 category rather than something that went wrong, and the count is logged in one go.
116 """
117
118 async def _liked_tracks(_user_id: str) -> AsyncGenerator[dict[str, Any]]:
119 yield _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
120 yield _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)
121
122 provider._soundcloud.get_track_details_liked = _liked_tracks
123
124 tracks = [track async for track in provider.get_library_tracks()]
125
126 assert [track.item_id for track in tracks] == ["2"]
127 assert sync_run.skipped_item_ids == {}
128 assert sync_run.failures == 0
129 assert not sync_run.incomplete_media_types
130
131
132async def test_playlist_tracks_skip_drm(provider: SoundcloudMusicProvider) -> None:
133 """DRM protected tracks are left out of playlist listings."""
134 provider._soundcloud.get_playlist_details.return_value = {
135 "id": 10,
136 "tracks": [
137 _track_obj(1, "Danceteria", DRM_TRANSCODINGS),
138 _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS),
139 ],
140 }
141
142 get_playlist_tracks: Any = SoundcloudMusicProvider.get_playlist_tracks.__wrapped__ # type: ignore[attr-defined]
143 tracks = await get_playlist_tracks(provider, "10")
144
145 assert [track.item_id for track in tracks] == ["2"]
146
147
148async def test_search_skips_drm_track_and_keeps_others(
149 provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
150) -> None:
151 """A DRM protected search hit is skipped without dropping the other results."""
152 provider._soundcloud.search.return_value = {
153 "collection": [
154 _track_obj(1, "Danceteria", DRM_TRANSCODINGS),
155 _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS),
156 _track_obj(3, "The Fate of Ophelia", DRM_TRANSCODINGS),
157 ]
158 }
159
160 search: Any = SoundcloudMusicProvider.search.__wrapped__ # type: ignore[attr-defined]
161 with caplog.at_level(logging.DEBUG):
162 result = await search(provider, "madonna", [MediaType.TRACK], 10)
163
164 assert [track.item_id for track in result.tracks] == ["2"]
165 # two skipped tracks must produce one summary line saying 2, not one line each
166 drm_lines = [line for line in caplog.text.splitlines() if "DRM protected" in line]
167 assert len(drm_lines) == 1
168 assert "Skipped 2 DRM protected track(s) in Soundcloud search results" in drm_lines[0]
169
170
171async def test_subscribed_feed_skips_drm_track_and_keeps_others(
172 provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
173) -> None:
174 """A DRM protected track in the subscribed feed is skipped without emptying it."""
175 provider._soundcloud.get_subscribe_feed.return_value = {
176 "collection": [
177 {"type": "track", "track": _track_obj(1, "Danceteria", DRM_TRANSCODINGS)},
178 {"type": "track", "track": _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)},
179 ]
180 }
181
182 get_feed_tracks: Any = (
183 SoundcloudMusicProvider._get_subscribed_feed_tracks.__wrapped__ # type: ignore[attr-defined]
184 )
185 with caplog.at_level(logging.DEBUG):
186 tracks = await get_feed_tracks(provider)
187
188 assert [track.item_id for track in tracks] == ["2"]
189 assert "Skipped 1 DRM protected track(s) in the Soundcloud subscribed feed" in caplog.text
190
191
192async def test_artist_toptracks_skips_drm_track_and_keeps_others(
193 provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
194) -> None:
195 """A DRM protected track in an artist's top tracks is skipped without dropping others."""
196 provider._soundcloud.get_tracks_from_user.return_value = {"collection": [{"id": 1}, {"id": 2}]}
197 provider._soundcloud.get_track_details.side_effect = [
198 [_track_obj(1, "Danceteria", DRM_TRANSCODINGS)],
199 [_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)],
200 ]
201
202 get_toptracks: Any = (
203 SoundcloudMusicProvider.get_artist_toptracks.__wrapped__ # type: ignore[attr-defined]
204 )
205 with caplog.at_level(logging.DEBUG):
206 tracks = await get_toptracks(provider, "99")
207
208 assert [track.item_id for track in tracks] == ["2"]
209 assert (
210 "Skipped 1 DRM protected track(s) in the top tracks of Soundcloud artist 99" in caplog.text
211 )
212
213
214async def test_similar_tracks_skips_drm_track_and_keeps_others(
215 provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
216) -> None:
217 """A DRM protected similar track is skipped without dropping the other results."""
218 provider._soundcloud.get_recommended.return_value = {"collection": [{"id": 1}, {"id": 2}]}
219 provider._soundcloud.get_track_details.side_effect = [
220 [_track_obj(1, "Danceteria", DRM_TRANSCODINGS)],
221 [_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)],
222 ]
223
224 get_similar: Any = SoundcloudMusicProvider.get_similar_tracks.__wrapped__ # type: ignore[attr-defined]
225 with caplog.at_level(logging.DEBUG):
226 tracks = await get_similar(provider, "5")
227
228 assert [track.item_id for track in tracks] == ["2"]
229 assert "Skipped 1 DRM protected track(s) in tracks similar to Soundcloud track 5" in caplog.text
230
231
232async def test_get_stream_details_reports_drm(provider: SoundcloudMusicProvider) -> None:
233 """Streaming an already imported DRM track fails with a message naming the cause."""
234 provider._soundcloud.get_track_details.return_value = [
235 _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
236 ]
237
238 with pytest.raises(MediaNotFoundError, match="DRM"):
239 await provider.get_stream_details("1", MediaType.TRACK)
240
241
242async def test_get_track_raises_media_not_found_when_drm_protected(
243 provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
244) -> None:
245 """A DRM protected track raises MediaNotFoundError naming the cause, without logging."""
246 provider._soundcloud.get_track_details.return_value = [
247 _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
248 ]
249
250 get_track: Any = SoundcloudMusicProvider.get_track.__wrapped__ # type: ignore[attr-defined]
251 with caplog.at_level(logging.DEBUG), pytest.raises(MediaNotFoundError, match="DRM"):
252 await get_track(provider, "1")
253
254 assert caplog.text == ""
255