/
/
/
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