/
/
/
1"""
2Unit tests for YandexMusicProvider (provider.py).
3
4These tests construct a partial provider instance via ``__new__`` (no
5``__init__``), attach the attributes the method-under-test reads, and
6exercise it directly. The pattern avoids the upstream Music Assistant
7provider-init machinery which would otherwise drag in a real
8``MusicAssistant`` instance.
9"""
10
11from __future__ import annotations
12
13import asyncio
14import logging
15from typing import TYPE_CHECKING, Any, cast
16from unittest import mock
17
18import pytest
19from music_assistant_models.errors import LoginFailed, ResourceTemporarilyUnavailable
20
21from music_assistant.models.music_provider import MusicProvider
22from music_assistant.providers.yandex_music.constants import (
23 CONF_BASE_URL,
24 CONF_REFRESH_TOKEN,
25 CONF_RESTRICTIVE_RATE_LIMITS,
26 CONF_TOKEN,
27 CONF_X_TOKEN,
28 DEFAULT_BASE_URL,
29 MY_WAVE_PLAYLIST_ID,
30 TRACK_BATCH_SIZE,
31)
32from music_assistant.providers.yandex_music.provider import YandexMusicProvider
33
34from .conftest import use_real_create_task
35
36# This fork-only setup key may not exist in the installed upstream package used by local mypy.
37_CONF_MANUAL_TOKEN = "manual_token"
38
39if TYPE_CHECKING:
40 from collections.abc import Callable
41
42
43def _make_provider() -> tuple[YandexMusicProvider, mock.AsyncMock]:
44 """
45 Return a YandexMusicProvider with a mocked Yandex API client.
46
47 The provider is constructed without calling ``__init__`` so the upstream
48 base-class init does not run. ``client`` is a property over ``_client``,
49 so we assign the underlying attribute directly.
50 """
51 provider = YandexMusicProvider.__new__(YandexMusicProvider)
52 mock_client = mock.AsyncMock()
53 mock_client.user_id = 12345
54 provider._client = mock_client
55 provider.logger = mock.MagicMock()
56 # ``instance_id`` and ``domain`` on the base class read from ``self.config``;
57 # tests that need them attach a minimal config stub.
58 return provider, mock_client
59
60
61def _make_auth_init_provider(
62 manual_token: str,
63) -> tuple[YandexMusicProvider, mock.MagicMock, mock.MagicMock]:
64 """Build the provider state needed to exercise manual-token initialization."""
65 provider = YandexMusicProvider.__new__(YandexMusicProvider)
66 provider._client = None
67 provider.mass = mock.MagicMock()
68 provider.logger = mock.MagicMock()
69 provider.logger.level = logging.INFO
70 values = {
71 _CONF_MANUAL_TOKEN: manual_token,
72 CONF_BASE_URL: DEFAULT_BASE_URL,
73 CONF_RESTRICTIVE_RATE_LIMITS: False,
74 }
75 provider.config = mock.MagicMock()
76 provider.config.get_value = mock.MagicMock(
77 side_effect=lambda key, default=None: values.get(key, default)
78 )
79 setup_values = {
80 CONF_TOKEN: "old-token",
81 CONF_X_TOKEN: "old-x-token",
82 CONF_REFRESH_TOKEN: "old-refresh-token",
83 }
84 update_setup_data = mock.MagicMock()
85 update_config_value = mock.MagicMock()
86 untyped_provider = cast("Any", provider)
87 untyped_provider.get_setup_value = mock.MagicMock(side_effect=setup_values.get)
88 untyped_provider._update_setup_data = update_setup_data
89 untyped_provider._update_config_value = update_config_value
90 return provider, update_setup_data, update_config_value
91
92
93async def test_manual_token_replacement_is_validated_then_promoted() -> None:
94 """A working replacement supersedes and removes every old session credential."""
95 provider, update_setup_data, update_config_value = _make_auth_init_provider("new-token")
96 client = mock.AsyncMock()
97
98 with (
99 mock.patch(
100 "music_assistant.providers.yandex_music.provider.YandexMusicClient",
101 return_value=client,
102 ) as client_class,
103 mock.patch("music_assistant.providers.yandex_music.provider.YandexMusicStreamingManager"),
104 mock.patch(
105 "music_assistant.providers.yandex_music.provider.refresh_music_token",
106 new=mock.AsyncMock(),
107 ) as refresh_music_token,
108 ):
109 await provider.handle_async_init()
110
111 supplied_token = client_class.call_args.args[0]
112 assert supplied_token.get_secret() == "new-token"
113 client.connect.assert_awaited_once()
114 refresh_music_token.assert_not_awaited()
115 update_setup_data.assert_has_calls(
116 [
117 mock.call(CONF_TOKEN, "new-token"),
118 mock.call(CONF_X_TOKEN, None),
119 mock.call(CONF_REFRESH_TOKEN, None),
120 ]
121 )
122 update_config_value.assert_called_once_with(_CONF_MANUAL_TOKEN, None, immediate=True)
123
124
125async def test_invalid_manual_token_keeps_existing_setup_credentials() -> None:
126 """A rejected replacement is discarded without damaging working setup data."""
127 provider, update_setup_data, update_config_value = _make_auth_init_provider("invalid-token")
128 client = mock.AsyncMock()
129 client.connect.side_effect = LoginFailed("rejected")
130
131 with (
132 mock.patch(
133 "music_assistant.providers.yandex_music.provider.YandexMusicClient",
134 return_value=client,
135 ) as client_class,
136 mock.patch(
137 "music_assistant.providers.yandex_music.provider.refresh_music_token",
138 new=mock.AsyncMock(),
139 ) as refresh_music_token,
140 pytest.raises(LoginFailed, match="rejected"),
141 ):
142 await provider.handle_async_init()
143
144 supplied_token = client_class.call_args.args[0]
145 assert supplied_token.get_secret() == "invalid-token"
146 assert client_class.call_count == 1
147 refresh_music_token.assert_not_awaited()
148 update_setup_data.assert_not_called()
149 update_config_value.assert_called_once_with(_CONF_MANUAL_TOKEN, None, immediate=True)
150
151
152# -- M4: get_playlist_tracks must not abort on a single empty batch -----------
153
154
155async def test_get_playlist_tracks_continues_on_empty_batch() -> None:
156 """
157 An empty batch mid-load must skip-and-continue, not discard prior batches.
158
159 With ``TRACK_BATCH_SIZE`` of 50, a 150-track playlist resolves in 3 batches.
160 If batch 2 transiently returns an empty list, the previous behavior raised
161 ``ResourceTemporarilyUnavailable`` and threw away the 50 tracks already
162 fetched from batch 1. The fixed behavior logs a warning and continues so
163 the user sees a partial playlist rather than nothing.
164 """
165 provider, mock_client = _make_provider()
166
167 # Build 3 batches of TRACK_BATCH_SIZE tracks (= 150 track refs total)
168 total = TRACK_BATCH_SIZE * 3
169 track_refs = [type("TR", (), {"track_id": str(i)})() for i in range(total)]
170 playlist_obj = type(
171 "PL",
172 (),
173 {
174 "tracks": track_refs,
175 "track_count": total,
176 },
177 )()
178 mock_client.get_playlist = mock.AsyncMock(return_value=playlist_obj)
179
180 batch1 = [type("YT", (), {"id": i})() for i in range(TRACK_BATCH_SIZE)]
181 batch3 = [
182 type("YT", (), {"id": i})() for i in range(2 * TRACK_BATCH_SIZE, 3 * TRACK_BATCH_SIZE)
183 ]
184 mock_client.get_tracks = mock.AsyncMock(side_effect=[batch1, [], batch3])
185
186 # parse_track does heavy MA object construction; bypass it for this test
187 # so we measure the batch-loop behavior, not the parser.
188 with mock.patch(
189 "music_assistant.providers.yandex_music.provider.parse_track",
190 side_effect=lambda _self, t: t,
191 ):
192 cached: Any = YandexMusicProvider._get_regular_playlist_tracks
193 result = await cached.__wrapped__(provider, "12345:67", 0)
194
195 # 50 from batch 1 + 50 from batch 3 = 100 tracks; the empty batch 2 must
196 # not abort the load.
197 assert len(result) == 2 * TRACK_BATCH_SIZE
198 assert mock_client.get_tracks.await_count == 3
199 # The warning must still be emitted so operators can see the empty batch.
200 assert any(
201 "empty" in str(c.args).lower() or "empty" in str(c.kwargs).lower()
202 for c in provider.logger.warning.call_args_list # type: ignore[attr-defined]
203 )
204
205
206async def test_get_playlist_tracks_raises_only_when_every_batch_is_empty() -> None:
207 """If every batch is empty, the existing terminal guard still fires."""
208 provider, mock_client = _make_provider()
209
210 total = TRACK_BATCH_SIZE * 2
211 track_refs = [type("TR", (), {"track_id": str(i)})() for i in range(total)]
212 playlist_obj = type(
213 "PL",
214 (),
215 {
216 "tracks": track_refs,
217 "track_count": total,
218 },
219 )()
220 mock_client.get_playlist = mock.AsyncMock(return_value=playlist_obj)
221 mock_client.get_tracks = mock.AsyncMock(side_effect=[[], []])
222
223 cached: Any = YandexMusicProvider._get_regular_playlist_tracks
224 with pytest.raises(ResourceTemporarilyUnavailable):
225 await cached.__wrapped__(provider, "12345:67", 0)
226
227
228# -- request coalescing is scoped to the branch that must run per call --------
229
230
231class _StubConfig:
232 """Minimal provider config for the @use_cache decorator."""
233
234 instance_id = "yandex_music_test"
235
236 def get_value(self, key: str, default: Any = None) -> Any:
237 """Return the default for every config key."""
238 return default
239
240
241@pytest.fixture
242def cached_provider() -> tuple[YandexMusicProvider, mock.AsyncMock]:
243 """Return a provider with an empty cache and real task coalescing."""
244 provider, mock_client = _make_provider()
245 provider.mass = mock.MagicMock()
246 provider.mass.cache = mock.AsyncMock()
247 provider.mass.cache.get_with_freshness = mock.AsyncMock(return_value=(None, False, False))
248 provider.mass.cache.set = mock.AsyncMock()
249 use_real_create_task(provider.mass)
250 provider.config = _StubConfig() # type: ignore[assignment]
251 provider.manifest = mock.MagicMock(domain="yandex_music")
252 provider._wave_states = {}
253 return provider, mock_client
254
255
256async def _wait_for_gated_fetch(started: Callable[[], bool]) -> None:
257 """Wait until the gated fetch runs, then let the other callers catch up with it."""
258 for _ in range(200):
259 if started():
260 break
261 await asyncio.sleep(0.01)
262 else:
263 pytest.fail("gated fetch never started")
264 await asyncio.sleep(0.05)
265
266
267async def test_regular_playlist_fetch_is_shared_between_callers(
268 cached_provider: tuple[YandexMusicProvider, mock.AsyncMock],
269) -> None:
270 """Concurrent callers for the same regular playlist share one provider fetch."""
271 provider, mock_client = cached_provider
272 gate = asyncio.Event()
273
274 async def _get_playlist(*_args: Any, **_kwargs: Any) -> Any:
275 await gate.wait()
276 return type("PL", (), {"tracks": [], "track_count": 0})()
277
278 mock_client.get_playlist = mock.AsyncMock(side_effect=_get_playlist)
279
280 tasks = [asyncio.create_task(provider.get_playlist_tracks("12345:67")) for _ in range(3)]
281 await _wait_for_gated_fetch(lambda: mock_client.get_playlist.await_count > 0)
282 gate.set()
283
284 assert await asyncio.gather(*tasks) == [[], [], []]
285 assert mock_client.get_playlist.await_count == 1
286
287
288async def test_my_wave_fetch_is_shared_between_callers(
289 cached_provider: tuple[YandexMusicProvider, mock.AsyncMock],
290) -> None:
291 """Concurrent My Wave callers share one fetch, so the rotor advances once."""
292 provider, _ = cached_provider
293 gate = asyncio.Event()
294
295 async def _fetch_batch(*_args: Any, **_kwargs: Any) -> tuple[list[Any], None]:
296 await gate.wait()
297 return [], None
298
299 fetch_batch = mock.AsyncMock(side_effect=_fetch_batch)
300 provider._fetch_rotor_session_batch = fetch_batch # type: ignore[method-assign]
301
302 tasks = [
303 asyncio.create_task(provider.get_playlist_tracks(MY_WAVE_PLAYLIST_ID)) for _ in range(3)
304 ]
305 await _wait_for_gated_fetch(lambda: fetch_batch.await_count > 0)
306 gate.set()
307
308 assert await asyncio.gather(*tasks) == [[], [], []]
309 assert fetch_batch.await_count == 1
310
311
312# -- M11: unload() must clear every per-session cache -------------------------
313
314
315async def test_unload_clears_all_per_session_caches() -> None:
316 """
317 unload() must drop every state dict mirrored in handle_async_init().
318
319 Without this, stale ``asyncio.Lock`` instances bound to the previous
320 event loop survive a provider reload and break wave-session locking
321 after a settings change.
322 """
323 provider = YandexMusicProvider.__new__(YandexMusicProvider)
324 provider._client = None
325 provider._streaming = None
326 provider._wave_states = {"my_wave": mock.MagicMock()}
327 provider._wave_bg_colors = {"img-key": "#abcdef"}
328 provider._liked_albums_cache = (123.0, [])
329 provider._audiobook_chapter_cache = {"book-1": (["track-1"], [1000])}
330 provider._audiobook_play_ids = {"book-1": "play-id-1"}
331 provider.logger = mock.MagicMock()
332
333 with mock.patch.object(MusicProvider, "unload", new=mock.AsyncMock()):
334 await provider.unload()
335
336 # Cast through ``Any`` so mypy does not narrow these attributes to their
337 # post-assignment types â assert against the runtime values after unload.
338 state: Any = vars(provider)
339 assert not state["_wave_states"]
340 assert not state["_wave_bg_colors"]
341 assert state["_liked_albums_cache"] is None
342 assert not state["_audiobook_chapter_cache"]
343 assert not state["_audiobook_play_ids"]
344