/
/
/
1"""Test Open Subsonic get_recommendations() rows and get_recommendation_items() fetching."""
2
3from __future__ import annotations
4
5import asyncio
6import pathlib
7from typing import Any
8from unittest.mock import AsyncMock, Mock
9
10import pytest
11from libopensonic.media import AlbumID3
12from music_assistant_models.enums import MediaType
13from music_assistant_models.media_items import ItemMapping, RecommendationFolder, UniqueList
14
15from music_assistant.providers.opensubsonic.sonic_provider import OpenSonicProvider
16
17FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures"
18
19ALL_ROW_IDS = [
20 "subsonic_newest_podcasts",
21 "subsonic_starred_albums",
22 "subsonic_new_albums",
23 "subsonic_most_played",
24]
25
26
27def _make_folder(provider: OpenSonicProvider, item_id: str, name: str) -> RecommendationFolder:
28 return RecommendationFolder(
29 item_id=item_id,
30 provider=provider.domain,
31 name=name,
32 items=UniqueList(
33 [
34 ItemMapping(
35 media_type=MediaType.ALBUM,
36 item_id=f"{item_id}_album",
37 provider=provider.instance_id,
38 name=f"{name} Album",
39 )
40 ]
41 ),
42 )
43
44
45def _enable_all_rows(provider: OpenSonicProvider) -> None:
46 provider._enable_podcasts = True
47 provider._show_faves = True
48 provider._show_new = True
49 provider._show_played = True
50
51
52def _stub_shelf_builders(provider: OpenSonicProvider) -> dict[str, AsyncMock]:
53 """Replace all four shelf-builder methods with AsyncMocks and enable all config gates."""
54 _enable_all_rows(provider)
55 mocks = {
56 "_podcast_recommendations": AsyncMock(
57 return_value=_make_folder(provider, "subsonic_newest_podcasts", "Newest Podcasts")
58 ),
59 "_favorites_recommendation": AsyncMock(
60 return_value=_make_folder(provider, "subsonic_starred_albums", "Starred Albums")
61 ),
62 "_new_recommendations": AsyncMock(
63 return_value=_make_folder(provider, "subsonic_new_albums", "New Albums")
64 ),
65 "_played_recommendations": AsyncMock(
66 return_value=_make_folder(provider, "subsonic_most_played", "Most Played Albums")
67 ),
68 }
69 for name, mock in mocks.items():
70 setattr(provider, name, mock)
71 return mocks
72
73
74def _install_cache_mocks(provider: OpenSonicProvider) -> None:
75 """Make the @use_cache decorator treat every call as a cache miss."""
76 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
77 return_value=(None, False, False)
78 )
79 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
80
81
82def _install_dict_backed_cache(provider: OpenSonicProvider) -> list[asyncio.Future[Any]]:
83 """
84 Back the @use_cache decorator with a real dict store.
85
86 Mirrors production: set() stores the serialized (to_dict) folder and, since the
87 caller passes base_class=RecommendationFolder, get_with_freshness reconstructs it
88 via RecommendationFolder.from_dict on a hit. Returns the list of background
89 cache-store tasks so tests can await them before the warm call.
90 """
91 store: dict[str, Any] = {}
92 tasks: list[asyncio.Future[Any]] = []
93
94 async def _cache_get(key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
95 if key not in store:
96 return None, False, False
97 data = store[key]
98 base_class = kwargs.get("base_class")
99 if base_class is not None and data is not None:
100 return base_class.from_dict(data), True, True
101 return data, True, True
102
103 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
104 store[key] = data.to_dict() if data is not None else None
105
106 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
107 task: asyncio.Future[Any] = asyncio.ensure_future(target)
108 tasks.append(task)
109 return task
110
111 provider.mass.cache.get_with_freshness = AsyncMock(side_effect=_cache_get) # type: ignore[method-assign]
112 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
113 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
114 return tasks
115
116
117@pytest.mark.asyncio
118async def test_get_recommendations_returns_all_rows_without_backend_calls(
119 provider: OpenSonicProvider,
120) -> None:
121 """With all config flags on, all four rows are returned in order with zero backend I/O."""
122 _enable_all_rows(provider)
123 provider.conn = Mock()
124
125 result = await provider.get_recommendations()
126
127 assert provider.conn.mock_calls == []
128 assert [folder.item_id for folder in result] == ALL_ROW_IDS
129 assert all(len(folder.items) == 0 for folder in result)
130 assert [folder.translation_key for folder in result] == [
131 "episodes_recently_added",
132 "starred_items",
133 "recently_added_albums",
134 "most_played_albums",
135 ]
136
137
138@pytest.mark.asyncio
139async def test_get_recommendations_omits_config_disabled_rows(
140 provider: OpenSonicProvider,
141) -> None:
142 """Rows whose config flag is off are absent from the rows listing."""
143 _enable_all_rows(provider)
144 provider._enable_podcasts = False
145 provider._show_new = False
146 provider.conn = Mock()
147
148 result = await provider.get_recommendations()
149
150 assert [folder.item_id for folder in result] == [
151 "subsonic_starred_albums",
152 "subsonic_most_played",
153 ]
154
155
156@pytest.mark.asyncio
157async def test_get_recommendation_items_builds_only_that_row(
158 provider: OpenSonicProvider,
159) -> None:
160 """Requesting one row runs only that row's shelf builder and returns its items."""
161 _install_cache_mocks(provider)
162 mocks = _stub_shelf_builders(provider)
163
164 result = await provider.get_recommendation_items("subsonic_new_albums")
165
166 mocks["_new_recommendations"].assert_awaited_once()
167 mocks["_podcast_recommendations"].assert_not_awaited()
168 mocks["_favorites_recommendation"].assert_not_awaited()
169 mocks["_played_recommendations"].assert_not_awaited()
170 assert [item.item_id for item in result] == ["subsonic_new_albums_album"]
171
172
173@pytest.mark.asyncio
174async def test_get_recommendation_items_unknown_id_returns_empty(
175 provider: OpenSonicProvider,
176) -> None:
177 """An unknown row item_id returns an empty UniqueList without any shelf builds."""
178 _install_cache_mocks(provider)
179 mocks = _stub_shelf_builders(provider)
180
181 result = await provider.get_recommendation_items("bogus_row")
182
183 for mock in mocks.values():
184 mock.assert_not_awaited()
185 assert isinstance(result, UniqueList)
186 assert len(result) == 0
187
188
189@pytest.mark.asyncio
190async def test_get_recommendation_items_config_disabled_row_returns_empty(
191 provider: OpenSonicProvider,
192) -> None:
193 """A row disabled in the provider config yields no items even when requested directly."""
194 _install_cache_mocks(provider)
195 mocks = _stub_shelf_builders(provider)
196 provider._show_new = False
197
198 result = await provider.get_recommendation_items("subsonic_new_albums")
199
200 mocks["_new_recommendations"].assert_not_awaited()
201 assert len(result) == 0
202
203
204@pytest.mark.asyncio
205async def test_get_recommendation_items_warm_cache_hit_skips_backend(
206 provider: OpenSonicProvider,
207) -> None:
208 """A second items call is served from the cached folder without hitting the backend."""
209 _enable_all_rows(provider)
210 store_tasks = _install_dict_backed_cache(provider)
211 album = AlbumID3.from_json(
212 (FIXTURES_DIR / "albums" / "spec.album.json").read_text(encoding="utf-8")
213 )
214 provider.conn = Mock()
215 provider.conn.get_album_list2 = AsyncMock(return_value=[album])
216
217 first = await provider.get_recommendation_items("subsonic_new_albums")
218 # let the decorator's background cache-store task complete
219 await asyncio.gather(*store_tasks)
220 second = await provider.get_recommendation_items("subsonic_new_albums")
221
222 # backend was fetched exactly once: the warm call was served from the cache
223 provider.conn.get_album_list2.assert_awaited_once()
224 assert [item.item_id for item in first] == [album.id]
225 assert [item.item_id for item in second] == [album.id]
226