music-assistant-server

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.controllers.music.recommendations.library import library_rows
12from music_assistant.mass import MusicAssistant
13from music_assistant.models.music_provider import MusicProvider
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    prov_a = _build(_RowsProvider, instance_id="prov_a")
104    prov_b = _build(_RowsProvider, instance_id="prov_b")
105    monkeypatch.setattr(
106        mass, "get_providers_supporting_feature", lambda *_a, **_k: [prov_a, prov_b]
107    )
108    folders = await mass.music.recommendations.get_recommendations()
109    assert [f.provider for f in folders[:3]] == ["library", "prov_a", "prov_b"]
110    assert all(f.provider == "library" for f in folders[3:])
111
112
113async def test_rows_interleave_uneven_provider_lengths_no_tail_dropped(
114    mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
115) -> None:
116    """Providers with fewer rows than the builtin sources still contribute all their rows."""
117    four = _build(_FourRowsProvider, instance_id="four")
118    one = _build(_OneRowProvider, instance_id="one")
119    monkeypatch.setattr(mass, "get_providers_supporting_feature", lambda *_a, **_k: [four, one])
120
121    folders = await mass.music.recommendations.get_recommendations()
122
123    builtin_ids = [f.item_id for f in library_rows()]
124    expected: list[tuple[str, str]] = []
125    for i, builtin_id in enumerate(builtin_ids):
126        expected.append(("library", builtin_id))
127        if i < 4:
128            expected.append(("four", f"four_row_{i}"))
129        if i < 1:
130            expected.append(("one", "one_row_0"))
131
132    assert [(f.provider, f.item_id) for f in folders] == expected
133    assert len(folders) == len(builtin_ids) + 4 + 1
134
135
136async def test_items_routed_to_provider(
137    mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
138) -> None:
139    """Items for a provider row are fetched from that provider."""
140    provider = _build(_RowsProvider)
141    monkeypatch.setattr(mass, "get_provider", lambda *_a, **_k: provider)
142    items = await mass.music.recommendations.get_recommendation_items("fake_instance", "row1")
143    assert [item.item_id for item in items] == ["prov-item"]
144
145
146async def test_items_unknown_provider_row_returns_empty(
147    mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
148) -> None:
149    """An unknown row id on a known provider returns an empty list."""
150    provider = _build(_RowsProvider)
151    monkeypatch.setattr(mass, "get_provider", lambda *_a, **_k: provider)
152    items = await mass.music.recommendations.get_recommendation_items(
153        "fake_instance", "no_such_row"
154    )
155    assert items == []
156
157
158async def test_items_unknown_provider_returns_empty(mass: MusicAssistant) -> None:
159    """An unknown provider instance returns an empty list."""
160    items = await mass.music.recommendations.get_recommendation_items("no_such_provider", "row1")
161    assert items == []
162
163
164@patch("music_assistant.controllers.music.controller.get_current_user")
165async def test_items_restricted_provider_returns_empty(
166    mock_get_user: Mock, mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
167) -> None:
168    """A user's provider filter blocks fetching items from a restricted music provider."""
169    mock_get_user.return_value = Mock(provider_filter=["allowed_instance"])
170    provider = _build(_RowsProvider, instance_id="restricted_instance")
171    provider.get_recommendation_items = AsyncMock()  # type: ignore[method-assign]
172    monkeypatch.setattr(mass, "get_provider", lambda *_a, **_k: provider)
173    items = await mass.music.recommendations.get_recommendation_items("restricted_instance", "row1")
174    assert items == []
175    provider.get_recommendation_items.assert_not_awaited()
176
177
178@patch("music_assistant.controllers.music.controller.get_current_user")
179async def test_rows_restricted_provider_returns_no_rows(
180    mock_get_user: Mock, mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
181) -> None:
182    """A user's provider filter excludes a restricted music provider's rows from the listing."""
183    mock_get_user.return_value = Mock(provider_filter=["allowed_instance"])
184    restricted = _build(_RowsProvider, instance_id="restricted_instance")
185    monkeypatch.setattr(mass, "get_providers_supporting_feature", lambda *_a, **_k: [restricted])
186    folders = await mass.music.recommendations.get_recommendations()
187    assert not any(f.provider == "restricted_instance" for f in folders)
188