/
/
/
1"""Test KION Music two-method recommendations: static rows + per-row item fetches."""
2
3from __future__ import annotations
4
5import json
6import pathlib
7from datetime import UTC, datetime
8from types import SimpleNamespace
9from typing import Any, cast
10from unittest.mock import AsyncMock, Mock
11
12import pytest
13from yandex_music import Album as YandexAlbum
14from yandex_music import Playlist as YandexPlaylist
15from yandex_music import Track as YandexTrack
16
17from music_assistant.providers.kion_music import SUPPORTED_FEATURES
18from music_assistant.providers.kion_music.constants import MY_WAVE_PLAYLIST_ID
19from music_assistant.providers.kion_music.provider import KionMusicProvider
20from tests.common import use_real_create_task
21
22from .conftest import DE_JSON_CLIENT
23
24FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures"
25
26# One client method (or method set) backs each recommendation row.
27CLIENT_METHODS = (
28 "get_my_wave_tracks",
29 "get_feed",
30 "get_chart",
31 "get_new_releases",
32 "get_albums",
33 "get_new_playlists",
34 "get_playlists",
35 "get_tag_playlists",
36 "get_landing_tags",
37)
38
39# Expected row ids in the order get_recommendations() returns them.
40ROW_IDS = [
41 MY_WAVE_PLAYLIST_ID,
42 "feed",
43 "chart",
44 "new_releases",
45 "new_playlists",
46 "top_picks",
47 "mood_mix",
48 "activity_mix",
49 "seasonal_mix",
50]
51
52
53def _load_fixture(relpath: str) -> dict[str, Any]:
54 """Load a JSON fixture relative to the fixtures dir."""
55 with open(FIXTURES_DIR / relpath) as f:
56 return cast("dict[str, Any]", json.load(f))
57
58
59def _make_client_mock() -> Mock:
60 """Build a client mock whose backend methods return parseable canned data."""
61 track = YandexTrack.de_json(_load_fixture("tracks/with_artist_and_album.json"), DE_JSON_CLIENT)
62 album = YandexAlbum.de_json(_load_fixture("albums/minimal.json"), DE_JSON_CLIENT)
63 playlist = YandexPlaylist.de_json(_load_fixture("playlists/minimal.json"), DE_JSON_CLIENT)
64 client = Mock()
65 client.user_id = 12345
66 client.get_my_wave_tracks = AsyncMock(return_value=([track], None))
67 client.get_feed = AsyncMock(
68 return_value=SimpleNamespace(
69 generated_playlists=[SimpleNamespace(data=playlist, ready=True)]
70 )
71 )
72 client.get_chart = AsyncMock(
73 return_value=SimpleNamespace(chart=SimpleNamespace(tracks=[SimpleNamespace(track=track)]))
74 )
75 client.get_new_releases = AsyncMock(return_value=SimpleNamespace(new_releases=[300]))
76 client.get_albums = AsyncMock(return_value=[album])
77 client.get_new_playlists = AsyncMock(
78 return_value=SimpleNamespace(new_playlists=[SimpleNamespace(uid=12345, kind=3)])
79 )
80 client.get_playlists = AsyncMock(return_value=[playlist])
81 client.get_tag_playlists = AsyncMock(return_value=[playlist])
82 client.get_landing_tags = AsyncMock(return_value=[])
83 return client
84
85
86def _install_cache_mocks(provider: KionMusicProvider) -> None:
87 """Make the @use_cache decorator treat every call as a cache miss."""
88 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
89 return_value=(None, False, False)
90 )
91 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
92
93
94def _awaited_methods(client: Mock) -> set[str]:
95 """Return the names of backend client methods that were actually awaited."""
96 return {name for name in CLIENT_METHODS if getattr(client, name).await_count}
97
98
99@pytest.fixture
100def provider() -> KionMusicProvider:
101 """Create a real KionMusicProvider with mocked dependencies."""
102 mass = Mock()
103 mass.metadata.locale = "en_US"
104 mass.translations.get_translation = Mock(return_value=None)
105 # default: every cache lookup is a miss (tests override to simulate warm entries)
106 mass.cache.get_with_freshness = AsyncMock(return_value=(None, False, False))
107 mass.cache.set = AsyncMock()
108 use_real_create_task(mass)
109 manifest = Mock()
110 manifest.domain = "kion_music"
111 config = Mock()
112 config.instance_id = "kion_music--test123"
113 config.name = "KION Music Test"
114 config.enabled = True
115 config.get_value.side_effect = lambda key, default=None: {
116 "log_level": "GLOBAL",
117 }.get(key, default)
118 provider = KionMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
119 provider._client = _make_client_mock()
120 return provider
121
122
123@pytest.mark.asyncio
124async def test_get_recommendations_returns_static_rows_without_backend_calls(
125 provider: KionMusicProvider,
126) -> None:
127 """get_recommendations() returns all nine row descriptors with zero backend I/O."""
128 client = cast("Mock", provider.client)
129
130 result = await provider.get_recommendations()
131
132 assert [f.item_id for f in result] == ROW_IDS
133 assert _awaited_methods(client) == set()
134 assert all(not f.items for f in result)
135 # Mood/Activity row titles are static; the rotating tag only shows as subtitle
136 # once the tag-list cache is warm (cold here, so no subtitle).
137 by_id = {f.item_id: f for f in result}
138 assert by_id["mood_mix"].name == "Mood Mix"
139 assert by_id["mood_mix"].translation_key == "mood_mix"
140 assert by_id["activity_mix"].name == "Activity Mix"
141 assert by_id["activity_mix"].translation_key == "activity_mix"
142 assert by_id["seasonal_mix"].name.startswith("Seasonal: ")
143
144
145@pytest.mark.asyncio
146@pytest.mark.parametrize(
147 ("item_id", "expected_backend_calls"),
148 [
149 (MY_WAVE_PLAYLIST_ID, {"get_my_wave_tracks"}),
150 ("feed", {"get_feed"}),
151 ("chart", {"get_chart"}),
152 ("new_releases", {"get_new_releases", "get_albums"}),
153 ("new_playlists", {"get_new_playlists", "get_playlists"}),
154 ("top_picks", {"get_tag_playlists"}),
155 ("mood_mix", {"get_landing_tags", "get_tag_playlists"}),
156 ("activity_mix", {"get_landing_tags", "get_tag_playlists"}),
157 ("seasonal_mix", {"get_tag_playlists"}),
158 ],
159)
160async def test_get_recommendation_items_triggers_only_that_rows_fetches(
161 provider: KionMusicProvider, item_id: str, expected_backend_calls: set[str]
162) -> None:
163 """get_recommendation_items(row) issues only that row's backend fetches and returns items."""
164 _install_cache_mocks(provider)
165 client = cast("Mock", provider.client)
166
167 result = await provider.get_recommendation_items(item_id)
168
169 assert _awaited_methods(client) == expected_backend_calls
170 assert len(result) > 0
171
172
173@pytest.mark.asyncio
174async def test_get_recommendation_items_unknown_id_returns_empty(
175 provider: KionMusicProvider,
176) -> None:
177 """An unknown row item_id returns an empty list without any backend calls."""
178 _install_cache_mocks(provider)
179 client = cast("Mock", provider.client)
180
181 result = await provider.get_recommendation_items("no_such_row")
182
183 assert list(result) == []
184 assert _awaited_methods(client) == set()
185
186
187def _install_tag_cache(provider: KionMusicProvider, tags_by_category: dict[str, list[str]]) -> None:
188 """Serve the validated-tag-list cache entries as warm hits, everything else as a miss."""
189
190 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
191 for category, tags in tags_by_category.items():
192 if key == f"_get_valid_tags_for_category.{category}":
193 return tags, True, True
194 return None, False, False
195
196 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
197 side_effect=_cache_get
198 )
199 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
200
201
202@pytest.mark.asyncio
203async def test_rows_subtitle_matches_served_items_tag(
204 provider: KionMusicProvider, monkeypatch: pytest.MonkeyPatch
205) -> None:
206 """With a warm tag cache the rows subtitle and the items call resolve the same tag."""
207 # freeze the clock so the hourly tag bucket cannot flip mid-test
208 monkeypatch.setattr(
209 "music_assistant.providers.kion_music.provider.utc",
210 lambda: datetime(2026, 7, 24, 12, 30, tzinfo=UTC),
211 )
212 mood_tags = ["chill", "focus"]
213 _install_tag_cache(provider, {"mood": mood_tags, "activity": ["workout"]})
214 client = cast("Mock", provider.client)
215
216 result = await provider.get_recommendations()
217
218 by_id = {f.item_id: f for f in result}
219 mood_tag = provider._rotating_row_tag("mood", mood_tags)
220 assert by_id["mood_mix"].subtitle == mood_tag.title()
221 assert by_id["activity_mix"].subtitle == "Workout"
222 # deterministic: a second rows call yields the same subtitle
223 again = {f.item_id: f for f in await provider.get_recommendations()}
224 assert again["mood_mix"].subtitle == by_id["mood_mix"].subtitle
225 # the tag came from the cache: rows still issue zero backend calls
226 assert _awaited_methods(client) == set()
227
228 await provider.get_recommendation_items("mood_mix")
229
230 # the items call independently derived the same tag from the cached list
231 client.get_tag_playlists.assert_any_await(mood_tag)
232 assert "get_landing_tags" not in _awaited_methods(client)
233
234
235@pytest.mark.asyncio
236async def test_rows_without_cached_tags_have_no_subtitle(provider: KionMusicProvider) -> None:
237 """With a cold tag cache the mood/activity rows have no subtitle."""
238 _install_cache_mocks(provider)
239
240 result = await provider.get_recommendations()
241
242 by_id = {f.item_id: f for f in result}
243 assert by_id["mood_mix"].subtitle is None
244 assert by_id["activity_mix"].subtitle is None
245