/
/
/
1"""Test Apple Music two-method recommendations served from the cached bulk payload."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING, Any, cast
7from unittest.mock import AsyncMock, Mock
8
9import pytest
10from music_assistant_models.media_items import Playlist, UniqueList
11
12from music_assistant.providers.apple_music.provider import AppleMusicProvider
13
14if TYPE_CHECKING:
15 from collections.abc import AsyncGenerator, Coroutine
16
17INSTANCE_ID = "apple_music--test1"
18
19
20def _station(station_id: str, name: str, is_live: bool = False) -> dict[str, Any]:
21 """Build a station content object for a me/recommendations response."""
22 return {"type": "stations", "id": station_id, "attributes": {"name": name, "isLive": is_live}}
23
24
25def _recommendations_response(sections: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
26 """Build a me/recommendations response from section title -> station objects."""
27 return {
28 "data": [
29 {
30 "id": f"rec{index}",
31 "attributes": {"title": {"stringForDisplay": title}},
32 "relationships": {"contents": {"data": stations}},
33 }
34 for index, (title, stations) in enumerate(sections.items())
35 ]
36 }
37
38
39def _default_response() -> dict[str, Any]:
40 """Build the default two-section payload response used by most tests."""
41 return _recommendations_response(
42 {
43 "Made for You": [_station("ra.111", "My Station One")],
44 "Stations for You": [
45 _station("ra.222", "My Station Two"),
46 _station("ra.333", "Live Station", is_live=True),
47 ],
48 }
49 )
50
51
52@pytest.fixture
53async def provider() -> AsyncGenerator[AppleMusicProvider]:
54 """Create a real AppleMusicProvider with mocked mass (dict-backed cache) and API client."""
55 cache_store: dict[str, Any] = {}
56 background_tasks: list[asyncio.Future[Any]] = []
57
58 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
59 if key in cache_store:
60 return cache_store[key], True, True
61 return None, False, False
62
63 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
64 cache_store[key] = data
65
66 def _create_task(target: Coroutine[Any, Any, Any], *_args: Any, **_kwargs: Any) -> Any:
67 task: asyncio.Future[Any] = asyncio.ensure_future(target)
68 background_tasks.append(task)
69 return task
70
71 mass = Mock()
72 mass.cache.get_with_freshness = AsyncMock(side_effect=_cache_get)
73 mass.cache.set = AsyncMock(side_effect=_cache_set)
74 mass.create_task = Mock(side_effect=_create_task)
75 manifest = Mock()
76 manifest.domain = "apple_music"
77 config = Mock()
78 config.instance_id = INSTANCE_ID
79 config.name = "Apple Music Test"
80 config.get_value.side_effect = lambda key, default=None: {"log_level": "GLOBAL"}.get(
81 key, default
82 )
83 provider = AppleMusicProvider(mass, manifest, config)
84 provider.api_client.get_data = AsyncMock(return_value=_default_response()) # type: ignore[method-assign]
85 yield provider
86 await asyncio.gather(*background_tasks, return_exceptions=True)
87
88
89@pytest.mark.asyncio
90async def test_get_recommendations_returns_rows_without_items(
91 provider: AppleMusicProvider,
92) -> None:
93 """get_recommendations() returns the payload's section rows, stripped of items."""
94 rows = await provider.get_recommendations()
95
96 assert [row.item_id for row in rows] == ["made_for_you", "stations_for_you"]
97 assert [row.name for row in rows] == ["Made for You", "Stations for You"]
98 assert all(row.provider == INSTANCE_ID for row in rows)
99 assert all(not row.items for row in rows)
100
101
102@pytest.mark.asyncio
103async def test_rows_and_items_share_one_backend_fetch(provider: AppleMusicProvider) -> None:
104 """The rows call and both per-row items calls are served from one me/recommendations fetch."""
105 api_get_data = cast("AsyncMock", provider.api_client.get_data)
106
107 await provider.get_recommendations()
108 # let the background cache-store task complete before the next calls
109 await asyncio.sleep(0)
110 items_one = await provider.get_recommendation_items("made_for_you")
111 items_two = await provider.get_recommendation_items("stations_for_you")
112
113 api_get_data.assert_awaited_once()
114 assert [item.item_id for item in items_one] == ["ra.111"]
115 assert all(isinstance(item, Playlist) for item in items_one)
116 # the live station in the payload is skipped
117 assert [item.item_id for item in items_two] == ["ra.222"]
118
119
120@pytest.mark.asyncio
121async def test_warm_cache_hit_serves_items_without_backend_fetch(
122 provider: AppleMusicProvider,
123) -> None:
124 """A second items call is served from the stored cache entry, not a new backend fetch."""
125 api_get_data = cast("AsyncMock", provider.api_client.get_data)
126
127 items_first = await provider.get_recommendation_items("made_for_you")
128 # let the background cache-store task complete before the next call
129 await asyncio.sleep(0)
130 items_second = await provider.get_recommendation_items("made_for_you")
131
132 api_get_data.assert_awaited_once()
133 assert [item.item_id for item in items_first] == ["ra.111"]
134 assert [item.item_id for item in items_second] == ["ra.111"]
135
136
137@pytest.mark.asyncio
138async def test_get_recommendation_items_unknown_id_returns_empty(
139 provider: AppleMusicProvider,
140) -> None:
141 """An item_id not present in the payload yields an empty UniqueList."""
142 result = await provider.get_recommendation_items("no_such_row")
143
144 assert isinstance(result, UniqueList)
145 assert list(result) == []
146
147
148@pytest.mark.asyncio
149async def test_browse_stations_returns_all_payload_stations(
150 provider: AppleMusicProvider,
151) -> None:
152 """browse_stations flattens the stations of every payload folder."""
153 stations = await provider.recommendation_manager.browse_stations()
154
155 assert [station.item_id for station in stations] == ["ra.111", "ra.222"]
156
157
158@pytest.mark.asyncio
159async def test_resolve_station_id_returns_rotated_id(provider: AppleMusicProvider) -> None:
160 """resolve_station_id maps a stale station id to the fresh id via the station name."""
161 api_get_data = cast("AsyncMock", provider.api_client.get_data)
162 rotated = _recommendations_response({"Made for You": [_station("ra.999", "My Station One")]})
163 api_get_data.side_effect = [_default_response(), rotated]
164
165 result = await provider.recommendation_manager.resolve_station_id("ra.111")
166
167 assert result == "ra.999"
168 # one payload fetch to populate the maps, one fresh fetch to learn the current id
169 assert api_get_data.await_count == 2
170
171
172@pytest.mark.asyncio
173async def test_resolve_station_id_stores_fresh_payload_for_rows_and_items(
174 provider: AppleMusicProvider,
175) -> None:
176 """The forced-fresh fetch of resolve_station_id is stored back to the payload cache."""
177 api_get_data = cast("AsyncMock", provider.api_client.get_data)
178 rotated = _recommendations_response({"Made for You": [_station("ra.999", "My Station One")]})
179 api_get_data.side_effect = [_default_response(), rotated]
180
181 result = await provider.recommendation_manager.resolve_station_id("ra.111")
182 items = await provider.get_recommendation_items("made_for_you")
183
184 assert result == "ra.999"
185 # rows/items serve the refreshed (rotated) payload from the cache, without a new fetch
186 assert [item.item_id for item in items] == ["ra.999"]
187 assert api_get_data.await_count == 2
188
189
190@pytest.mark.asyncio
191async def test_resolve_station_id_after_restart_with_cached_payload(
192 provider: AppleMusicProvider,
193) -> None:
194 """After a restart the station maps are rebuilt from the cache-served payload folders."""
195 api_get_data = cast("AsyncMock", provider.api_client.get_data)
196 rotated = _recommendations_response({"Made for You": [_station("ra.999", "My Station One")]})
197 api_get_data.side_effect = [_default_response(), rotated]
198 # warm the persistent cache, then simulate a process restart: the in-memory
199 # station maps are gone while the cached payload entry survives
200 await provider.get_recommendations()
201 await asyncio.sleep(0)
202 manager = provider.recommendation_manager
203 manager._station_id_to_name.clear()
204 manager._station_name_to_id.clear()
205
206 result = await manager.resolve_station_id("ra.111")
207
208 assert result == "ra.999"
209 # the cache-served payload populated the maps without a backend fetch;
210 # only the initial warm-up and the forced-fresh rotation fetch hit the API
211 assert api_get_data.await_count == 2
212
213
214@pytest.mark.asyncio
215async def test_resolve_station_id_unknown_returns_none(provider: AppleMusicProvider) -> None:
216 """resolve_station_id returns None for an id absent from the payload, without a fresh fetch."""
217 api_get_data = cast("AsyncMock", provider.api_client.get_data)
218
219 result = await provider.recommendation_manager.resolve_station_id("ra.unknown")
220
221 assert result is None
222 api_get_data.assert_awaited_once()
223