/
/
/
1"""Tests for routing and security on the music/recommendations/items endpoint."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import AsyncMock, MagicMock, Mock, patch
7
8from music_assistant_models.enums import MediaType, ProviderFeature, ProviderType
9from music_assistant_models.media_items import ItemMapping, RecommendationFolder, UniqueList
10
11from music_assistant.mass import MusicAssistant
12from music_assistant.models.music_provider import MusicProvider
13from music_assistant.providers.recommendations import LibraryRecommendationsProvider
14
15if TYPE_CHECKING:
16 import pytest
17 from music_assistant_models.media_items import BrowseFolder, MediaItemType
18
19
20class _RowsProvider(MusicProvider):
21 """A fake provider exposing one recommendation row with one item."""
22
23 async def get_recommendations(self) -> list[RecommendationFolder]:
24 return [
25 RecommendationFolder(
26 item_id="row1",
27 provider=self.instance_id,
28 name="Row 1",
29 translation_key="row1_key",
30 icon="mdi-row1",
31 )
32 ]
33
34 async def get_recommendation_items(
35 self, item_id: str
36 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
37 if item_id != "row1":
38 return UniqueList()
39 return UniqueList(
40 [
41 ItemMapping.from_dict(
42 {
43 "item_id": "prov-item",
44 "provider": self.instance_id,
45 "media_type": MediaType.TRACK.value,
46 "name": "Provider Item",
47 }
48 )
49 ]
50 )
51
52
53class _FourRowsProvider(MusicProvider):
54 """A fake provider exposing four recommendation rows."""
55
56 async def get_recommendations(self) -> list[RecommendationFolder]:
57 return [
58 RecommendationFolder(
59 item_id=f"four_row_{i}",
60 provider=self.instance_id,
61 name=f"Four Row {i}",
62 translation_key=f"four_row_{i}_key",
63 icon="mdi-four",
64 )
65 for i in range(4)
66 ]
67
68
69class _OneRowProvider(MusicProvider):
70 """A fake provider exposing a single recommendation row."""
71
72 async def get_recommendations(self) -> list[RecommendationFolder]:
73 return [
74 RecommendationFolder(
75 item_id="one_row_0",
76 provider=self.instance_id,
77 name="One Row 0",
78 translation_key="one_row_0_key",
79 icon="mdi-one",
80 )
81 ]
82
83
84def _build(provider_cls: type[MusicProvider], instance_id: str = "fake_instance") -> MusicProvider:
85 """Construct a minimal provider with stubbed mass/manifest/config."""
86 mass = MagicMock()
87 manifest = MagicMock()
88 manifest.type = ProviderType.MUSIC
89 manifest.domain = "fake"
90 config = MagicMock()
91 config.name = "Fake Provider"
92 config.instance_id = instance_id
93 config.get_value = MagicMock(return_value="GLOBAL")
94 return provider_cls(
95 mass, manifest, config, supported_features={ProviderFeature.RECOMMENDATIONS}
96 )
97
98
99async def test_rows_interleaved_builtin_first(
100 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
101) -> None:
102 """Builtin and provider rows are interleaved one per source per pass, builtin first."""
103 recommendations_provider = mass.get_provider("recommendations")
104 assert recommendations_provider is not None
105 assert isinstance(recommendations_provider, LibraryRecommendationsProvider)
106 prov_a = _build(_RowsProvider, instance_id="prov_a")
107 prov_b = _build(_RowsProvider, instance_id="prov_b")
108 monkeypatch.setattr(
109 mass,
110 "get_providers_supporting_feature",
111 lambda *_a, **_k: [recommendations_provider, prov_a, prov_b],
112 )
113 folders = await mass.music.recommendations.get_recommendations()
114 assert [f.provider for f in folders[:3]] == ["recommendations", "prov_a", "prov_b"]
115 assert all(f.provider == "recommendations" for f in folders[3:])
116
117
118async def test_rows_interleave_uneven_provider_lengths_no_tail_dropped(
119 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
120) -> None:
121 """Providers with fewer rows than the builtin sources still contribute all their rows."""
122 recommendations_provider = mass.get_provider("recommendations")
123 assert recommendations_provider is not None
124 assert isinstance(recommendations_provider, LibraryRecommendationsProvider)
125 four = _build(_FourRowsProvider, instance_id="four")
126 one = _build(_OneRowProvider, instance_id="one")
127 monkeypatch.setattr(
128 mass,
129 "get_providers_supporting_feature",
130 lambda *_a, **_k: [recommendations_provider, four, one],
131 )
132
133 folders = await mass.music.recommendations.get_recommendations()
134
135 provider = mass.get_provider("recommendations")
136 assert provider is not None
137 assert isinstance(provider, LibraryRecommendationsProvider)
138 builtin_ids = [f.item_id for f in await provider.get_recommendations()]
139 expected: list[tuple[str, str]] = []
140 for i, builtin_id in enumerate(builtin_ids):
141 expected.append(("recommendations", builtin_id))
142 if i < 4:
143 expected.append(("four", f"four_row_{i}"))
144 if i < 1:
145 expected.append(("one", "one_row_0"))
146
147 assert [(f.provider, f.item_id) for f in folders] == expected
148 assert len(folders) == len(builtin_ids) + 4 + 1
149
150
151async def test_items_routed_to_provider(
152 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
153) -> None:
154 """Items for a provider row are fetched from that provider."""
155 provider = _build(_RowsProvider)
156 monkeypatch.setattr(mass, "get_provider", lambda *_a, **_k: provider)
157 items = await mass.music.recommendations.get_recommendation_items("fake_instance", "row1")
158 assert [item.item_id for item in items] == ["prov-item"]
159
160
161async def test_items_unknown_provider_row_returns_empty(
162 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
163) -> None:
164 """An unknown row id on a known provider returns an empty list."""
165 provider = _build(_RowsProvider)
166 monkeypatch.setattr(mass, "get_provider", lambda *_a, **_k: provider)
167 items = await mass.music.recommendations.get_recommendation_items(
168 "fake_instance", "no_such_row"
169 )
170 assert items == []
171
172
173async def test_items_unknown_provider_returns_empty(mass: MusicAssistant) -> None:
174 """An unknown provider instance returns an empty list."""
175 items = await mass.music.recommendations.get_recommendation_items("no_such_provider", "row1")
176 assert items == []
177
178
179@patch("music_assistant.controllers.music.controller.get_current_user")
180async def test_items_restricted_provider_returns_empty(
181 mock_get_user: Mock, mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
182) -> None:
183 """A user's provider filter blocks fetching items from a restricted music provider."""
184 mock_get_user.return_value = Mock(provider_filter=["allowed_instance"])
185 provider = _build(_RowsProvider, instance_id="restricted_instance")
186 provider.get_recommendation_items = AsyncMock() # type: ignore[method-assign]
187 monkeypatch.setattr(mass, "get_provider", lambda *_a, **_k: provider)
188 items = await mass.music.recommendations.get_recommendation_items("restricted_instance", "row1")
189 assert items == []
190 provider.get_recommendation_items.assert_not_awaited()
191
192
193@patch("music_assistant.controllers.music.controller.get_current_user")
194async def test_rows_restricted_provider_returns_no_rows(
195 mock_get_user: Mock, mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
196) -> None:
197 """A user's provider filter excludes a restricted music provider's rows from the listing."""
198 mock_get_user.return_value = Mock(provider_filter=["allowed_instance"])
199 restricted = _build(_RowsProvider, instance_id="restricted_instance")
200 monkeypatch.setattr(mass, "get_providers_supporting_feature", lambda *_a, **_k: [restricted])
201 folders = await mass.music.recommendations.get_recommendations()
202 assert not any(f.provider == "restricted_instance" for f in folders)
203
204
205async def test_items_providers_forwarded_to_builtin_provider(
206 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
207) -> None:
208 """The providers filter is forwarded to the builtin library recommendations provider."""
209 builtin_provider = mass.get_provider("recommendations")
210 assert builtin_provider is not None
211 assert isinstance(builtin_provider, LibraryRecommendationsProvider)
212 spy = AsyncMock(return_value=UniqueList())
213 monkeypatch.setattr(builtin_provider, "get_recommendation_items", spy)
214
215 await mass.music.recommendations.get_recommendation_items(
216 "recommendations", "recently_played", providers=["prov_a"]
217 )
218
219 spy.assert_awaited_once_with("recently_played", providers=["prov_a"])
220
221
222async def test_items_providers_ignored_for_external_provider(
223 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
224) -> None:
225 """An external provider row's SPI is unchanged: it never receives the providers filter."""
226 provider = _build(_RowsProvider)
227 monkeypatch.setattr(mass, "get_provider", lambda *_a, **_k: provider)
228
229 items = await mass.music.recommendations.get_recommendation_items(
230 "fake_instance", "row1", providers=["prov_a"]
231 )
232
233 assert [item.item_id for item in items] == ["prov-item"]
234