music-assistant-server
12.3 KB•PY
test_streaming.py
12.3 KB • 364 lines • python
1"""Test Tidal Streaming Manager."""
2
3from unittest.mock import AsyncMock, MagicMock, Mock
4
5import pytest
6from music_assistant_models.enums import ContentType, ExternalID, StreamType
7from music_assistant_models.errors import MediaNotFoundError
8from music_assistant_models.media_items import Track
9
10from music_assistant.providers.tidal.streaming import TidalStreamingManager
11
12
13@pytest.fixture
14def provider_mock() -> Mock:
15 """Return a mock provider."""
16 provider = Mock()
17 provider.domain = "tidal"
18 provider.instance_id = "tidal_instance"
19 provider.config.get_value.return_value = "HIGH"
20 provider.api = AsyncMock()
21 provider.api.OPEN_API_URL = "https://openapi.tidal.com/v2"
22
23 # Mock throttler bypass as async context manager using MagicMock
24 bypass_ctx = MagicMock()
25 bypass_ctx.__aenter__ = AsyncMock(return_value=None)
26 bypass_ctx.__aexit__ = AsyncMock(return_value=None)
27 provider.api.throttler = Mock()
28 provider.api.throttler.bypass = Mock(return_value=bypass_ctx)
29
30 provider.get_track = AsyncMock()
31
32 # Mock mass
33 provider.mass = Mock()
34 provider.mass.cache.get = AsyncMock(return_value=None)
35 provider.mass.cache.set = AsyncMock()
36 provider.mass.cache.delete = AsyncMock()
37 provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None)
38
39 return provider
40
41
42@pytest.fixture
43def streaming_manager(provider_mock: Mock) -> TidalStreamingManager:
44 """Return a TidalStreamingManager instance."""
45 return TidalStreamingManager(provider_mock)
46
47
48@pytest.fixture
49def mock_track() -> Mock:
50 """Return a mock track."""
51 track = Mock(spec=Track)
52 track.item_id = "123"
53 track.duration = 180
54 return track
55
56
57async def test_get_stream_details_lossless(
58 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
59) -> None:
60 """Test get_stream_details with LOSSLESS quality."""
61 provider_mock.get_track.return_value = mock_track
62 provider_mock.api.get.return_value = (
63 {
64 "manifestMimeType": "application/vnd.tidal.bts",
65 "urls": ["https://example.com/stream.flac"],
66 "audioQuality": "LOSSLESS",
67 "sampleRate": 44100,
68 "bitDepth": 16,
69 },
70 None,
71 )
72
73 stream_details = await streaming_manager.get_stream_details("123")
74
75 assert stream_details.item_id == "123"
76 assert stream_details.provider == "tidal_instance"
77 assert stream_details.audio_format.content_type == ContentType.FLAC
78 assert stream_details.audio_format.sample_rate == 44100
79 assert stream_details.audio_format.bit_depth == 16
80 assert stream_details.stream_type == StreamType.HTTP
81 assert stream_details.path == "https://example.com/stream.flac"
82 assert stream_details.can_seek is True
83
84 provider_mock.get_track.assert_called_with("123")
85 provider_mock.api.get.assert_called_with(
86 "tracks/123/playbackinfopostpaywall",
87 params={
88 "playbackmode": "STREAM",
89 "assetpresentation": "FULL",
90 "audioquality": "HIGH",
91 },
92 )
93
94
95async def test_get_stream_details_hires(
96 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
97) -> None:
98 """Test get_stream_details with HIRES_LOSSLESS quality."""
99 provider_mock.get_track.return_value = mock_track
100 provider_mock.api.get.return_value = {
101 "urls": ["https://example.com/stream.flac"],
102 "audioQuality": "HIRES_LOSSLESS",
103 "sampleRate": 96000,
104 "bitDepth": 24,
105 }
106
107 stream_details = await streaming_manager.get_stream_details("123")
108
109 assert stream_details.audio_format.content_type == ContentType.FLAC
110 assert stream_details.audio_format.sample_rate == 96000
111 assert stream_details.audio_format.bit_depth == 24
112
113
114async def test_get_stream_details_with_dash_manifest(
115 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
116) -> None:
117 """Test get_stream_details with DASH manifest."""
118 provider_mock.get_track.return_value = mock_track
119 provider_mock.api.get.return_value = {
120 "manifestMimeType": "application/dash+xml",
121 "manifest": "base64encodedmanifestdata",
122 "audioQuality": "HIGH",
123 "sampleRate": 44100,
124 "bitDepth": 16,
125 }
126
127 stream_details = await streaming_manager.get_stream_details("123")
128
129 assert isinstance(stream_details.path, str)
130 assert stream_details.path.startswith("data:application/dash+xml;base64,")
131 assert "base64encodedmanifestdata" in stream_details.path
132
133
134async def test_get_stream_details_with_codec(
135 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
136) -> None:
137 """Test get_stream_details with codec specified."""
138 provider_mock.get_track.return_value = mock_track
139 provider_mock.api.get.return_value = {
140 "urls": ["https://example.com/stream.aac"],
141 "audioQuality": "HIGH",
142 "codec": "AAC",
143 "sampleRate": 44100,
144 "bitDepth": 16,
145 }
146
147 stream_details = await streaming_manager.get_stream_details("123")
148
149 assert stream_details.audio_format.content_type == ContentType.AAC
150
151
152async def test_get_stream_details_defaults_to_mp4(
153 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
154) -> None:
155 """Test get_stream_details defaults to MP4 when no quality/codec."""
156 provider_mock.get_track.return_value = mock_track
157 provider_mock.api.get.return_value = {
158 "urls": ["https://example.com/stream.m4a"],
159 "sampleRate": 44100,
160 "bitDepth": 16,
161 }
162
163 stream_details = await streaming_manager.get_stream_details("123")
164
165 assert stream_details.audio_format.content_type == ContentType.MP4
166
167
168async def test_get_stream_details_no_urls_raises_error(
169 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
170) -> None:
171 """Test get_stream_details raises error when no URLs."""
172 provider_mock.get_track.return_value = mock_track
173 provider_mock.api.get.return_value = {
174 "audioQuality": "HIGH",
175 "sampleRate": 44100,
176 "bitDepth": 16,
177 }
178
179 with pytest.raises(MediaNotFoundError, match="No stream URL found"):
180 await streaming_manager.get_stream_details("123")
181
182
183async def test_get_stream_details_track_not_found_no_isrc(
184 streaming_manager: TidalStreamingManager, provider_mock: Mock
185) -> None:
186 """Test get_stream_details when track not found and no ISRC fallback."""
187 provider_mock.get_track.side_effect = MediaNotFoundError("Track not found")
188 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = None
189
190 with pytest.raises(MediaNotFoundError, match="Track 123 not found"):
191 await streaming_manager.get_stream_details("123")
192
193
194async def test_get_track_by_isrc_from_cache(
195 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
196) -> None:
197 """Test _get_track_by_isrc returns cached result."""
198 provider_mock.mass.cache.get.return_value = "cached_track_456"
199 provider_mock.get_track.return_value = mock_track
200
201 result = await streaming_manager._get_track_by_isrc("123")
202
203 assert result == mock_track
204 provider_mock.mass.cache.get.assert_called_with(
205 "123",
206 provider="tidal_instance",
207 category=2, # CACHE_CATEGORY_ISRC_MAP
208 )
209 provider_mock.get_track.assert_called_with("cached_track_456")
210
211
212async def test_get_track_by_isrc_cache_miss_lookup_success(
213 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
214) -> None:
215 """Test _get_track_by_isrc performs ISRC lookup on cache miss."""
216 # Cache miss
217 provider_mock.mass.cache.get.return_value = None
218
219 # Library item with ISRC
220 lib_track = Mock()
221 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
222 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
223
224 # API lookup
225 provider_mock.api.get.return_value = {"data": [{"id": 456}]}
226
227 # Final track fetch
228 provider_mock.get_track.return_value = mock_track
229
230 result = await streaming_manager._get_track_by_isrc("123")
231
232 assert result == mock_track
233
234 # Verify API call
235 provider_mock.api.get.assert_called_with(
236 "/tracks",
237 params={"filter[isrc]": "US1234567890"},
238 base_url=provider_mock.api.OPEN_API_URL,
239 )
240
241 # Verify cache set
242 provider_mock.mass.cache.set.assert_called_with(
243 key="123",
244 data="456",
245 provider="tidal_instance",
246 category=2, # CACHE_CATEGORY_ISRC_MAP
247 persistent=True,
248 expiration=86400 * 90,
249 )
250
251 # Verify final track fetch
252 provider_mock.get_track.assert_called_with("456")
253
254
255async def test_get_track_by_isrc_no_library_item(
256 streaming_manager: TidalStreamingManager, provider_mock: Mock
257) -> None:
258 """Test _get_track_by_isrc returns None when no library item."""
259 provider_mock.mass.cache.get.return_value = None
260 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = None
261
262 result = await streaming_manager._get_track_by_isrc("123")
263
264 assert result is None
265
266
267async def test_get_track_by_isrc_no_isrc_external_id(
268 streaming_manager: TidalStreamingManager, provider_mock: Mock
269) -> None:
270 """Test _get_track_by_isrc returns None when library item has no ISRC."""
271 provider_mock.mass.cache.get.return_value = None
272
273 lib_track = Mock()
274 lib_track.external_ids = [(ExternalID.BARCODE, "some-id")]
275 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
276
277 result = await streaming_manager._get_track_by_isrc("123")
278
279 assert result is None
280
281
282async def test_get_track_by_isrc_api_returns_empty(
283 streaming_manager: TidalStreamingManager, provider_mock: Mock
284) -> None:
285 """Test _get_track_by_isrc returns None when API returns no data."""
286 provider_mock.mass.cache.get.return_value = None
287
288 lib_track = Mock()
289 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
290 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
291
292 provider_mock.api.get.return_value = {"data": []}
293
294 result = await streaming_manager._get_track_by_isrc("123")
295
296 assert result is None
297
298
299async def test_get_track_by_isrc_cached_track_not_found(
300 streaming_manager: TidalStreamingManager, provider_mock: Mock
301) -> None:
302 """Test _get_track_by_isrc deletes cache when cached track not found."""
303 provider_mock.mass.cache.get.return_value = "cached_track_999"
304 provider_mock.get_track.side_effect = MediaNotFoundError("Track not found")
305
306 # Should continue with ISRC lookup
307 lib_track = Mock()
308 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
309 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
310
311 provider_mock.api.get.return_value = {"data": []}
312
313 result = await streaming_manager._get_track_by_isrc("123")
314
315 # Should delete invalid cache entry
316 provider_mock.mass.cache.delete.assert_called_with(
317 "123",
318 provider="tidal_instance",
319 category=2, # CACHE_CATEGORY_ISRC_MAP
320 )
321
322 assert result is None
323
324
325async def test_get_stream_details_with_isrc_fallback(
326 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
327) -> None:
328 """Test get_stream_details uses ISRC fallback when direct lookup fails."""
329 # Direct lookup fails
330 provider_mock.get_track.side_effect = [
331 MediaNotFoundError("Track not found"), # First call
332 mock_track, # Second call from ISRC lookup
333 mock_track, # Third call for stream details
334 ]
335
336 # ISRC lookup succeeds
337 lib_track = Mock()
338 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
339 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
340
341 provider_mock.api.get.return_value = (
342 {"data": [{"id": 456}]}, # ISRC lookup response
343 None,
344 )
345
346 # Stream details
347 provider_mock.api.get.side_effect = [
348 ({"data": [{"id": 456}]}, None), # ISRC lookup
349 (
350 { # Stream details
351 "urls": ["https://example.com/stream.flac"],
352 "audioQuality": "LOSSLESS",
353 "sampleRate": 44100,
354 "bitDepth": 16,
355 },
356 None,
357 ),
358 ]
359
360 stream_details = await streaming_manager.get_stream_details("123")
361
362 assert stream_details.item_id == "123"
363 assert stream_details.path == "https://example.com/stream.flac"
364