music-assistant-server

1"""Tests for get_stream_details with direct FLAC streaming support."""
2
3from __future__ import annotations
4
5from typing import Any, cast
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.enums import ContentType
10from music_assistant_models.errors import MediaNotFoundError
11from zvuk_music import StreamQuality
12
13from music_assistant.providers.zvuk_music.api_client import ZvukMusicClient
14from music_assistant.providers.zvuk_music.provider import ZvukMusicProvider
15
16
17def _make_mock_track(has_flac: bool = True, duration: int = 240) -> MagicMock:
18    """
19    Create a mock ZvukTrack with configurable has_flac and duration.
20
21    :param has_flac: Whether FLAC is available for this track.
22    :param duration: Track duration in seconds.
23    :return: Mock track object.
24    """
25    track = MagicMock()
26    track.has_flac = has_flac
27    track.duration = duration
28    return track
29
30
31def _make_provider(quality_pref: str) -> ZvukMusicProvider:
32    """
33    Create a ZvukMusicProvider with mocked MA and config.
34
35    :param quality_pref: Quality preference string ("lossless" or "high").
36    :return: Configured provider instance.
37    """
38    provider = MagicMock(spec=ZvukMusicProvider)
39
40    config = MagicMock()
41    config.get_value = MagicMock(return_value=quality_pref)
42    provider.config = config
43
44    provider.instance_id = "zvuk_music--test"
45    provider.client = MagicMock(spec=ZvukMusicClient)
46    provider.logger = MagicMock()
47
48    return provider
49
50
51class TestGetDirectStreamUrl:
52    """Tests for ZvukMusicClient.get_direct_stream_url."""
53
54    def _mock_inner(self, zvuk_client: ZvukMusicClient, stream_url: str | None) -> AsyncMock:
55        """Patch _ensure_connected; inner mock's get_direct_stream_url returns stream_url."""
56        mock_inner_client = MagicMock()
57        mock_inner_client.get_direct_stream_url = AsyncMock(
58            return_value=MagicMock(stream=stream_url) if stream_url else None
59        )
60        cast("Any", zvuk_client)._ensure_connected = MagicMock(return_value=mock_inner_client)
61        return cast("AsyncMock", mock_inner_client.get_direct_stream_url)
62
63    @pytest.mark.asyncio
64    async def test_returns_stream_url_for_flac(self) -> None:
65        """get_direct_stream_url returns the stream URL from the library response."""
66        client = ZvukMusicClient(token="test")
67        self._mock_inner(client, "https://cdn.zvuk.com/track.flac?token=abc")
68
69        url = await client.get_direct_stream_url("12345", "flac")
70
71        assert url == "https://cdn.zvuk.com/track.flac?token=abc"
72
73    @pytest.mark.asyncio
74    async def test_returns_none_when_library_returns_none(self) -> None:
75        """get_direct_stream_url returns None when library returns None."""
76        client = ZvukMusicClient(token="test")
77        self._mock_inner(client, None)
78
79        url = await client.get_direct_stream_url("12345", "flac")
80
81        assert url is None
82
83    @pytest.mark.asyncio
84    async def test_passes_quality_as_stream_quality_enum(self) -> None:
85        """get_direct_stream_url passes StreamQuality enum to the library."""
86        client = ZvukMusicClient(token="test")
87        mock_lib = self._mock_inner(client, "https://cdn.zvuk.com/track.mp3")
88
89        await client.get_direct_stream_url("99999", "high")
90
91        mock_lib.assert_awaited_once_with("99999", StreamQuality.HIGH)
92
93
94class TestGetStreamDetailsFlac:
95    """Tests for get_stream_details with FLAC streaming logic."""
96
97    @pytest.mark.asyncio
98    async def test_lossless_with_has_flac_requests_flac(self) -> None:
99        """When lossless is requested and FLAC URL is available, ContentType.FLAC is returned."""
100        provider = MagicMock(spec=ZvukMusicProvider)
101        provider.config = MagicMock()
102        provider.config.get_value = MagicMock(return_value="lossless")
103        provider.instance_id = "zvuk_music--test"
104
105        mock_client = MagicMock(spec=ZvukMusicClient)
106        mock_client.get_track = AsyncMock(return_value=_make_mock_track(has_flac=True))
107        mock_client.get_direct_stream_url = AsyncMock(
108            return_value="https://cdn.zvuk.com/track.flac"
109        )
110        provider.client = mock_client
111        provider.logger = MagicMock()
112
113        result = await ZvukMusicProvider.get_stream_details(provider, "12345")
114
115        assert result.audio_format.content_type == ContentType.FLAC
116        assert result.audio_format.bit_rate == 0
117        mock_client.get_direct_stream_url.assert_called_with("12345", "flac")
118
119    @pytest.mark.asyncio
120    async def test_lossless_always_tries_flac_first(self) -> None:
121        """When lossless is requested, FLAC is always attempted first regardless of has_flac."""
122        provider = MagicMock(spec=ZvukMusicProvider)
123        provider.config = MagicMock()
124        provider.config.get_value = MagicMock(return_value="lossless")
125        provider.instance_id = "zvuk_music--test"
126
127        mock_client = MagicMock(spec=ZvukMusicClient)
128        # has_flac=False but tiny API actually returns FLAC URL
129        mock_client.get_track = AsyncMock(return_value=_make_mock_track(has_flac=False))
130        mock_client.get_direct_stream_url = AsyncMock(
131            return_value="https://cdn.zvuk.com/track.flac"
132        )
133        provider.client = mock_client
134        provider.logger = MagicMock()
135
136        result = await ZvukMusicProvider.get_stream_details(provider, "12345")
137
138        assert result.audio_format.content_type == ContentType.FLAC
139        # Must try flac first even when has_flac=False
140        calls = [c.args[1] for c in mock_client.get_direct_stream_url.call_args_list]
141        assert calls[0] == "flac"
142
143    @pytest.mark.asyncio
144    async def test_flac_url_failure_falls_back_to_high(self) -> None:
145        """When FLAC URL request returns None, falls back to HIGH MP3."""
146        provider = MagicMock(spec=ZvukMusicProvider)
147        provider.config = MagicMock()
148        provider.config.get_value = MagicMock(return_value="lossless")
149        provider.instance_id = "zvuk_music--test"
150
151        mock_client = MagicMock(spec=ZvukMusicClient)
152        mock_client.get_track = AsyncMock(return_value=_make_mock_track(has_flac=True))
153
154        def stream_url_side_effect(_track_id: str, quality: str) -> str | None:
155            if quality == "flac":
156                return None  # FLAC unavailable for this track
157            return "https://cdn.zvuk.com/track.mp3"
158
159        mock_client.get_direct_stream_url = AsyncMock(side_effect=stream_url_side_effect)
160        provider.client = mock_client
161        provider.logger = MagicMock()
162
163        result = await ZvukMusicProvider.get_stream_details(provider, "12345")
164
165        assert result.audio_format.content_type == ContentType.MP3
166        assert result.audio_format.bit_rate == 320
167
168    @pytest.mark.asyncio
169    async def test_high_quality_pref_skips_flac(self) -> None:
170        """When high (not lossless) is preferred, FLAC is never requested."""
171        provider = MagicMock(spec=ZvukMusicProvider)
172        provider.config = MagicMock()
173        provider.config.get_value = MagicMock(return_value="high")
174        provider.instance_id = "zvuk_music--test"
175
176        mock_client = MagicMock(spec=ZvukMusicClient)
177        mock_client.get_track = AsyncMock(return_value=_make_mock_track(has_flac=True))
178        mock_client.get_direct_stream_url = AsyncMock(return_value="https://cdn.zvuk.com/track.mp3")
179        provider.client = mock_client
180        provider.logger = MagicMock()
181
182        result = await ZvukMusicProvider.get_stream_details(provider, "12345")
183
184        assert result.audio_format.content_type == ContentType.MP3
185        calls = [c.args[1] for c in mock_client.get_direct_stream_url.call_args_list]
186        assert "flac" not in calls
187
188    @pytest.mark.asyncio
189    async def test_stream_path_is_url(self) -> None:
190        """StreamDetails.path contains the URL returned by get_direct_stream_url."""
191        provider = MagicMock(spec=ZvukMusicProvider)
192        provider.config = MagicMock()
193        provider.config.get_value = MagicMock(return_value="lossless")
194        provider.instance_id = "zvuk_music--test"
195
196        expected_url = "https://cdn.zvuk.com/track.flac?token=xyz"
197        mock_client = MagicMock(spec=ZvukMusicClient)
198        mock_client.get_track = AsyncMock(return_value=_make_mock_track(has_flac=True))
199        mock_client.get_direct_stream_url = AsyncMock(return_value=expected_url)
200        provider.client = mock_client
201        provider.logger = MagicMock()
202
203        result = await ZvukMusicProvider.get_stream_details(provider, "12345")
204
205        assert result.path == expected_url
206
207    @pytest.mark.asyncio
208    async def test_duration_from_track_metadata(self) -> None:
209        """StreamDetails.duration is populated from track.duration."""
210        provider = MagicMock(spec=ZvukMusicProvider)
211        provider.config = MagicMock()
212        provider.config.get_value = MagicMock(return_value="high")
213        provider.instance_id = "zvuk_music--test"
214
215        mock_client = MagicMock(spec=ZvukMusicClient)
216        mock_client.get_track = AsyncMock(
217            return_value=_make_mock_track(has_flac=False, duration=333)
218        )
219        mock_client.get_direct_stream_url = AsyncMock(return_value="https://cdn.zvuk.com/track.mp3")
220        provider.client = mock_client
221        provider.logger = MagicMock()
222
223        result = await ZvukMusicProvider.get_stream_details(provider, "12345")
224
225        assert result.duration == 333
226
227    @pytest.mark.asyncio
228    async def test_raises_when_all_urls_none(self) -> None:
229        """MediaNotFoundError is raised when all quality attempts return None."""
230        provider = MagicMock(spec=ZvukMusicProvider)
231        provider.config = MagicMock()
232        provider.config.get_value = MagicMock(return_value="high")
233        provider.instance_id = "zvuk_music--test"
234
235        mock_client = MagicMock(spec=ZvukMusicClient)
236        mock_client.get_track = AsyncMock(return_value=_make_mock_track(has_flac=False))
237        mock_client.get_direct_stream_url = AsyncMock(return_value=None)
238        provider.client = mock_client
239        provider.logger = MagicMock()
240
241        with pytest.raises(MediaNotFoundError):
242            await ZvukMusicProvider.get_stream_details(provider, "12345")
243