/
/
/
1"""Test Tidal Streaming Manager."""
2
3from collections.abc import Coroutine
4from sqlite3 import OperationalError
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock, Mock
7
8import pytest
9from music_assistant_models.enums import ContentType, ExternalID, StreamType
10from music_assistant_models.errors import MediaNotFoundError
11from music_assistant_models.media_items import AudioFormat, Track
12
13from music_assistant.providers.tidal.streaming import TidalStreamingManager
14
15
16@pytest.fixture
17def provider_mock() -> Mock:
18 """Return a mock provider."""
19 provider = Mock()
20 provider.domain = "tidal"
21 provider.instance_id = "tidal_instance"
22 provider.config.get_value.return_value = "HIGH"
23 provider.api = AsyncMock()
24 provider.api.OPEN_API_URL = "https://openapi.tidal.com/v2"
25
26 # Mock throttler bypass as async context manager using MagicMock
27 bypass_ctx = MagicMock()
28 bypass_ctx.__aenter__ = AsyncMock(return_value=None)
29 bypass_ctx.__aexit__ = AsyncMock(return_value=None)
30 provider.api.throttler = Mock()
31 provider.api.throttler.bypass = Mock(return_value=bypass_ctx)
32
33 provider.get_track = AsyncMock()
34
35 # Mock mass
36 provider.mass = Mock()
37 provider.mass.cache.get = AsyncMock(return_value=None)
38 provider.mass.cache.set = AsyncMock()
39 provider.mass.cache.delete = AsyncMock()
40 provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None)
41
42 return provider
43
44
45@pytest.fixture
46def streaming_manager(provider_mock: Mock) -> TidalStreamingManager:
47 """Return a TidalStreamingManager instance."""
48 return TidalStreamingManager(provider_mock)
49
50
51@pytest.fixture
52def mock_track() -> Mock:
53 """Return a mock track."""
54 track = Mock(spec=Track)
55 track.item_id = "123"
56 track.duration = 180
57 return track
58
59
60async def test_get_stream_details_lossless(
61 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
62) -> None:
63 """Test get_stream_details with LOSSLESS quality."""
64 provider_mock.get_track.return_value = mock_track
65 provider_mock.api.get.return_value = (
66 {
67 "manifestMimeType": "application/vnd.tidal.bts",
68 "urls": ["https://example.com/stream.flac"],
69 "audioQuality": "LOSSLESS",
70 "sampleRate": 44100,
71 "bitDepth": 16,
72 },
73 None,
74 )
75
76 stream_details = await streaming_manager.get_stream_details("123")
77
78 assert stream_details.item_id == "123"
79 assert stream_details.provider == "tidal_instance"
80 assert stream_details.audio_format.content_type == ContentType.FLAC
81 assert stream_details.audio_format.sample_rate == 44100
82 assert stream_details.audio_format.bit_depth == 16
83 assert stream_details.stream_type == StreamType.HTTP
84 assert stream_details.path == "https://example.com/stream.flac"
85 assert stream_details.can_seek is True
86
87 provider_mock.get_track.assert_called_with("123")
88 provider_mock.api.get.assert_called_with(
89 "tracks/123/playbackinfopostpaywall",
90 params={
91 "playbackmode": "STREAM",
92 "assetpresentation": "FULL",
93 "audioquality": "HIGH",
94 },
95 )
96
97
98async def test_get_stream_details_hires(
99 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
100) -> None:
101 """Test get_stream_details with HIRES_LOSSLESS quality."""
102 provider_mock.get_track.return_value = mock_track
103 provider_mock.api.get.return_value = {
104 "urls": ["https://example.com/stream.flac"],
105 "audioQuality": "HIRES_LOSSLESS",
106 "sampleRate": 96000,
107 "bitDepth": 24,
108 }
109
110 stream_details = await streaming_manager.get_stream_details("123")
111
112 assert stream_details.audio_format.content_type == ContentType.FLAC
113 assert stream_details.audio_format.sample_rate == 96000
114 assert stream_details.audio_format.bit_depth == 24
115
116
117async def test_get_stream_details_with_dash_manifest(
118 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
119) -> None:
120 """Test get_stream_details with DASH manifest."""
121 provider_mock.get_track.return_value = mock_track
122 provider_mock.api.get.return_value = {
123 "manifestMimeType": "application/dash+xml",
124 "manifest": "base64encodedmanifestdata",
125 "audioQuality": "HIGH",
126 "sampleRate": 44100,
127 "bitDepth": 16,
128 }
129
130 stream_details = await streaming_manager.get_stream_details("123")
131
132 assert isinstance(stream_details.path, str)
133 assert stream_details.path.startswith("data:application/dash+xml;base64,")
134 assert "base64encodedmanifestdata" in stream_details.path
135
136
137async def test_get_stream_details_with_codec(
138 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
139) -> None:
140 """Test get_stream_details with codec specified."""
141 provider_mock.get_track.return_value = mock_track
142 provider_mock.api.get.return_value = {
143 "urls": ["https://example.com/stream.aac"],
144 "audioQuality": "HIGH",
145 "codec": "AAC",
146 "sampleRate": 44100,
147 "bitDepth": 16,
148 }
149
150 stream_details = await streaming_manager.get_stream_details("123")
151
152 assert stream_details.audio_format.content_type == ContentType.AAC
153
154
155async def test_get_stream_details_defaults_to_mp4(
156 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
157) -> None:
158 """Test get_stream_details defaults to MP4 when no quality/codec."""
159 provider_mock.get_track.return_value = mock_track
160 provider_mock.api.get.return_value = {
161 "urls": ["https://example.com/stream.m4a"],
162 "sampleRate": 44100,
163 "bitDepth": 16,
164 }
165
166 stream_details = await streaming_manager.get_stream_details("123")
167
168 assert stream_details.audio_format.content_type == ContentType.MP4
169
170
171async def test_get_stream_details_no_urls_raises_error(
172 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
173) -> None:
174 """Test get_stream_details raises error when no URLs."""
175 provider_mock.get_track.return_value = mock_track
176 provider_mock.api.get.return_value = {
177 "audioQuality": "HIGH",
178 "sampleRate": 44100,
179 "bitDepth": 16,
180 }
181
182 with pytest.raises(MediaNotFoundError, match="No stream URL found"):
183 await streaming_manager.get_stream_details("123")
184
185
186async def test_get_stream_details_track_not_found_no_isrc(
187 streaming_manager: TidalStreamingManager, provider_mock: Mock
188) -> None:
189 """Test get_stream_details when track not found and no ISRC fallback."""
190 provider_mock.get_track.side_effect = MediaNotFoundError("Track not found")
191 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = None
192
193 with pytest.raises(MediaNotFoundError, match="Track 123 not found"):
194 await streaming_manager.get_stream_details("123")
195
196
197async def test_get_track_by_isrc_from_cache(
198 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
199) -> None:
200 """Test _get_track_by_isrc returns cached result."""
201 provider_mock.mass.cache.get.return_value = "cached_track_456"
202 provider_mock.get_track.return_value = mock_track
203
204 result = await streaming_manager._get_track_by_isrc("123")
205
206 assert result == mock_track
207 provider_mock.mass.cache.get.assert_called_with(
208 "123",
209 provider="tidal_instance",
210 category=2, # CACHE_CATEGORY_ISRC_MAP
211 )
212 provider_mock.get_track.assert_called_with("cached_track_456")
213
214
215async def test_get_track_by_isrc_cache_miss_lookup_success(
216 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
217) -> None:
218 """Test _get_track_by_isrc performs ISRC lookup on cache miss."""
219 # Cache miss
220 provider_mock.mass.cache.get.return_value = None
221
222 # Library item with ISRC
223 lib_track = Mock()
224 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
225 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
226
227 # API lookup
228 provider_mock.api.get.return_value = {"data": [{"id": 456}]}
229
230 # Final track fetch
231 provider_mock.get_track.return_value = mock_track
232
233 result = await streaming_manager._get_track_by_isrc("123")
234
235 assert result == mock_track
236
237 # Verify API call
238 provider_mock.api.get.assert_called_with(
239 "/tracks",
240 params={"filter[isrc]": "US1234567890"},
241 base_url=provider_mock.api.OPEN_API_URL,
242 )
243
244 # Verify cache set
245 provider_mock.mass.cache.set.assert_called_with(
246 key="123",
247 data="456",
248 provider="tidal_instance",
249 category=2, # CACHE_CATEGORY_ISRC_MAP
250 persistent=True,
251 expiration=86400 * 90,
252 )
253
254 # Verify final track fetch
255 provider_mock.get_track.assert_called_with("456")
256
257
258async def test_get_track_by_isrc_no_library_item(
259 streaming_manager: TidalStreamingManager, provider_mock: Mock
260) -> None:
261 """Test _get_track_by_isrc returns None when no library item."""
262 provider_mock.mass.cache.get.return_value = None
263 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = None
264
265 result = await streaming_manager._get_track_by_isrc("123")
266
267 assert result is None
268
269
270async def test_get_track_by_isrc_no_isrc_external_id(
271 streaming_manager: TidalStreamingManager, provider_mock: Mock
272) -> None:
273 """Test _get_track_by_isrc returns None when library item has no ISRC."""
274 provider_mock.mass.cache.get.return_value = None
275
276 lib_track = Mock()
277 lib_track.external_ids = [(ExternalID.BARCODE, "some-id")]
278 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
279
280 result = await streaming_manager._get_track_by_isrc("123")
281
282 assert result is None
283
284
285async def test_get_track_by_isrc_api_returns_empty(
286 streaming_manager: TidalStreamingManager, provider_mock: Mock
287) -> None:
288 """Test _get_track_by_isrc returns None when API returns no data."""
289 provider_mock.mass.cache.get.return_value = None
290
291 lib_track = Mock()
292 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
293 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
294
295 provider_mock.api.get.return_value = {"data": []}
296
297 result = await streaming_manager._get_track_by_isrc("123")
298
299 assert result is None
300
301
302async def test_get_track_by_isrc_cached_track_not_found(
303 streaming_manager: TidalStreamingManager, provider_mock: Mock
304) -> None:
305 """Test _get_track_by_isrc deletes cache when cached track not found."""
306 provider_mock.mass.cache.get.return_value = "cached_track_999"
307 provider_mock.get_track.side_effect = MediaNotFoundError("Track not found")
308
309 # Should continue with ISRC lookup
310 lib_track = Mock()
311 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
312 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
313
314 provider_mock.api.get.return_value = {"data": []}
315
316 result = await streaming_manager._get_track_by_isrc("123")
317
318 # Should delete invalid cache entry
319 provider_mock.mass.cache.delete.assert_called_with(
320 "123",
321 provider="tidal_instance",
322 category=2, # CACHE_CATEGORY_ISRC_MAP
323 )
324
325 assert result is None
326
327
328async def test_get_stream_details_with_isrc_fallback(
329 streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
330) -> None:
331 """Test get_stream_details uses ISRC fallback when direct lookup fails."""
332 # Direct lookup fails
333 provider_mock.get_track.side_effect = [
334 MediaNotFoundError("Track not found"), # First call
335 mock_track, # Second call from ISRC lookup
336 mock_track, # Third call for stream details
337 ]
338
339 # ISRC lookup succeeds
340 lib_track = Mock()
341 lib_track.external_ids = [(ExternalID.ISRC, "US1234567890")]
342 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
343
344 provider_mock.api.get.return_value = (
345 {"data": [{"id": 456}]}, # ISRC lookup response
346 None,
347 )
348
349 # Stream details
350 provider_mock.api.get.side_effect = [
351 ({"data": [{"id": 456}]}, None), # ISRC lookup
352 (
353 { # Stream details
354 "urls": ["https://example.com/stream.flac"],
355 "audioQuality": "LOSSLESS",
356 "sampleRate": 44100,
357 "bitDepth": 16,
358 },
359 None,
360 ),
361 ]
362
363 stream_details = await streaming_manager.get_stream_details("123")
364
365 assert stream_details.item_id == "123"
366 assert stream_details.path == "https://example.com/stream.flac"
367
368
369async def test_get_stream_details_schedules_background_mapping_update(
370 streaming_manager: TidalStreamingManager,
371 provider_mock: Mock,
372 mock_track: Mock,
373 monkeypatch: pytest.MonkeyPatch,
374) -> None:
375 """Ensure get_stream_details schedules the background mapping update task."""
376 provider_mock.get_track.return_value = mock_track
377 provider_mock.api.get.return_value = {
378 "urls": ["https://example.com/stream.flac"],
379 "audioQuality": "LOSSLESS",
380 "sampleRate": 44100,
381 "bitDepth": 16,
382 }
383
384 created: list[tuple[str, AudioFormat]] = []
385
386 async def _fake_worker(provider_track_id: str, resolved_audio_format: AudioFormat) -> None:
387 created.append((provider_track_id, resolved_audio_format))
388
389 # Patch the worker method so we can validate the coroutine is created with expected args
390 monkeypatch.setattr(
391 streaming_manager, "_async_update_provider_mapping_audio_format", _fake_worker
392 )
393
394 captured_coros: list[Coroutine[Any, Any, None]] = []
395
396 def _fake_create_task(coro: Coroutine[Any, Any, None]) -> None:
397 # Don't schedule; just capture the coroutine so the test can await it.
398 captured_coros.append(coro)
399
400 provider_mock.mass.create_task = _fake_create_task
401
402 stream_details = await streaming_manager.get_stream_details("123")
403
404 assert len(captured_coros) == 1
405
406 # Execute the captured coroutine (safe because we patched the worker)
407 await captured_coros[0]
408
409 assert created == [("123", stream_details.audio_format)]
410
411
412async def test_async_update_provider_mapping_audio_format_no_library_item(
413 streaming_manager: TidalStreamingManager, provider_mock: Mock
414) -> None:
415 """Ensure no update occurs when no library item is found."""
416 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = None
417 provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
418
419 await streaming_manager._async_update_provider_mapping_audio_format(
420 provider_track_id="123",
421 resolved_audio_format=AudioFormat(
422 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
423 ),
424 )
425
426 provider_mock.mass.music.tracks.update_provider_mapping.assert_not_called()
427
428
429async def test_async_update_provider_mapping_audio_format_no_mapping(
430 streaming_manager: TidalStreamingManager, provider_mock: Mock
431) -> None:
432 """Ensure no update occurs when no provider mapping is found."""
433 lib_track = Mock()
434 lib_track.item_id = 1
435 lib_track.provider_mappings = set()
436 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
437 provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
438
439 await streaming_manager._async_update_provider_mapping_audio_format(
440 provider_track_id="123",
441 resolved_audio_format=AudioFormat(
442 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
443 ),
444 )
445
446 provider_mock.mass.music.tracks.update_provider_mapping.assert_not_called()
447
448
449async def test_async_update_provider_mapping_audio_format_same_format_no_update(
450 streaming_manager: TidalStreamingManager, provider_mock: Mock
451) -> None:
452 """Ensure no update occurs when the audio format is unchanged."""
453 fmt = AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16)
454 mapping = Mock()
455 mapping.provider_instance = provider_mock.instance_id
456 mapping.item_id = "123"
457 mapping.audio_format = fmt
458
459 lib_track = Mock()
460 lib_track.item_id = 1
461 lib_track.provider_mappings = {mapping}
462 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
463 provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
464
465 await streaming_manager._async_update_provider_mapping_audio_format(
466 provider_track_id="123",
467 resolved_audio_format=fmt,
468 )
469
470 provider_mock.mass.music.tracks.update_provider_mapping.assert_not_called()
471
472
473async def test_async_update_provider_mapping_audio_format_different_format_updates(
474 streaming_manager: TidalStreamingManager, provider_mock: Mock
475) -> None:
476 """Ensure update occurs when the audio format is different."""
477 old_fmt = AudioFormat(content_type=ContentType.MP4, sample_rate=44100, bit_depth=16)
478 new_fmt = AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16)
479
480 mapping = Mock()
481 mapping.provider_instance = provider_mock.instance_id
482 mapping.item_id = "123"
483 mapping.audio_format = old_fmt
484
485 lib_track = Mock()
486 lib_track.item_id = 1
487 lib_track.provider_mappings = {mapping}
488 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
489 provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
490
491 await streaming_manager._async_update_provider_mapping_audio_format(
492 provider_track_id="123",
493 resolved_audio_format=new_fmt,
494 )
495
496 provider_mock.mass.music.tracks.update_provider_mapping.assert_awaited_once()
497 provider_mock.mass.music.tracks.update_provider_mapping.assert_awaited_with(
498 item_id=1,
499 provider_instance_id=provider_mock.instance_id,
500 provider_item_id="123",
501 audio_format=new_fmt,
502 )
503
504
505async def test_async_update_provider_mapping_audio_format_sqlite_operational_error_logs_debug(
506 streaming_manager: TidalStreamingManager, provider_mock: Mock
507) -> None:
508 """Ensure OperationalError is logged at debug level."""
509 provider_mock.logger = Mock()
510 provider_mock.mass.music.tracks.get_library_item_by_prov_id.side_effect = OperationalError(
511 "database is locked"
512 )
513
514 await streaming_manager._async_update_provider_mapping_audio_format(
515 provider_track_id="123",
516 resolved_audio_format=AudioFormat(
517 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
518 ),
519 )
520
521 provider_mock.logger.debug.assert_called()
522
523
524async def test_async_update_provider_mapping_audio_format_unexpected_error_logs_exception(
525 streaming_manager: TidalStreamingManager, provider_mock: Mock
526) -> None:
527 """Ensure unexpected errors are logged at exception level."""
528 provider_mock.logger = Mock()
529
530 lib_track = Mock()
531 lib_track.item_id = 1
532 lib_track.provider_mappings = set()
533 provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
534
535 # Force an unexpected error after resolving lib_track
536 provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock(
537 side_effect=RuntimeError("boom")
538 )
539
540 # Create a mapping that triggers the update path
541 mapping = Mock()
542 mapping.provider_instance = provider_mock.instance_id
543 mapping.item_id = "123"
544 mapping.audio_format = AudioFormat(
545 content_type=ContentType.MP4, sample_rate=44100, bit_depth=16
546 )
547 lib_track.provider_mappings = {mapping}
548
549 await streaming_manager._async_update_provider_mapping_audio_format(
550 provider_track_id="123",
551 resolved_audio_format=AudioFormat(
552 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
553 ),
554 )
555
556 provider_mock.logger.exception.assert_called()
557