/
/
/
1"""Test the MusicMe two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock
8
9import pytest
10from music_assistant_models.media_items import Radio, RecommendationFolder, UniqueList
11
12from music_assistant.providers.musicme.provider import MusicMeProvider
13from tests.common import use_real_create_task
14
15HOME_DATA = {"results": {"items": [{"id": 1, "name": "Home Radio"}]}}
16NEWS_DATA = {"results": {"albums": [{"barcode": "b1", "name": "Album", "streamable": 2}]}}
17TOPS_DATA = {"results": {"artists": [{"id": 2, "name": "Artist"}]}}
18RADIOS_DATA = {"results": {"theme-airplays": [{"id": 3, "name": "Radio"}]}}
19
20
21def _stub_api_get(provider: MusicMeProvider) -> AsyncMock:
22 """Attach an _api_get stub that returns canned data keyed by endpoint prefix."""
23
24 async def _fake(endpoint: str) -> dict[str, Any] | None:
25 if endpoint.startswith("/home"):
26 return HOME_DATA
27 if endpoint.startswith("/news"):
28 return NEWS_DATA
29 if endpoint.startswith("/tops"):
30 return TOPS_DATA
31 if endpoint.startswith("/radios"):
32 return RADIOS_DATA
33 return None
34
35 mock = AsyncMock(side_effect=_fake)
36 provider._api_get = mock # type: ignore[method-assign]
37 return mock
38
39
40def _install_cache_mocks(provider: MusicMeProvider) -> None:
41 """Make the @use_cache decorator treat every call as a cache miss."""
42 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
43 return_value=(None, False, False)
44 )
45 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
46 use_real_create_task(provider.mass)
47
48
49@pytest.mark.asyncio
50async def test_get_recommendations_static_rows_zero_backend_calls(
51 provider: MusicMeProvider,
52) -> None:
53 """get_recommendations returns the four static rows, without items and without I/O."""
54 api_mock = _stub_api_get(provider)
55
56 rows = await provider.get_recommendations()
57
58 api_mock.assert_not_awaited()
59 assert [row.item_id for row in rows] == [
60 f"{provider.instance_id}_home",
61 f"{provider.instance_id}_news",
62 f"{provider.instance_id}_tops",
63 f"{provider.instance_id}_radios",
64 ]
65 assert [row.name for row in rows] == ["Featured", "New releases", "Top artists", "Radios"]
66 assert [row.translation_key for row in rows] == [
67 "featured",
68 "new_releases",
69 "top_artists",
70 "radios",
71 ]
72 assert [row.icon for row in rows] == ["mdi-star", "mdi-new-box", "mdi-trending-up", "mdi-radio"]
73 assert all(not row.items for row in rows)
74
75
76@pytest.mark.asyncio
77@pytest.mark.parametrize(
78 ("suffix", "endpoint", "expected_item_ids"),
79 [
80 ("home", "/home", ["1"]),
81 ("news", "/news/0", ["b1"]),
82 ("tops", "/tops", ["2"]),
83 ("radios", "/radios", ["3"]),
84 ],
85)
86async def test_get_recommendation_items_fetches_only_that_row(
87 provider: MusicMeProvider,
88 suffix: str,
89 endpoint: str,
90 expected_item_ids: list[str],
91) -> None:
92 """get_recommendation_items triggers exactly the requested row's backend fetch."""
93 _install_cache_mocks(provider)
94 api_mock = _stub_api_get(provider)
95
96 items = await provider.get_recommendation_items(f"{provider.instance_id}_{suffix}")
97
98 api_mock.assert_awaited_once()
99 assert api_mock.call_args.args[0].split("?", 1)[0] == endpoint
100 assert isinstance(items, UniqueList)
101 assert [item.item_id for item in items] == expected_item_ids
102
103
104@pytest.mark.asyncio
105async def test_get_recommendation_items_unknown_id_returns_empty(
106 provider: MusicMeProvider,
107) -> None:
108 """An unknown row item_id returns an empty result without backend calls."""
109 _install_cache_mocks(provider)
110 api_mock = _stub_api_get(provider)
111
112 items = await provider.get_recommendation_items("bogus")
113
114 api_mock.assert_not_awaited()
115 assert items == []
116
117
118@pytest.mark.asyncio
119async def test_get_recommendation_items_cache_hit_skips_backend(
120 provider: MusicMeProvider,
121) -> None:
122 """A fresh cache hit is served as media items without backend calls."""
123 # production get_with_freshness reconstructs a base_class entry via from_dict
124 # before returning it, so the decorator receives a RecommendationFolder
125 cached_folder = RecommendationFolder(
126 name="Featured",
127 translation_key="featured",
128 item_id=f"{provider.instance_id}_home",
129 provider=provider.instance_id,
130 icon="mdi-star",
131 items=UniqueList([provider._parse_radio({"id": 1, "name": "Home Radio"})]),
132 )
133 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
134 return_value=(cached_folder, True, True)
135 )
136 api_mock = _stub_api_get(provider)
137
138 items = await provider.get_recommendation_items(f"{provider.instance_id}_home")
139
140 api_mock.assert_not_awaited()
141 assert len(items) == 1
142 assert isinstance(items[0], Radio)
143 assert items[0].item_id == "1"
144
145
146@pytest.mark.asyncio
147async def test_get_recommendation_items_warm_cache_hit_skips_second_fetch(
148 provider: MusicMeProvider,
149) -> None:
150 """The second items call for a row is served from the cache, not the backend."""
151 cache_store: dict[str, Any] = {}
152 background_tasks: list[asyncio.Future[Any]] = []
153
154 async def _cache_get(key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
155 if key not in cache_store:
156 return None, False, False
157 data = cache_store[key]
158 base_class = kwargs.get("base_class")
159 if base_class is not None and data is not None:
160 return base_class.from_dict(data), True, True
161 return data, True, True
162
163 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
164 # mirrors production: entries are stored serialized (to_dict) and, since the
165 # caller passes base_class=RecommendationFolder, reconstructed via from_dict above
166 cache_store[key] = data.to_dict() if data is not None else None
167
168 def _create_task(target: Any, **_kwargs: Any) -> asyncio.Future[Any]:
169 task: asyncio.Future[Any] = asyncio.ensure_future(target)
170 background_tasks.append(task)
171 return task
172
173 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
174 side_effect=_cache_get
175 )
176 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
177 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
178 api_mock = _stub_api_get(provider)
179
180 first = await provider.get_recommendation_items(f"{provider.instance_id}_home")
181 # let the background cache-store task complete before the second call
182 await asyncio.gather(*background_tasks)
183 second = await provider.get_recommendation_items(f"{provider.instance_id}_home")
184
185 api_mock.assert_awaited_once()
186 assert [item.item_id for item in first] == ["1"]
187 assert [item.item_id for item in second] == ["1"]
188 assert isinstance(second[0], Radio)
189