/
/
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 InvalidDataError, MediaNotFoundError
12
13from music_assistant.providers.soundcloud import (
14 DrmProtectedTrackError,
15 SoundcloudMusicProvider,
16 _is_drm_protected,
17)
18
19# transcodings as returned by the Soundcloud API for DRM protected content: the encrypted
20# HLS variants are the only playable ones, while the mp3 entries are advertised but 404
21DRM_TRANSCODINGS = [
22 {
23 "preset": "aac_160k",
24 "url": "https://api/aac_cbc",
25 "format": {"protocol": "cbc-encrypted-hls"},
26 },
27 {
28 "preset": "aac_160k",
29 "url": "https://api/aac_ctr",
30 "format": {"protocol": "ctr-encrypted-hls"},
31 },
32 {"preset": "mp3_1_0", "url": "https://api/mp3_hls", "format": {"protocol": "hls"}},
33 {"preset": "mp3_1_0", "url": "https://api/mp3_prog", "format": {"protocol": "progressive"}},
34]
35PLAIN_TRANSCODINGS = [
36 {"preset": "mp3_0_1", "url": "https://api/mp3_hls", "format": {"protocol": "hls"}},
37 {"preset": "mp3_0_1", "url": "https://api/mp3_prog", "format": {"protocol": "progressive"}},
38]
39
40
41def _track_obj(track_id: int, title: str, transcodings: list[dict[str, Any]]) -> dict[str, Any]:
42 """Build a Soundcloud API track object with the given transcodings."""
43 return {
44 "id": track_id,
45 "title": title,
46 "duration": 235818,
47 "full_duration": 235771,
48 "kind": "track",
49 "permalink_url": f"https://soundcloud.com/artist/{track_id}",
50 "policy": "MONETIZE",
51 "monetization_model": "AD_SUPPORTED",
52 "track_authorization": "auth-token",
53 "user": {"id": 1, "username": "Some Artist", "permalink": "some-artist"},
54 "media": {"transcodings": transcodings},
55 }
56
57
58def test_is_drm_protected_detects_encrypted_transcodings() -> None:
59 """Encrypted HLS transcodings mark a track as DRM protected."""
60 assert _is_drm_protected(_track_obj(1, "Danceteria", DRM_TRANSCODINGS)) is True
61
62
63def test_is_drm_protected_allows_plain_transcodings() -> None:
64 """A track with only plain transcodings is not DRM protected."""
65 assert _is_drm_protected(_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)) is False
66
67
68def test_is_drm_protected_allows_partial_track_object() -> None:
69 """A partial track object without media details is not treated as DRM protected."""
70 assert _is_drm_protected({"id": 3, "kind": "track"}) is False
71
72
73async def test_parse_track_rejects_drm_track(provider: SoundcloudMusicProvider) -> None:
74 """Parsing a DRM protected track raises an error the listing paths already skip."""
75 with pytest.raises(DrmProtectedTrackError):
76 await provider._parse_track(_track_obj(1, "Danceteria", DRM_TRANSCODINGS))
77 assert issubclass(DrmProtectedTrackError, InvalidDataError)
78
79
80async def test_parse_track_accepts_plain_track(provider: SoundcloudMusicProvider) -> None:
81 """Parsing a track without DRM returns a Track."""
82 track = await provider._parse_track(_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS))
83 assert track.item_id == "2"
84 assert track.name == "Lofi Beat"
85
86
87async def test_library_tracks_skip_drm_and_log_count(
88 provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
89) -> None:
90 """DRM protected tracks are not imported and the number skipped is logged."""
91
92 async def _liked_tracks(_user_id: str) -> AsyncGenerator[dict[str, Any]]:
93 yield _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
94 yield _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)
95 yield _track_obj(3, "The Fate of Ophelia", DRM_TRANSCODINGS)
96
97 provider._soundcloud.get_track_details_liked = _liked_tracks
98
99 with caplog.at_level(logging.INFO):
100 tracks = [track async for track in provider.get_library_tracks()]
101
102 assert [track.item_id for track in tracks] == ["2"]
103 assert "2" in caplog.text
104 assert "DRM" in caplog.text
105
106
107async def test_playlist_tracks_skip_drm(provider: SoundcloudMusicProvider) -> None:
108 """DRM protected tracks are left out of playlist listings."""
109 provider._soundcloud.get_playlist_details.return_value = {
110 "id": 10,
111 "tracks": [
112 _track_obj(1, "Danceteria", DRM_TRANSCODINGS),
113 _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS),
114 ],
115 }
116
117 get_playlist_tracks: Any = SoundcloudMusicProvider.get_playlist_tracks.__wrapped__ # type: ignore[attr-defined]
118 tracks = await get_playlist_tracks(provider, "10")
119
120 assert [track.item_id for track in tracks] == ["2"]
121
122
123async def test_search_skips_drm_track_and_keeps_others(
124 provider: SoundcloudMusicProvider,
125) -> None:
126 """A DRM protected search hit is skipped without dropping the other results."""
127 provider._soundcloud.search.return_value = {
128 "collection": [
129 _track_obj(1, "Danceteria", DRM_TRANSCODINGS),
130 _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS),
131 ]
132 }
133
134 search: Any = SoundcloudMusicProvider.search.__wrapped__ # type: ignore[attr-defined]
135 result = await search(provider, "madonna", [MediaType.TRACK], 10)
136
137 assert [track.item_id for track in result.tracks] == ["2"]
138
139
140async def test_get_stream_details_reports_drm(provider: SoundcloudMusicProvider) -> None:
141 """Streaming an already imported DRM track fails with a message naming the cause."""
142 provider._soundcloud.get_track_details.return_value = [
143 _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
144 ]
145
146 with pytest.raises(MediaNotFoundError, match="DRM"):
147 await provider.get_stream_details("1", MediaType.TRACK)
148
149
150async def test_get_track_raises_media_not_found_when_unparsable(
151 provider: SoundcloudMusicProvider,
152) -> None:
153 """An unparsable track results in MediaNotFoundError instead of an internal error."""
154 provider._soundcloud.get_track_details.return_value = [
155 _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
156 ]
157
158 get_track: Any = SoundcloudMusicProvider.get_track.__wrapped__ # type: ignore[attr-defined]
159 with pytest.raises(MediaNotFoundError):
160 await get_track(provider, "1")
161