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