music-assistant-server

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, 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(provider_mock: Mock) -> Mock:
18    """Return the shared provider mock with the streaming quality and throttler bypass wired."""
19    provider_mock.config.get_value.return_value = "HIGH"
20
21    # the streaming manager enters api.throttler.bypass() as an async context manager,
22    # which a MagicMock supports out of the box
23    provider_mock.api.throttler.bypass = Mock(return_value=MagicMock())
24
25    return provider_mock
26
27
28@pytest.fixture
29def streaming_manager(provider_mock: Mock) -> TidalStreamingManager:
30    """Return a TidalStreamingManager instance."""
31    return TidalStreamingManager(provider_mock)
32
33
34@pytest.fixture
35def mock_track() -> Mock:
36    """Return a mock track."""
37    track = Mock(spec=Track)
38    track.item_id = "123"
39    track.duration = 180
40    return track
41
42
43async def test_get_stream_details_lossless(
44    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
45) -> None:
46    """Test get_stream_details with LOSSLESS quality."""
47    provider_mock.get_track.return_value = mock_track
48    provider_mock.api.get.return_value = {
49        "manifestMimeType": "application/vnd.tidal.bts",
50        "urls": ["https://example.com/stream.flac"],
51        "audioQuality": "LOSSLESS",
52        "sampleRate": 44100,
53        "bitDepth": 16,
54    }
55
56    stream_details = await streaming_manager.get_stream_details("123")
57
58    assert stream_details.item_id == "123"
59    assert stream_details.provider == "tidal_instance"
60    assert stream_details.audio_format.content_type == ContentType.FLAC
61    assert stream_details.audio_format.sample_rate == 44100
62    assert stream_details.audio_format.bit_depth == 16
63    assert stream_details.stream_type == StreamType.HTTP
64    assert stream_details.path == "https://example.com/stream.flac"
65    assert stream_details.can_seek is True
66
67    provider_mock.get_track.assert_called_with("123")
68    provider_mock.api.get.assert_called_with(
69        "tracks/123/playbackinfopostpaywall",
70        params={
71            "playbackmode": "STREAM",
72            "assetpresentation": "FULL",
73            "audioquality": "HIGH",
74        },
75    )
76
77
78async def test_get_stream_details_hires(
79    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
80) -> None:
81    """Test get_stream_details with HIRES_LOSSLESS quality."""
82    provider_mock.get_track.return_value = mock_track
83    provider_mock.api.get.return_value = {
84        "urls": ["https://example.com/stream.flac"],
85        "audioQuality": "HIRES_LOSSLESS",
86        "sampleRate": 96000,
87        "bitDepth": 24,
88    }
89
90    stream_details = await streaming_manager.get_stream_details("123")
91
92    assert stream_details.audio_format.content_type == ContentType.FLAC
93    assert stream_details.audio_format.sample_rate == 96000
94    assert stream_details.audio_format.bit_depth == 24
95
96
97async def test_get_stream_details_with_dash_manifest(
98    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
99) -> None:
100    """Test get_stream_details with DASH manifest served via HTTP route."""
101    provider_mock.get_track.return_value = mock_track
102    provider_mock.api.get.return_value = {
103        "manifestMimeType": "application/dash+xml",
104        "manifest": "bWFuaWZlc3REYXRh",
105        "audioQuality": "HIGH",
106        "sampleRate": 44100,
107        "bitDepth": 16,
108    }
109    # Mock the stream server's dynamic route registration
110    provider_mock.mass.streams.register_dynamic_route = Mock(return_value=lambda: None)
111    provider_mock.mass.streams.base_url = "http://localhost:8097"
112
113    stream_details = await streaming_manager.get_stream_details("123")
114
115    assert isinstance(stream_details.path, str)
116    assert stream_details.path.startswith("http://localhost:8097/tidal-dash/")
117    assert "base64" not in stream_details.path
118    # Verify the route was registered
119    provider_mock.mass.streams.register_dynamic_route.assert_called_once()
120    # Verify cleanup was scheduled with duration-based TTL (180s track + 300s buffer = 480s)
121    provider_mock.mass.call_later.assert_called_with(
122        480,
123        streaming_manager._remove_dash_route,
124        "/tidal-dash/a3aca34e43c1737c7738507842972fa9",
125        task_id="tidal-dash-cleanup-a3aca34e43c1737c7738507842972fa9",
126    )
127
128
129async def test_get_stream_details_with_dash_manifest_handler(
130    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
131) -> None:
132    """Test the DASH manifest handler returns correct body and content type."""
133    provider_mock.get_track.return_value = mock_track
134    provider_mock.api.get.return_value = {
135        "manifestMimeType": "application/dash+xml",
136        "manifest": "bWFuaWZlc3REYXRh",
137        "audioQuality": "HIGH",
138        "sampleRate": 44100,
139        "bitDepth": 16,
140    }
141    # Capture the registered handler
142    register_mock = Mock()
143    provider_mock.mass.streams.register_dynamic_route = register_mock
144    provider_mock.mass.streams.base_url = "http://localhost:8097"
145
146    await streaming_manager.get_stream_details("123")
147
148    # Extract the handler that was registered as second argument
149    handler = register_mock.call_args[0][1]
150    response = await handler(Mock())
151
152    assert response.body == b"manifestData"  # base64 decoded
153    assert response.content_type == "application/dash+xml"
154    # Verify Cache-Control header is set to prevent proxy caching
155    assert response.headers.get("Cache-Control") == "no-cache"
156
157
158async def test_get_stream_details_with_dash_manifest_duplicate_registration(
159    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
160) -> None:
161    """Test cleanup is still scheduled when reusing an already-registered route."""
162    provider_mock.get_track.return_value = mock_track
163    provider_mock.api.get.return_value = {
164        "manifestMimeType": "application/dash+xml",
165        "manifest": "bWFuaWZlc3REYXRh",
166        "audioQuality": "HIGH",
167        "sampleRate": 44100,
168        "bitDepth": 16,
169    }
170    # First call succeeds, second raises RuntimeError (duplicate route)
171    provider_mock.mass.streams.register_dynamic_route = Mock(
172        side_effect=[None, RuntimeError("duplicate")]
173    )
174    provider_mock.mass.streams.base_url = "http://localhost:8097"
175    # call_later must count calls
176    call_later_mock = Mock()
177    provider_mock.mass.call_later = call_later_mock
178
179    # Call get_stream_details twice (same track → same manifest hash)
180    await streaming_manager.get_stream_details("123")
181    await streaming_manager.get_stream_details("123")
182
183    # route registration was attempted twice
184    assert provider_mock.mass.streams.register_dynamic_route.call_count == 2
185    # cleanup was scheduled both times (not skipped on duplicate)
186    assert call_later_mock.call_count == 2
187
188
189async def test_remove_dash_route(
190    streaming_manager: TidalStreamingManager, provider_mock: Mock
191) -> None:
192    """Test _remove_dash_route calls unregister on the stream server."""
193    unregister_mock = Mock()
194    provider_mock.mass.streams.unregister_dynamic_route = unregister_mock
195
196    streaming_manager._remove_dash_route("/tidal-dash/abc123")
197
198    unregister_mock.assert_called_once_with("/tidal-dash/abc123", method="GET")
199
200
201async def test_remove_dash_route_handles_runtime_error(
202    streaming_manager: TidalStreamingManager, provider_mock: Mock
203) -> None:
204    """Test _remove_dash_route silently swallows RuntimeError."""
205    provider_mock.mass.streams.unregister_dynamic_route = Mock(
206        side_effect=RuntimeError("not found")
207    )
208
209    # Should not raise
210    streaming_manager._remove_dash_route("/tidal-dash/abc123")
211
212
213async def test_get_stream_details_with_codec(
214    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
215) -> None:
216    """Test get_stream_details with codec specified."""
217    provider_mock.get_track.return_value = mock_track
218    provider_mock.api.get.return_value = {
219        "urls": ["https://example.com/stream.aac"],
220        "audioQuality": "HIGH",
221        "codec": "AAC",
222        "sampleRate": 44100,
223        "bitDepth": 16,
224    }
225
226    stream_details = await streaming_manager.get_stream_details("123")
227
228    assert stream_details.audio_format.content_type == ContentType.AAC
229
230
231async def test_get_stream_details_defaults_to_mp4(
232    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
233) -> None:
234    """Test get_stream_details defaults to MP4 when no quality/codec."""
235    provider_mock.get_track.return_value = mock_track
236    provider_mock.api.get.return_value = {
237        "urls": ["https://example.com/stream.m4a"],
238        "sampleRate": 44100,
239        "bitDepth": 16,
240    }
241
242    stream_details = await streaming_manager.get_stream_details("123")
243
244    assert stream_details.audio_format.content_type == ContentType.MP4
245
246
247async def test_get_stream_details_no_urls_raises_error(
248    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
249) -> None:
250    """Test get_stream_details raises error when no URLs."""
251    provider_mock.get_track.return_value = mock_track
252    provider_mock.api.get.return_value = {
253        "audioQuality": "HIGH",
254        "sampleRate": 44100,
255        "bitDepth": 16,
256    }
257
258    with pytest.raises(MediaNotFoundError, match="No stream URL found"):
259        await streaming_manager.get_stream_details("123")
260
261
262async def test_get_stream_details_track_not_found_no_isrc(
263    streaming_manager: TidalStreamingManager, provider_mock: Mock
264) -> None:
265    """Test get_stream_details when track not found and resolver can't resolve it."""
266    provider_mock.get_track.side_effect = MediaNotFoundError("Track not found")
267    provider_mock.resolve_live_track_id.return_value = None
268
269    with pytest.raises(MediaNotFoundError, match="Track 123 not found"):
270        await streaming_manager.get_stream_details("123")
271
272    provider_mock.resolve_live_track_id.assert_called_with("123")
273
274
275async def test_get_stream_details_with_isrc_fallback(
276    streaming_manager: TidalStreamingManager, provider_mock: Mock
277) -> None:
278    """Test get_stream_details uses the resolver fallback when direct lookup fails."""
279    # The live track carries a DIFFERENT id, so the assertions below can tell the
280    # resolved track from the requested (dead) one.
281    live_track = Mock(spec=Track)
282    live_track.item_id = "456"
283    live_track.duration = 180
284    provider_mock.get_track.side_effect = [
285        MediaNotFoundError("Track not found"),  # First call
286        live_track,  # Second call: fetch the resolved live track
287    ]
288    provider_mock.resolve_live_track_id.return_value = "456"
289
290    provider_mock.api.get.return_value = {
291        "urls": ["https://example.com/stream.flac"],
292        "audioQuality": "LOSSLESS",
293        "sampleRate": 44100,
294        "bitDepth": 16,
295    }
296
297    stream_details = await streaming_manager.get_stream_details("123")
298
299    provider_mock.resolve_live_track_id.assert_called_with("123")
300    provider_mock.get_track.assert_called_with("456")
301    assert stream_details.item_id == "456"
302    assert provider_mock.api.get.call_args[0][0] == "tracks/456/playbackinfopostpaywall"
303    assert stream_details.path == "https://example.com/stream.flac"
304
305
306async def test_get_stream_details_heals_on_playback_info_404(
307    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
308) -> None:
309    """
310    Test a cached track whose id churned is healed when the playback info 404s.
311
312    The track lookup is cached for days, so a churned id passes the lookup and the
313    404 first surfaces on the playbackinfo request; that must trigger the resolver
314    and one retry with the live id instead of failing playback.
315    """
316    live_track = Mock(spec=Track)
317    live_track.item_id = "456"
318    live_track.duration = 180
319    provider_mock.get_track.side_effect = [mock_track, live_track]
320    provider_mock.resolve_live_track_id.return_value = "456"
321    provider_mock.api.get.side_effect = [
322        MediaNotFoundError("Track not found"),  # playback info for the dead cached id
323        {
324            "urls": ["https://example.com/stream.flac"],
325            "audioQuality": "LOSSLESS",
326            "sampleRate": 44100,
327            "bitDepth": 16,
328        },
329    ]
330
331    stream_details = await streaming_manager.get_stream_details("123")
332
333    provider_mock.resolve_live_track_id.assert_called_once_with("123")
334    assert provider_mock.api.get.call_count == 2
335    assert provider_mock.api.get.call_args_list[1][0][0] == "tracks/456/playbackinfopostpaywall"
336    assert stream_details.path == "https://example.com/stream.flac"
337
338
339async def test_get_stream_details_playback_info_404_unresolvable(
340    streaming_manager: TidalStreamingManager, provider_mock: Mock, mock_track: Mock
341) -> None:
342    """Test an unresolvable playback-info 404 propagates instead of retrying blindly."""
343    provider_mock.get_track.return_value = mock_track
344    provider_mock.resolve_live_track_id.return_value = None
345    provider_mock.api.get.side_effect = MediaNotFoundError("Track not found")
346
347    with pytest.raises(MediaNotFoundError):
348        await streaming_manager.get_stream_details("123")
349
350    assert provider_mock.api.get.call_count == 1
351
352
353async def test_get_stream_details_schedules_background_mapping_update(
354    streaming_manager: TidalStreamingManager,
355    provider_mock: Mock,
356    mock_track: Mock,
357    monkeypatch: pytest.MonkeyPatch,
358) -> None:
359    """Ensure get_stream_details schedules the background mapping update task."""
360    provider_mock.get_track.return_value = mock_track
361    provider_mock.api.get.return_value = {
362        "urls": ["https://example.com/stream.flac"],
363        "audioQuality": "LOSSLESS",
364        "sampleRate": 44100,
365        "bitDepth": 16,
366    }
367
368    created: list[tuple[str, AudioFormat]] = []
369
370    async def _fake_worker(provider_track_id: str, resolved_audio_format: AudioFormat) -> None:
371        created.append((provider_track_id, resolved_audio_format))
372
373    # Patch the worker method so we can validate the coroutine is created with expected args
374    monkeypatch.setattr(
375        streaming_manager, "_async_update_provider_mapping_audio_format", _fake_worker
376    )
377
378    captured_coros: list[Coroutine[Any, Any, None]] = []
379
380    def _fake_create_task(coro: Coroutine[Any, Any, None]) -> None:
381        # Don't schedule; just capture the coroutine so the test can await it.
382        captured_coros.append(coro)
383
384    provider_mock.mass.create_task = _fake_create_task
385
386    stream_details = await streaming_manager.get_stream_details("123")
387
388    assert len(captured_coros) == 1
389
390    # Execute the captured coroutine (safe because we patched the worker)
391    await captured_coros[0]
392
393    assert created == [("123", stream_details.audio_format)]
394
395
396async def test_async_update_provider_mapping_audio_format_no_library_item(
397    streaming_manager: TidalStreamingManager, provider_mock: Mock
398) -> None:
399    """Ensure no update occurs when no library item is found."""
400    provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = None
401    provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
402
403    await streaming_manager._async_update_provider_mapping_audio_format(
404        provider_track_id="123",
405        resolved_audio_format=AudioFormat(
406            content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
407        ),
408    )
409
410    provider_mock.mass.music.tracks.update_provider_mapping.assert_not_called()
411
412
413async def test_async_update_provider_mapping_audio_format_no_mapping(
414    streaming_manager: TidalStreamingManager, provider_mock: Mock
415) -> None:
416    """Ensure no update occurs when no provider mapping is found."""
417    lib_track = Mock()
418    lib_track.item_id = 1
419    lib_track.provider_mappings = set()
420    provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
421    provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
422
423    await streaming_manager._async_update_provider_mapping_audio_format(
424        provider_track_id="123",
425        resolved_audio_format=AudioFormat(
426            content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
427        ),
428    )
429
430    provider_mock.mass.music.tracks.update_provider_mapping.assert_not_called()
431
432
433async def test_async_update_provider_mapping_audio_format_same_format_no_update(
434    streaming_manager: TidalStreamingManager, provider_mock: Mock
435) -> None:
436    """Ensure no update occurs when the audio format is unchanged."""
437    fmt = AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16)
438    mapping = Mock()
439    mapping.provider_instance = provider_mock.instance_id
440    mapping.item_id = "123"
441    mapping.audio_format = fmt
442
443    lib_track = Mock()
444    lib_track.item_id = 1
445    lib_track.provider_mappings = {mapping}
446    provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
447    provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
448
449    await streaming_manager._async_update_provider_mapping_audio_format(
450        provider_track_id="123",
451        resolved_audio_format=fmt,
452    )
453
454    provider_mock.mass.music.tracks.update_provider_mapping.assert_not_called()
455
456
457async def test_async_update_provider_mapping_audio_format_different_format_updates(
458    streaming_manager: TidalStreamingManager, provider_mock: Mock
459) -> None:
460    """Ensure update occurs when the audio format is different."""
461    old_fmt = AudioFormat(content_type=ContentType.MP4, sample_rate=44100, bit_depth=16)
462    new_fmt = AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16)
463
464    mapping = Mock()
465    mapping.provider_instance = provider_mock.instance_id
466    mapping.item_id = "123"
467    mapping.audio_format = old_fmt
468
469    lib_track = Mock()
470    lib_track.item_id = 1
471    lib_track.provider_mappings = {mapping}
472    provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
473    provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock()
474
475    await streaming_manager._async_update_provider_mapping_audio_format(
476        provider_track_id="123",
477        resolved_audio_format=new_fmt,
478    )
479
480    provider_mock.mass.music.tracks.update_provider_mapping.assert_awaited_once()
481    provider_mock.mass.music.tracks.update_provider_mapping.assert_awaited_with(
482        item_id=1,
483        provider_instance_id=provider_mock.instance_id,
484        provider_item_id="123",
485        audio_format=new_fmt,
486    )
487
488
489async def test_async_update_provider_mapping_audio_format_sqlite_operational_error_logs_debug(
490    streaming_manager: TidalStreamingManager, provider_mock: Mock
491) -> None:
492    """Ensure OperationalError is logged at debug level."""
493    provider_mock.logger = Mock()
494    provider_mock.mass.music.tracks.get_library_item_by_prov_id.side_effect = OperationalError(
495        "database is locked"
496    )
497
498    await streaming_manager._async_update_provider_mapping_audio_format(
499        provider_track_id="123",
500        resolved_audio_format=AudioFormat(
501            content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
502        ),
503    )
504
505    provider_mock.logger.debug.assert_called()
506
507
508async def test_async_update_provider_mapping_audio_format_unexpected_error_logs_exception(
509    streaming_manager: TidalStreamingManager, provider_mock: Mock
510) -> None:
511    """Ensure unexpected errors are logged at exception level."""
512    provider_mock.logger = Mock()
513
514    lib_track = Mock()
515    lib_track.item_id = 1
516    lib_track.provider_mappings = set()
517    provider_mock.mass.music.tracks.get_library_item_by_prov_id.return_value = lib_track
518
519    # Force an unexpected error after resolving lib_track
520    provider_mock.mass.music.tracks.update_provider_mapping = AsyncMock(
521        side_effect=RuntimeError("boom")
522    )
523
524    # Create a mapping that triggers the update path
525    mapping = Mock()
526    mapping.provider_instance = provider_mock.instance_id
527    mapping.item_id = "123"
528    mapping.audio_format = AudioFormat(
529        content_type=ContentType.MP4, sample_rate=44100, bit_depth=16
530    )
531    lib_track.provider_mappings = {mapping}
532
533    await streaming_manager._async_update_provider_mapping_audio_format(
534        provider_track_id="123",
535        resolved_audio_format=AudioFormat(
536            content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16
537        ),
538    )
539
540    provider_mock.logger.exception.assert_called()
541