/
/
/
1"""Tests for Soundcloud stream URL selection and stream details."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6from unittest.mock import AsyncMock, patch
7
8import pytest
9from music_assistant_models.enums import MediaType, StreamType
10from music_assistant_models.errors import MediaNotFoundError
11
12if TYPE_CHECKING:
13 from music_assistant.providers.soundcloud import SoundcloudMusicProvider
14
15# transcoding endpoints as advertised by the Soundcloud API, and the CDN urls they resolve to
16PROGRESSIVE_API = "https://api-v2.soundcloud.com/media/soundcloud:tracks:1/abc/stream/progressive"
17HLS_API = "https://api-v2.soundcloud.com/media/soundcloud:tracks:1/abc/stream/hls"
18AAC_API = "https://api-v2.soundcloud.com/media/soundcloud:tracks:1/abc/stream/aac"
19PROGRESSIVE_CDN = "https://cf-media.sndcdn.com/gBClm0KpoWnf.128.mp3"
20HLS_CDN = "https://cf-hls-media.sndcdn.com/playlist/gBClm0KpoWnf.128.mp3/playlist.m3u8"
21
22NOW = 1700000000
23
24
25def _transcoding(preset: str, protocol: str, url: str) -> dict[str, Any]:
26 """Build a single transcoding entry."""
27 return {"preset": preset, "url": url, "format": {"protocol": protocol}}
28
29
30def _track_info(
31 transcodings: list[dict[str, Any]], *, track_authorization: str | None = "auth-token"
32) -> dict[str, Any]:
33 """Build a Soundcloud API track object carrying the given transcodings."""
34 track_info: dict[str, Any] = {"id": 1, "media": {"transcodings": transcodings}}
35 if track_authorization is not None:
36 track_info["track_authorization"] = track_authorization
37 return track_info
38
39
40def _resolver(responses: dict[str, dict[str, Any]]) -> AsyncMock:
41 """Return an API stub resolving transcoding urls to their CDN response."""
42
43 async def _get(url: str, **_kwargs: Any) -> dict[str, Any]:
44 for api_url, response in responses.items():
45 if url.startswith(api_url):
46 return response
47 # a transcoding that no longer exists answers with an empty body
48 return {}
49
50 return AsyncMock(side_effect=_get)
51
52
53async def test_stream_url_needs_track_authorization(provider: SoundcloudMusicProvider) -> None:
54 """Without a track authorization token no stream url can be built."""
55 provider._soundcloud.get = _resolver({PROGRESSIVE_API: {"url": PROGRESSIVE_CDN}})
56
57 result = await provider._get_stream_url(
58 _track_info(
59 [_transcoding("mp3_0_1", "progressive", PROGRESSIVE_API)], track_authorization=None
60 )
61 )
62
63 assert result is None
64 provider._soundcloud.get.assert_not_awaited()
65
66
67async def test_stream_url_prefers_progressive_over_hls(provider: SoundcloudMusicProvider) -> None:
68 """Progressive is picked over HLS, which only covers a limited content window."""
69 provider._soundcloud.get = _resolver(
70 {PROGRESSIVE_API: {"url": PROGRESSIVE_CDN}, HLS_API: {"url": HLS_CDN}}
71 )
72
73 # hls is listed first, as the Soundcloud API does
74 result = await provider._get_stream_url(
75 _track_info(
76 [
77 _transcoding("mp3_0_1", "hls", HLS_API),
78 _transcoding("mp3_0_1", "progressive", PROGRESSIVE_API),
79 ]
80 )
81 )
82
83 assert result == PROGRESSIVE_CDN
84 requested = provider._soundcloud.get.await_args_list[0].args[0]
85 assert requested.startswith(PROGRESSIVE_API)
86 assert "client_id=test-client-id" in requested
87 assert "track_authorization=auth-token" in requested
88
89
90async def test_stream_url_falls_back_to_hls(provider: SoundcloudMusicProvider) -> None:
91 """An mp3 HLS transcoding is used when the track has no progressive variant."""
92 provider._soundcloud.get = _resolver({HLS_API: {"url": HLS_CDN}})
93
94 result = await provider._get_stream_url(_track_info([_transcoding("mp3_0_1", "hls", HLS_API)]))
95
96 assert result == HLS_CDN
97
98
99async def test_stream_url_ignores_non_mp3_presets(provider: SoundcloudMusicProvider) -> None:
100 """Presets we cannot play are never requested."""
101 provider._soundcloud.get = _resolver({AAC_API: {"url": "https://example.com/aac"}})
102
103 result = await provider._get_stream_url(_track_info([_transcoding("aac_160k", "hls", AAC_API)]))
104
105 assert result is None
106 provider._soundcloud.get.assert_not_awaited()
107
108
109async def test_stream_url_none_when_transcoding_is_gone(provider: SoundcloudMusicProvider) -> None:
110 """A transcoding that no longer resolves yields no stream url."""
111 # empty response body, as returned for the phantom mp3 entries of DRM protected tracks
112 provider._soundcloud.get = _resolver({})
113
114 result = await provider._get_stream_url(
115 _track_info([_transcoding("mp3_1_0", "progressive", PROGRESSIVE_API)])
116 )
117
118 assert result is None
119
120
121async def _stream_details(provider: SoundcloudMusicProvider, cdn_url: str) -> Any:
122 """Return stream details for a track resolving to the given CDN url."""
123 provider._soundcloud.get_track_details.return_value = [
124 _track_info([_transcoding("mp3_0_1", "progressive", PROGRESSIVE_API)])
125 ]
126 provider._soundcloud.get = _resolver({PROGRESSIVE_API: {"url": cdn_url}})
127 with patch("music_assistant.providers.soundcloud.time.time", return_value=NOW):
128 return await provider.get_stream_details("1", MediaType.TRACK)
129
130
131async def test_stream_details_for_progressive_url(provider: SoundcloudMusicProvider) -> None:
132 """A progressive CDN url is streamed over HTTP and can be seeked."""
133 streamdetails = await _stream_details(provider, PROGRESSIVE_CDN)
134
135 assert streamdetails.item_id == "1"
136 assert streamdetails.provider == provider.instance_id
137 assert streamdetails.path == PROGRESSIVE_CDN
138 assert streamdetails.stream_type == StreamType.HTTP
139 assert streamdetails.can_seek
140 assert streamdetails.allow_seek
141
142
143async def test_stream_details_for_hls_url(provider: SoundcloudMusicProvider) -> None:
144 """An HLS CDN url is streamed as HLS."""
145 streamdetails = await _stream_details(provider, HLS_CDN)
146
147 assert streamdetails.stream_type == StreamType.HLS
148
149
150@pytest.mark.parametrize("param", ["Expires", "expire"])
151async def test_stream_details_expiry_from_url(
152 provider: SoundcloudMusicProvider, param: str
153) -> None:
154 """The expiry is taken from the CDN url so a seek never uses an expired url."""
155 streamdetails = await _stream_details(provider, f"{PROGRESSIVE_CDN}?{param}={NOW + 3600}")
156
157 # ten seconds of headroom are kept before the url actually expires
158 assert streamdetails.expiration == 3590
159
160
161async def test_stream_details_expiry_without_query(provider: SoundcloudMusicProvider) -> None:
162 """A url without expiry information falls back to a conservative default."""
163 streamdetails = await _stream_details(provider, PROGRESSIVE_CDN)
164
165 assert streamdetails.expiration == 30
166
167
168async def test_stream_details_expiry_clamped_for_stale_url(
169 provider: SoundcloudMusicProvider,
170) -> None:
171 """An already expired url never results in a negative expiry."""
172 streamdetails = await _stream_details(provider, f"{PROGRESSIVE_CDN}?Expires={NOW - 3600}")
173
174 assert streamdetails.expiration == 30
175
176
177async def test_stream_details_without_stream_url(provider: SoundcloudMusicProvider) -> None:
178 """A track without any usable transcoding reports that no stream url is available."""
179 provider._soundcloud.get_track_details.return_value = [_track_info([])]
180 provider._soundcloud.get = _resolver({})
181
182 with pytest.raises(MediaNotFoundError, match="No stream URL available"):
183 await provider.get_stream_details("1", MediaType.TRACK)
184