/
/
/
1"""Test Yandex Music Recommendations."""
2
3from __future__ import annotations
4
5import hashlib
6import json
7import pathlib
8from datetime import UTC, datetime
9from typing import Any
10from unittest.mock import AsyncMock, Mock, patch
11
12import pytest
13from music_assistant_models.enums import MediaType, ProviderFeature
14from music_assistant_models.errors import InvalidDataError
15from music_assistant_models.media_items import (
16 Album,
17 Artist,
18 ItemMapping,
19 Playlist,
20 RecommendationFolder,
21 Track,
22 UniqueList,
23)
24from yandex_music import Track as YandexTrack
25
26from music_assistant.providers.yandex_music.constants import (
27 MY_WAVE_PLAYLIST_ID,
28 RADIO_TRACK_ID_SEP,
29 ROTOR_STATION_MY_WAVE,
30)
31from music_assistant.providers.yandex_music.provider import YandexMusicProvider, _WaveState
32
33from .conftest import DE_JSON_CLIENT, provider_dir, use_real_create_task
34
35_RECOMMENDATION_STRINGS = json.loads((provider_dir() / "strings.json").read_text(encoding="utf-8"))[
36 "media"
37]["recommendations"]
38
39ROW_IDS = [
40 MY_WAVE_PLAYLIST_ID,
41 "feed",
42 "chart",
43 "new_releases",
44 "new_playlists",
45 "top_picks",
46 "mood_mix",
47 "activity_mix",
48 "seasonal_mix",
49]
50
51
52def _media_item_mock(spec: type) -> Mock:
53 """
54 Return a media item stand-in that a copy hands back unchanged.
55
56 :param spec: The media item class the mock stands in for.
57 """
58 item = Mock(spec=spec)
59 item.__deepcopy__ = lambda _memo: item
60 return item
61
62
63@pytest.fixture
64def provider_mock() -> Mock:
65 """Return a mock Yandex Music provider."""
66 provider = Mock(spec=YandexMusicProvider)
67 provider.domain = "yandex_music"
68 provider.instance_id = "yandex_music_instance"
69 provider.supported_features = {ProviderFeature.RECOMMENDATIONS}
70 provider.logger = Mock()
71
72 # Mock client
73 provider.client = AsyncMock()
74 provider.client.user_id = 12345
75
76 # Mock config
77 provider.config = Mock()
78 provider.config.get_value = Mock(side_effect=lambda key: 150 if "max_tracks" in key else None)
79
80 # Mock mass with cache
81 provider.mass = Mock()
82 provider.mass.metadata = Mock()
83 provider.mass.metadata.locale = "en_US"
84 provider.mass.cache = AsyncMock()
85 provider.mass.cache.get = AsyncMock(return_value=None) # Cache always misses
86 provider.mass.cache.get_with_freshness = AsyncMock(return_value=(None, False, False))
87 provider.mass.cache.set = AsyncMock()
88 use_real_create_task(provider.mass)
89
90 # Resolve media labels through the real helper; unauthored keys fall back.
91 provider.mass.translations.get_translation = Mock(return_value=None)
92 provider._media_label = YandexMusicProvider._media_label.__get__(provider, YandexMusicProvider)
93
94 return provider
95
96
97def _install_wave_state(provider_mock: Mock) -> _WaveState:
98 """Stub _get_wave_state to return a fresh in-memory _WaveState per provider_mock."""
99 wave = _WaveState()
100 provider_mock._get_wave_state = Mock(return_value=wave)
101 return wave
102
103
104@pytest.mark.asyncio
105async def test_get_my_wave_recommendations_success(provider_mock: Mock) -> None:
106 """Test _get_my_wave_recommendations returns data when session API provides tracks."""
107 _install_wave_state(provider_mock)
108 mock_track = Mock()
109 mock_track.id = "12345"
110 mock_track.track_id = "12345"
111
112 # Mock the session-API helper; return the same track every time â matches
113 # the old single-track-per-batch test intent where the fake rotor returns
114 # the same shape across repeated batch calls.
115 provider_mock._fetch_rotor_session_batch = AsyncMock(return_value=([mock_track], "batch_a"))
116
117 mock_parsed_track = _media_item_mock(Track)
118 mock_parsed_track.item_id = f"12345{RADIO_TRACK_ID_SEP}{ROTOR_STATION_MY_WAVE}"
119 mock_parsed_track.name = "Test Track"
120 mock_parsed_track.provider_mappings = []
121 provider_mock._parse_my_wave_track = Mock(return_value=mock_parsed_track)
122
123 result = await YandexMusicProvider._get_my_wave_recommendations(provider_mock)
124
125 assert result is not None
126 assert isinstance(result, RecommendationFolder)
127 assert result.item_id == MY_WAVE_PLAYLIST_ID
128 assert result.provider == provider_mock.instance_id
129 assert result.name == _RECOMMENDATION_STRINGS[MY_WAVE_PLAYLIST_ID]["name"]
130 assert result.translation_key == MY_WAVE_PLAYLIST_ID
131 assert result.icon == "mdi-waveform"
132 assert len(result.items) > 0
133
134
135@pytest.mark.asyncio
136async def test_get_my_wave_recommendations_empty(provider_mock: Mock) -> None:
137 """Test _get_my_wave_recommendations returns None when session API yields no tracks."""
138 _install_wave_state(provider_mock)
139 provider_mock._fetch_rotor_session_batch = AsyncMock(return_value=([], None))
140
141 result = await YandexMusicProvider._get_my_wave_recommendations(provider_mock)
142
143 assert result is None
144
145
146@pytest.mark.asyncio
147async def test_get_my_wave_recommendations_duplicate_filtering(provider_mock: Mock) -> None:
148 """Test _get_my_wave_recommendations filters duplicate tracks across batches."""
149 _install_wave_state(provider_mock)
150 mock_track1 = Mock()
151 mock_track1.id = "12345"
152 mock_track1.track_id = "12345"
153
154 mock_track2 = Mock()
155 mock_track2.id = "12345" # Same ID
156 mock_track2.track_id = "12345"
157
158 # First batch returns track1, second batch returns track2 (duplicate)
159 provider_mock._fetch_rotor_session_batch = AsyncMock(
160 side_effect=[
161 ([mock_track1], "batch_a"),
162 ([mock_track2], "batch_b"),
163 ]
164 )
165
166 mock_parsed_track = _media_item_mock(Track)
167 mock_parsed_track.item_id = f"12345{RADIO_TRACK_ID_SEP}{ROTOR_STATION_MY_WAVE}"
168 mock_parsed_track.name = "Test Track"
169 mock_parsed_track.provider_mappings = []
170
171 # _parse_my_wave_track returns track on first call, None on duplicate
172 provider_mock._parse_my_wave_track = Mock(side_effect=[mock_parsed_track, None])
173
174 result = await YandexMusicProvider._get_my_wave_recommendations(provider_mock)
175
176 assert result is not None
177 # Should only have 1 track despite 2 API calls (duplicate filtered)
178 assert len(result.items) == 1
179
180
181@pytest.mark.asyncio
182async def test_get_my_wave_recommendations_invalid_data_error(provider_mock: Mock) -> None:
183 """Test _get_my_wave_recommendations handles parse failures gracefully."""
184 _install_wave_state(provider_mock)
185 mock_track = Mock()
186 mock_track.id = "12345"
187 mock_track.track_id = "12345"
188
189 provider_mock._fetch_rotor_session_batch = AsyncMock(return_value=([mock_track], "batch_a"))
190
191 # _parse_my_wave_track returns None (simulates parse error handled internally)
192 provider_mock._parse_my_wave_track = Mock(return_value=None)
193
194 result = await YandexMusicProvider._get_my_wave_recommendations(provider_mock)
195
196 # Should return None as no valid tracks were parsed
197 assert result is None
198
199
200@pytest.mark.asyncio
201async def test_get_feed_recommendations_success(provider_mock: Mock) -> None:
202 """Test _get_feed_recommendations returns data when API provides feed."""
203 # Mock feed with generated playlists
204 mock_gen_playlist = Mock()
205 mock_gen_playlist.ready = True
206 mock_gen_playlist.data = Mock() # Playlist data
207
208 mock_feed = Mock()
209 mock_feed.generated_playlists = [mock_gen_playlist]
210
211 provider_mock.client.get_feed = AsyncMock(return_value=mock_feed)
212
213 # Mock parse_playlist
214 mock_parsed_playlist = Mock(spec=Playlist)
215 mock_parsed_playlist.item_id = "playlist_1"
216 mock_parsed_playlist.name = "Playlist of the Day"
217
218 with patch(
219 "music_assistant.providers.yandex_music.provider.parse_playlist",
220 return_value=mock_parsed_playlist,
221 ):
222 result = await YandexMusicProvider._get_feed_recommendations(provider_mock)
223
224 assert result is not None
225 assert isinstance(result, RecommendationFolder)
226 assert result.item_id == "feed"
227 assert result.provider == provider_mock.instance_id
228 assert result.name == _RECOMMENDATION_STRINGS["feed"]["name"]
229 assert result.translation_key == "feed"
230 assert result.icon == "mdi-account-music"
231 assert len(result.items) > 0
232
233
234@pytest.mark.asyncio
235async def test_get_feed_recommendations_empty(provider_mock: Mock) -> None:
236 """Test _get_feed_recommendations returns None when feed is empty."""
237 provider_mock.client.get_feed = AsyncMock(return_value=None)
238
239 result = await YandexMusicProvider._get_feed_recommendations(provider_mock)
240
241 assert result is None
242
243
244@pytest.mark.asyncio
245async def test_get_feed_recommendations_no_generated_playlists(provider_mock: Mock) -> None:
246 """Test _get_feed_recommendations returns None when no generated playlists."""
247 mock_feed = Mock()
248 mock_feed.generated_playlists = []
249
250 provider_mock.client.get_feed = AsyncMock(return_value=mock_feed)
251
252 result = await YandexMusicProvider._get_feed_recommendations(provider_mock)
253
254 assert result is None
255
256
257@pytest.mark.asyncio
258async def test_get_feed_recommendations_invalid_data_error(provider_mock: Mock) -> None:
259 """Test _get_feed_recommendations handles InvalidDataError gracefully."""
260 mock_gen_playlist = Mock()
261 mock_gen_playlist.ready = True
262 mock_gen_playlist.data = Mock()
263
264 mock_feed = Mock()
265 mock_feed.generated_playlists = [mock_gen_playlist]
266
267 provider_mock.client.get_feed = AsyncMock(return_value=mock_feed)
268
269 with patch(
270 "music_assistant.providers.yandex_music.provider.parse_playlist",
271 side_effect=InvalidDataError("Parse error"),
272 ):
273 result = await YandexMusicProvider._get_feed_recommendations(provider_mock)
274
275 assert result is None
276 provider_mock.logger.debug.assert_called()
277
278
279@pytest.mark.asyncio
280async def test_get_chart_recommendations_success(provider_mock: Mock) -> None:
281 """Test _get_chart_recommendations returns data when API provides chart."""
282 # Mock TrackShort with .track attribute
283 mock_track_short = Mock()
284 mock_track_obj = Mock() # The actual Track object
285 mock_track_short.track = mock_track_obj
286
287 mock_chart = Mock()
288 mock_chart.tracks = [mock_track_short]
289
290 mock_chart_info = Mock()
291 mock_chart_info.chart = mock_chart
292
293 provider_mock.client.get_chart = AsyncMock(return_value=mock_chart_info)
294
295 # Mock parse_track
296 mock_parsed_track = _media_item_mock(Track)
297 mock_parsed_track.item_id = "track_1"
298 mock_parsed_track.name = "Chart Track 1"
299
300 with patch(
301 "music_assistant.providers.yandex_music.provider.parse_track",
302 return_value=mock_parsed_track,
303 ):
304 result = await YandexMusicProvider._get_chart_recommendations(provider_mock)
305
306 assert result is not None
307 assert isinstance(result, RecommendationFolder)
308 assert result.item_id == "chart"
309 assert result.provider == provider_mock.instance_id
310 assert result.name == _RECOMMENDATION_STRINGS["chart"]["name"]
311 assert result.translation_key == "chart"
312 assert result.icon == "mdi-chart-line"
313 assert len(result.items) > 0
314
315
316@pytest.mark.asyncio
317async def test_get_chart_recommendations_empty(provider_mock: Mock) -> None:
318 """Test _get_chart_recommendations returns None when chart is empty."""
319 provider_mock.client.get_chart = AsyncMock(return_value=None)
320
321 result = await YandexMusicProvider._get_chart_recommendations(provider_mock)
322
323 assert result is None
324
325
326@pytest.mark.asyncio
327async def test_get_chart_recommendations_no_tracks(provider_mock: Mock) -> None:
328 """Test _get_chart_recommendations returns None when chart has no tracks."""
329 mock_chart = Mock()
330 mock_chart.tracks = []
331
332 mock_chart_info = Mock()
333 mock_chart_info.chart = mock_chart
334
335 provider_mock.client.get_chart = AsyncMock(return_value=mock_chart_info)
336
337 result = await YandexMusicProvider._get_chart_recommendations(provider_mock)
338
339 assert result is None
340
341
342@pytest.mark.asyncio
343async def test_get_chart_recommendations_invalid_data_error(provider_mock: Mock) -> None:
344 """Test _get_chart_recommendations handles InvalidDataError gracefully."""
345 mock_track_short = Mock()
346 mock_track_obj = Mock()
347 mock_track_short.track = mock_track_obj
348
349 mock_chart = Mock()
350 mock_chart.tracks = [mock_track_short]
351
352 mock_chart_info = Mock()
353 mock_chart_info.chart = mock_chart
354
355 provider_mock.client.get_chart = AsyncMock(return_value=mock_chart_info)
356
357 with patch(
358 "music_assistant.providers.yandex_music.provider.parse_track",
359 side_effect=InvalidDataError("Parse error"),
360 ):
361 result = await YandexMusicProvider._get_chart_recommendations(provider_mock)
362
363 assert result is None
364 provider_mock.logger.debug.assert_called()
365
366
367@pytest.mark.asyncio
368async def test_get_new_releases_recommendations_success(provider_mock: Mock) -> None:
369 """Test _get_new_releases_recommendations returns data when API provides releases."""
370 # Mock releases with album IDs
371 mock_releases = Mock()
372 mock_releases.new_releases = [123, 456, 789]
373
374 provider_mock.client.get_new_releases = AsyncMock(return_value=mock_releases)
375
376 # Mock get_albums to return album objects
377 mock_album = Mock()
378 provider_mock.client.get_albums = AsyncMock(return_value=[mock_album])
379
380 # Mock parse_album
381 mock_parsed_album = Mock(spec=Album)
382 mock_parsed_album.item_id = "album_1"
383 mock_parsed_album.name = "New Album"
384
385 with patch(
386 "music_assistant.providers.yandex_music.provider.parse_album",
387 return_value=mock_parsed_album,
388 ):
389 result = await YandexMusicProvider._get_new_releases_recommendations(provider_mock)
390
391 assert result is not None
392 assert isinstance(result, RecommendationFolder)
393 assert result.item_id == "new_releases"
394 assert result.provider == provider_mock.instance_id
395 assert result.name == _RECOMMENDATION_STRINGS["new_releases"]["name"]
396 assert result.translation_key == "new_releases"
397 assert result.icon == "mdi-new-box"
398 assert len(result.items) > 0
399
400
401@pytest.mark.asyncio
402async def test_get_new_releases_recommendations_empty(provider_mock: Mock) -> None:
403 """Test _get_new_releases_recommendations returns None when releases are empty."""
404 provider_mock.client.get_new_releases = AsyncMock(return_value=None)
405
406 result = await YandexMusicProvider._get_new_releases_recommendations(provider_mock)
407
408 assert result is None
409
410
411@pytest.mark.asyncio
412async def test_get_new_releases_recommendations_no_releases(provider_mock: Mock) -> None:
413 """Test _get_new_releases_recommendations returns None when no releases."""
414 mock_releases = Mock()
415 mock_releases.new_releases = []
416
417 provider_mock.client.get_new_releases = AsyncMock(return_value=mock_releases)
418
419 result = await YandexMusicProvider._get_new_releases_recommendations(provider_mock)
420
421 assert result is None
422
423
424@pytest.mark.asyncio
425async def test_get_new_releases_recommendations_invalid_data_error(provider_mock: Mock) -> None:
426 """Test _get_new_releases_recommendations handles InvalidDataError gracefully."""
427 mock_releases = Mock()
428 mock_releases.new_releases = [123]
429
430 provider_mock.client.get_new_releases = AsyncMock(return_value=mock_releases)
431 provider_mock.client.get_albums = AsyncMock(return_value=[Mock()])
432
433 with patch(
434 "music_assistant.providers.yandex_music.provider.parse_album",
435 side_effect=InvalidDataError("Parse error"),
436 ):
437 result = await YandexMusicProvider._get_new_releases_recommendations(provider_mock)
438
439 assert result is None
440 provider_mock.logger.debug.assert_called()
441
442
443@pytest.mark.asyncio
444async def test_get_new_playlists_recommendations_success(provider_mock: Mock) -> None:
445 """Test _get_new_playlists_recommendations returns data when API provides playlists."""
446 # Mock playlist ID object
447 mock_playlist_id = Mock()
448 mock_playlist_id.uid = "user123"
449 mock_playlist_id.kind = "456"
450
451 mock_result = Mock()
452 mock_result.new_playlists = [mock_playlist_id]
453
454 provider_mock.client.get_new_playlists = AsyncMock(return_value=mock_result)
455
456 # Mock get_playlists to return playlist objects
457 mock_playlist = Mock()
458 provider_mock.client.get_playlists = AsyncMock(return_value=[mock_playlist])
459
460 # Mock parse_playlist
461 mock_parsed_playlist = Mock(spec=Playlist)
462 mock_parsed_playlist.item_id = "playlist_1"
463 mock_parsed_playlist.name = "New Playlist"
464
465 with patch(
466 "music_assistant.providers.yandex_music.provider.parse_playlist",
467 return_value=mock_parsed_playlist,
468 ):
469 result = await YandexMusicProvider._get_new_playlists_recommendations(provider_mock)
470
471 assert result is not None
472 assert isinstance(result, RecommendationFolder)
473 assert result.item_id == "new_playlists"
474 assert result.provider == provider_mock.instance_id
475 assert result.name == _RECOMMENDATION_STRINGS["new_playlists"]["name"]
476 assert result.translation_key == "new_playlists"
477 assert result.icon == "mdi-playlist-star"
478 assert len(result.items) > 0
479
480
481@pytest.mark.asyncio
482async def test_get_new_playlists_recommendations_empty(provider_mock: Mock) -> None:
483 """Test _get_new_playlists_recommendations returns None when result is empty."""
484 provider_mock.client.get_new_playlists = AsyncMock(return_value=None)
485
486 result = await YandexMusicProvider._get_new_playlists_recommendations(provider_mock)
487
488 assert result is None
489
490
491@pytest.mark.asyncio
492async def test_get_new_playlists_recommendations_no_playlists(provider_mock: Mock) -> None:
493 """Test _get_new_playlists_recommendations returns None when no playlists."""
494 mock_result = Mock()
495 mock_result.new_playlists = []
496
497 provider_mock.client.get_new_playlists = AsyncMock(return_value=mock_result)
498
499 result = await YandexMusicProvider._get_new_playlists_recommendations(provider_mock)
500
501 assert result is None
502
503
504@pytest.mark.asyncio
505async def test_get_new_playlists_recommendations_invalid_data_error(provider_mock: Mock) -> None:
506 """Test _get_new_playlists_recommendations handles InvalidDataError gracefully."""
507 mock_playlist_id = Mock()
508 mock_playlist_id.uid = "user123"
509 mock_playlist_id.kind = "456"
510
511 mock_result = Mock()
512 mock_result.new_playlists = [mock_playlist_id]
513
514 provider_mock.client.get_new_playlists = AsyncMock(return_value=mock_result)
515 provider_mock.client.get_playlists = AsyncMock(return_value=[Mock()])
516
517 with patch(
518 "music_assistant.providers.yandex_music.provider.parse_playlist",
519 side_effect=InvalidDataError("Parse error"),
520 ):
521 result = await YandexMusicProvider._get_new_playlists_recommendations(provider_mock)
522
523 assert result is None
524 provider_mock.logger.debug.assert_called()
525
526
527@pytest.mark.asyncio
528async def test_get_top_picks_recommendations_success(provider_mock: Mock) -> None:
529 """Test _get_top_picks_recommendations returns data when API provides playlists."""
530 mock_playlist = Mock()
531 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[mock_playlist])
532
533 # Mock parse_playlist
534 mock_parsed_playlist = Mock(spec=Playlist)
535 mock_parsed_playlist.item_id = "playlist_1"
536 mock_parsed_playlist.name = "Top Pick"
537
538 with patch(
539 "music_assistant.providers.yandex_music.provider.parse_playlist",
540 return_value=mock_parsed_playlist,
541 ):
542 result = await YandexMusicProvider._get_top_picks_recommendations(provider_mock)
543
544 assert result is not None
545 assert isinstance(result, RecommendationFolder)
546 assert result.item_id == "top_picks"
547 assert result.provider == provider_mock.instance_id
548 assert result.name == _RECOMMENDATION_STRINGS["top_picks"]["name"]
549 assert result.translation_key == "top_picks"
550 assert result.icon == "mdi-star"
551 assert len(result.items) > 0
552 # Verify it called with "top" tag
553 provider_mock.client.get_tag_playlists.assert_called_once_with("top")
554
555
556@pytest.mark.asyncio
557async def test_get_top_picks_recommendations_empty(provider_mock: Mock) -> None:
558 """Test _get_top_picks_recommendations returns None when API returns empty."""
559 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[])
560
561 result = await YandexMusicProvider._get_top_picks_recommendations(provider_mock)
562
563 assert result is None
564
565
566@pytest.mark.asyncio
567async def test_get_top_picks_recommendations_invalid_data_error(provider_mock: Mock) -> None:
568 """Test _get_top_picks_recommendations handles InvalidDataError gracefully."""
569 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[Mock()])
570
571 with patch(
572 "music_assistant.providers.yandex_music.provider.parse_playlist",
573 side_effect=InvalidDataError("Parse error"),
574 ):
575 result = await YandexMusicProvider._get_top_picks_recommendations(provider_mock)
576
577 assert result is None
578 provider_mock.logger.debug.assert_called()
579
580
581@pytest.mark.asyncio
582async def test_get_mood_mix_recommendations_success(provider_mock: Mock) -> None:
583 """Test _get_mood_mix_recommendations returns data with deterministic random choice."""
584 mock_playlist = Mock()
585 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[mock_playlist])
586
587 # Mock parse_playlist
588 mock_parsed_playlist = Mock(spec=Playlist)
589 mock_parsed_playlist.item_id = "playlist_1"
590 mock_parsed_playlist.name = "Chill Playlist"
591
592 # No need to patch random.choice - tag is now passed as argument
593 with patch(
594 "music_assistant.providers.yandex_music.provider.parse_playlist",
595 return_value=mock_parsed_playlist,
596 ):
597 result = await YandexMusicProvider._get_mood_mix_recommendations(provider_mock, "chill")
598
599 assert result is not None
600 assert isinstance(result, RecommendationFolder)
601 assert result.item_id == "mood_mix"
602 assert result.provider == provider_mock.instance_id
603 # Name should include the mood tag
604 assert "Chill" in result.name or "chill" in result.name.lower()
605 assert result.icon == "mdi-emoticon-outline"
606 assert len(result.items) > 0
607 # Verify it called with mood tag
608 provider_mock.client.get_tag_playlists.assert_called_once_with("chill")
609
610
611@pytest.mark.asyncio
612async def test_get_mood_mix_recommendations_empty(provider_mock: Mock) -> None:
613 """Test _get_mood_mix_recommendations returns None when API returns empty."""
614 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[])
615
616 result = await YandexMusicProvider._get_mood_mix_recommendations(provider_mock, "sad")
617
618 assert result is None
619
620
621@pytest.mark.asyncio
622async def test_get_mood_mix_recommendations_invalid_data_error(provider_mock: Mock) -> None:
623 """Test _get_mood_mix_recommendations handles InvalidDataError gracefully."""
624 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[Mock()])
625
626 with patch(
627 "music_assistant.providers.yandex_music.provider.parse_playlist",
628 side_effect=InvalidDataError("Parse error"),
629 ):
630 result = await YandexMusicProvider._get_mood_mix_recommendations(provider_mock, "romantic")
631
632 assert result is None
633 provider_mock.logger.debug.assert_called()
634
635
636@pytest.mark.asyncio
637async def test_get_activity_mix_recommendations_success(provider_mock: Mock) -> None:
638 """Test _get_activity_mix_recommendations returns data with deterministic random choice."""
639 mock_playlist = Mock()
640 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[mock_playlist])
641
642 # Mock parse_playlist
643 mock_parsed_playlist = Mock(spec=Playlist)
644 mock_parsed_playlist.item_id = "playlist_1"
645 mock_parsed_playlist.name = "Workout Playlist"
646
647 # No need to patch random.choice - tag is now passed as argument
648 with patch(
649 "music_assistant.providers.yandex_music.provider.parse_playlist",
650 return_value=mock_parsed_playlist,
651 ):
652 result = await YandexMusicProvider._get_activity_mix_recommendations(
653 provider_mock, "workout"
654 )
655
656 assert result is not None
657 assert isinstance(result, RecommendationFolder)
658 assert result.item_id == "activity_mix"
659 assert result.provider == provider_mock.instance_id
660 # Name should include the activity tag
661 assert "Workout" in result.name or "workout" in result.name.lower()
662 assert result.icon == "mdi-run"
663 assert len(result.items) > 0
664 # Verify it called with activity tag
665 provider_mock.client.get_tag_playlists.assert_called_once_with("workout")
666
667
668@pytest.mark.asyncio
669async def test_get_activity_mix_recommendations_empty(provider_mock: Mock) -> None:
670 """Test _get_activity_mix_recommendations returns None when API returns empty."""
671 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[])
672
673 result = await YandexMusicProvider._get_activity_mix_recommendations(provider_mock, "focus")
674
675 assert result is None
676
677
678@pytest.mark.asyncio
679async def test_get_activity_mix_recommendations_invalid_data_error(provider_mock: Mock) -> None:
680 """Test _get_activity_mix_recommendations handles InvalidDataError gracefully."""
681 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[Mock()])
682
683 with patch(
684 "music_assistant.providers.yandex_music.provider.parse_playlist",
685 side_effect=InvalidDataError("Parse error"),
686 ):
687 result = await YandexMusicProvider._get_activity_mix_recommendations(
688 provider_mock, "morning"
689 )
690
691 assert result is None
692 provider_mock.logger.debug.assert_called()
693
694
695@pytest.mark.asyncio
696async def test_get_seasonal_mix_recommendations_winter(provider_mock: Mock) -> None:
697 """Test _get_seasonal_mix_recommendations returns winter playlists in January."""
698 mock_playlist = Mock()
699 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[mock_playlist])
700
701 # Mock parse_playlist
702 mock_parsed_playlist = Mock(spec=Playlist)
703 mock_parsed_playlist.item_id = "playlist_1"
704 mock_parsed_playlist.name = "Winter Playlist"
705
706 # Patch datetime to return January (month 1)
707 mock_utc = Mock()
708 mock_utc.return_value.month = 1
709
710 with (
711 patch("music_assistant.providers.yandex_music.provider.utc", mock_utc),
712 patch(
713 "music_assistant.providers.yandex_music.provider.parse_playlist",
714 return_value=mock_parsed_playlist,
715 ),
716 ):
717 result = await YandexMusicProvider._get_seasonal_mix_recommendations(provider_mock)
718
719 assert result is not None
720 assert isinstance(result, RecommendationFolder)
721 assert result.item_id == "seasonal_mix"
722 assert result.provider == provider_mock.instance_id
723 # Name should include winter
724 assert "Winter" in result.name or "winter" in result.name.lower()
725 assert result.icon == "mdi-weather-sunny"
726 assert len(result.items) > 0
727 # Verify it called with winter tag
728 provider_mock.client.get_tag_playlists.assert_called_once_with("winter")
729
730
731@pytest.mark.asyncio
732async def test_get_seasonal_mix_recommendations_summer(provider_mock: Mock) -> None:
733 """Test _get_seasonal_mix_recommendations returns summer playlists in July."""
734 mock_playlist = Mock()
735 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[mock_playlist])
736
737 mock_parsed_playlist = Mock(spec=Playlist)
738 mock_parsed_playlist.item_id = "playlist_1"
739 mock_parsed_playlist.name = "Summer Playlist"
740
741 # Patch datetime to return July (month 7)
742 mock_utc = Mock()
743 mock_utc.return_value.month = 7
744
745 with (
746 patch("music_assistant.providers.yandex_music.provider.utc", mock_utc),
747 patch(
748 "music_assistant.providers.yandex_music.provider.parse_playlist",
749 return_value=mock_parsed_playlist,
750 ),
751 ):
752 result = await YandexMusicProvider._get_seasonal_mix_recommendations(provider_mock)
753
754 assert result is not None
755 # Verify it called with summer tag
756 provider_mock.client.get_tag_playlists.assert_called_once_with("summer")
757
758
759@pytest.mark.asyncio
760async def test_get_seasonal_mix_recommendations_spring_fallback(provider_mock: Mock) -> None:
761 """Spring with no playlists falls back to autumn (Yandex coverage gap)."""
762 mock_playlist = Mock()
763 # First call (spring) â empty; second call (autumn) â has playlists
764 provider_mock.client.get_tag_playlists = AsyncMock(
765 side_effect=[[], [mock_playlist]],
766 )
767
768 mock_parsed_playlist = Mock(spec=Playlist)
769 mock_parsed_playlist.item_id = "playlist_1"
770 mock_parsed_playlist.name = "Autumn Playlist"
771
772 # Patch datetime to return March (month 3 - spring)
773 mock_utc = Mock()
774 mock_utc.return_value.month = 3
775
776 with (
777 patch("music_assistant.providers.yandex_music.provider.utc", mock_utc),
778 patch(
779 "music_assistant.providers.yandex_music.provider.parse_playlist",
780 return_value=mock_parsed_playlist,
781 ),
782 ):
783 result = await YandexMusicProvider._get_seasonal_mix_recommendations(provider_mock)
784
785 assert result is not None
786 # Verify call sequence: spring first, autumn fallback after empty result
787 assert provider_mock.client.get_tag_playlists.await_args_list[0].args == ("spring",)
788 assert provider_mock.client.get_tag_playlists.await_args_list[1].args == ("autumn",)
789
790
791@pytest.mark.asyncio
792async def test_get_seasonal_mix_recommendations_empty(provider_mock: Mock) -> None:
793 """Test _get_seasonal_mix_recommendations returns None when API returns empty."""
794 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[])
795
796 mock_utc = Mock()
797 mock_utc.return_value.month = 6
798
799 with patch("music_assistant.providers.yandex_music.provider.utc", mock_utc):
800 result = await YandexMusicProvider._get_seasonal_mix_recommendations(provider_mock)
801
802 assert result is None
803
804
805@pytest.mark.asyncio
806async def test_get_seasonal_mix_recommendations_invalid_data_error(provider_mock: Mock) -> None:
807 """Test _get_seasonal_mix_recommendations handles InvalidDataError gracefully."""
808 provider_mock.client.get_tag_playlists = AsyncMock(return_value=[Mock()])
809
810 mock_utc = Mock()
811 mock_utc.return_value.month = 9
812
813 with (
814 patch("music_assistant.providers.yandex_music.provider.utc", mock_utc),
815 patch(
816 "music_assistant.providers.yandex_music.provider.parse_playlist",
817 side_effect=InvalidDataError("Parse error"),
818 ),
819 ):
820 result = await YandexMusicProvider._get_seasonal_mix_recommendations(provider_mock)
821
822 assert result is None
823 provider_mock.logger.debug.assert_called()
824
825
826@pytest.mark.asyncio
827async def test_get_recommendations_returns_static_rows_without_backend_calls(
828 provider_mock: Mock,
829) -> None:
830 """Recommendation discovery returns ordered empty row descriptors."""
831 rows = await YandexMusicProvider.get_recommendations(provider_mock)
832
833 assert [row.item_id for row in rows] == ROW_IDS
834 assert all(not row.items for row in rows)
835 assert provider_mock.client.mock_calls == []
836
837
838@pytest.mark.asyncio
839async def test_get_recommendation_items_chart_triggers_only_chart_fetch(
840 provider_mock: Mock,
841) -> None:
842 """Loading Chart calls only the Chart row helper."""
843 track = Mock(spec=Track)
844 folder = RecommendationFolder(
845 item_id="chart",
846 provider=provider_mock.instance_id,
847 name="Chart",
848 items=UniqueList([track]),
849 )
850 provider_mock._get_chart_recommendations = AsyncMock(return_value=folder)
851
852 items = await YandexMusicProvider.get_recommendation_items(provider_mock, "chart")
853
854 provider_mock._get_chart_recommendations.assert_awaited_once()
855 assert list(items) == [track]
856 assert provider_mock.client.mock_calls == []
857
858
859@pytest.mark.asyncio
860async def test_get_recommendation_items_uses_deterministic_tag_outside_cached_helper(
861 provider_mock: Mock,
862) -> None:
863 """Mood row chooses its rotating tag before calling the cached helper."""
864 playlist = Mock(spec=Playlist)
865 folder = RecommendationFolder(
866 item_id="mood_mix",
867 provider=provider_mock.instance_id,
868 name="Mood Mix",
869 items=UniqueList([playlist]),
870 )
871 provider_mock._get_valid_tags_for_category = AsyncMock(return_value=["focus", "chill"])
872 provider_mock._rotating_row_tag = Mock(return_value="focus")
873 provider_mock._get_mood_mix_recommendations = AsyncMock(return_value=folder)
874
875 items = await YandexMusicProvider.get_recommendation_items(provider_mock, "mood_mix")
876
877 provider_mock._get_valid_tags_for_category.assert_awaited_once_with("mood")
878 provider_mock._rotating_row_tag.assert_called_once_with("mood", ["focus", "chill"])
879 provider_mock._get_mood_mix_recommendations.assert_awaited_once_with("focus")
880 assert list(items) == [playlist]
881
882
883@pytest.mark.asyncio
884async def test_get_recommendation_items_empty_or_unknown_returns_empty(
885 provider_mock: Mock,
886) -> None:
887 """Missing tags, empty helpers, and unknown row IDs yield an empty list."""
888 provider_mock._get_valid_tags_for_category = AsyncMock(return_value=[])
889 provider_mock._get_feed_recommendations = AsyncMock(return_value=None)
890
891 assert not await YandexMusicProvider.get_recommendation_items(provider_mock, "mood_mix")
892 assert not await YandexMusicProvider.get_recommendation_items(provider_mock, "feed")
893 assert not await YandexMusicProvider.get_recommendation_items(provider_mock, "unknown")
894
895
896@pytest.mark.asyncio
897async def test_rotating_row_tag_subtitle_from_warm_cache(provider_mock: Mock) -> None:
898 """A warm tag cache produces a localized rotating-row subtitle."""
899 provider_mock.mass.cache.get_with_freshness = AsyncMock(
900 return_value=(["chill", "focus"], True, True)
901 )
902 provider_mock._rotating_row_tag = YandexMusicProvider._rotating_row_tag.__get__(
903 provider_mock, YandexMusicProvider
904 )
905
906 label = await YandexMusicProvider._rotating_row_tag_subtitle(provider_mock, "mood")
907
908 assert label in {"Chill", "Focus"}
909 provider_mock.client.get_landing_tags.assert_not_awaited()
910
911
912@pytest.mark.asyncio
913async def test_rotating_row_tag_subtitle_cold_cache_returns_none(provider_mock: Mock) -> None:
914 """Descriptor discovery avoids I/O when the tag cache is cold."""
915 provider_mock.mass.cache.get_with_freshness = AsyncMock(return_value=(None, False, False))
916
917 label = await YandexMusicProvider._rotating_row_tag_subtitle(provider_mock, "activity")
918
919 assert label is None
920 provider_mock.client.get_landing_tags.assert_not_awaited()
921
922
923def test_rotating_row_tag_is_deterministic_and_category_scoped(provider_mock: Mock) -> None:
924 """Hourly tag selection is stable and incorporates the category."""
925 fixed_now = datetime(2026, 7, 30, 12, tzinfo=UTC)
926 tags = ["chill", "focus", "happy", "calm"]
927
928 with (
929 patch(
930 "music_assistant.providers.yandex_music.provider.utc",
931 return_value=fixed_now,
932 ),
933 patch(
934 "music_assistant.providers.yandex_music.provider.hashlib.sha256",
935 wraps=hashlib.sha256,
936 ) as sha256,
937 ):
938 mood_first = YandexMusicProvider._rotating_row_tag(provider_mock, "mood", tags)
939 mood_second = YandexMusicProvider._rotating_row_tag(provider_mock, "mood", tags)
940 activity = YandexMusicProvider._rotating_row_tag(provider_mock, "activity", tags)
941
942 assert mood_first == mood_second
943 assert mood_first in tags
944 assert activity in tags
945 assert sha256.call_args_list[0].args == sha256.call_args_list[1].args
946 assert sha256.call_args_list[0].args != sha256.call_args_list[2].args
947
948
949@pytest.mark.asyncio
950async def test_get_similar_artists_returns_parsed(provider_mock: Mock) -> None:
951 """get_similar_artists parses each artist from the underlying client."""
952 yandex_artists = [Mock(), Mock(), Mock()]
953 provider_mock.client.get_similar_artists = AsyncMock(return_value=yandex_artists)
954
955 parsed = [_media_item_mock(Artist) for _ in yandex_artists]
956 with patch(
957 "music_assistant.providers.yandex_music.provider.parse_artist",
958 side_effect=parsed,
959 ):
960 result = await YandexMusicProvider.get_similar_artists(provider_mock, "42", limit=10)
961
962 provider_mock.client.get_similar_artists.assert_awaited_once_with("42", limit=10)
963 assert result == parsed
964
965
966@pytest.mark.asyncio
967async def test_get_similar_artists_skips_invalid(provider_mock: Mock) -> None:
968 """get_similar_artists skips artists that fail to parse."""
969 yandex_artists = [Mock(), Mock()]
970 provider_mock.client.get_similar_artists = AsyncMock(return_value=yandex_artists)
971
972 parsed_ok = _media_item_mock(Artist)
973 with patch(
974 "music_assistant.providers.yandex_music.provider.parse_artist",
975 side_effect=[InvalidDataError("missing id"), parsed_ok],
976 ):
977 result = await YandexMusicProvider.get_similar_artists(provider_mock, "99")
978
979 assert result == [parsed_ok]
980
981
982@pytest.mark.asyncio
983async def test_get_similar_artists_empty(provider_mock: Mock) -> None:
984 """get_similar_artists returns [] when client returns no artists."""
985 provider_mock.client.get_similar_artists = AsyncMock(return_value=[])
986
987 result = await YandexMusicProvider.get_similar_artists(provider_mock, "42")
988
989 assert result == []
990
991
992# -- M18: integration coverage with the real parser path ---------------------
993#
994# Most tests in this file mock ``_parse_my_wave_track`` / ``parse_playlist``
995# directly. That keeps them fast but leaves the orchestrator-parser seam
996# untested: a regression where the orchestrator builds malformed inputs for
997# the parser, or where the parser returns shape-incompatible output, would
998# slip through. The tests below exercise the real parser end-to-end with a
999# minimal Yandex track fixture.
1000
1001
1002def _real_yandex_track() -> Any:
1003 """Return a minimally-shaped Yandex track for the real ``parse_track`` to consume."""
1004 fixture = pathlib.Path(__file__).parent / "fixtures" / "tracks" / "with_artist_and_album.json"
1005 return YandexTrack.de_json(json.loads(fixture.read_text()), DE_JSON_CLIENT)
1006
1007
1008@pytest.mark.asyncio
1009async def test_get_my_wave_recommendations_with_real_parser(provider_mock: Mock) -> None:
1010 """
1011 End-to-end: real ``_parse_my_wave_track`` against a real Yandex track fixture.
1012
1013 Mocking ``_parse_my_wave_track`` directly (as the success/duplicate/error
1014 tests above do) cannot catch a regression where ``parse_track`` returns a
1015 malformed Track, where the composite ``item_id`` is not stamped onto the
1016 provider_mappings, or where the wave-state lock is not respected. This
1017 test binds the real method and asserts on the produced ``Track``.
1018 """
1019 yt = _real_yandex_track()
1020 assert yt is not None
1021
1022 # Bind the real method so it actually calls parse_track + composite item_id logic.
1023 real_parse = YandexMusicProvider._parse_my_wave_track.__get__(provider_mock)
1024 provider_mock._parse_my_wave_track = real_parse
1025 provider_mock._fetch_rotor_session_batch = AsyncMock(return_value=([yt], "batch_x"))
1026 # Wave-state lock is held during the loop; supply a fresh state.
1027 provider_mock._get_wave_state = Mock(return_value=_WaveState())
1028 # get_item_mapping is consumed by parse_track for provider_mappings; mock the spec
1029 # method to return a minimal ItemMapping-like stub.
1030 provider_mock.get_item_mapping = Mock(
1031 side_effect=lambda mt, key, name: ItemMapping(
1032 media_type=MediaType(mt) if isinstance(mt, str) else mt,
1033 item_id=key,
1034 provider=provider_mock.instance_id,
1035 name=name,
1036 )
1037 )
1038
1039 result = await YandexMusicProvider._get_my_wave_recommendations(provider_mock)
1040
1041 assert result is not None
1042 assert isinstance(result, RecommendationFolder)
1043 assert len(result.items) == 1
1044 track = result.items[0]
1045 assert isinstance(track, Track)
1046 # The orchestrator-parser contract: composite item_id, real Yandex track id.
1047 assert track.item_id == f"500{RADIO_TRACK_ID_SEP}{ROTOR_STATION_MY_WAVE}"
1048 assert track.name == "Track With Album"
1049 # provider_mappings carry the composite id too â not the bare track id.
1050 assert all(pm.item_id == track.item_id for pm in track.provider_mappings)
1051