/
/
/
1"""Tests for the Last.fm Recommendations two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncIterator
7from typing import cast
8from unittest.mock import AsyncMock, Mock, patch
9
10import pytest
11from music_assistant_models.enums import MediaType
12from music_assistant_models.media_items import ItemMapping, RecommendationFolder, UniqueList
13
14from music_assistant.providers.lastfm_recommendations import (
15 SUPPORTED_FEATURES,
16 LastFMRecommendationsProvider,
17)
18
19INSTANCE_ID = "lastfm_recommendations--test1"
20
21
22@pytest.fixture
23def mass_mock() -> Mock:
24 """Return a mock MusicAssistant instance."""
25 mass = Mock()
26 mass.http_session = AsyncMock()
27 return mass
28
29
30@pytest.fixture
31def manifest_mock() -> Mock:
32 """Return a mock provider manifest."""
33 manifest = Mock()
34 manifest.domain = "lastfm_recommendations"
35 return manifest
36
37
38@pytest.fixture
39def config_mock() -> Mock:
40 """Return a mock provider config."""
41 config = Mock()
42 config.name = "Last.fm Test"
43 config.instance_id = INSTANCE_ID
44 config.get_value.side_effect = lambda key, default=None: {"log_level": "INFO"}.get(key, default)
45 return config
46
47
48@pytest.fixture
49async def provider(
50 mass_mock: Mock, manifest_mock: Mock, config_mock: Mock
51) -> LastFMRecommendationsProvider:
52 """Return an initialized provider with a mocked API client and untouched mass mock."""
53 provider = LastFMRecommendationsProvider(
54 mass_mock, manifest_mock, config_mock, SUPPORTED_FEATURES
55 )
56 with patch("music_assistant.providers.lastfm_recommendations.LastFMAPIClient") as api_cls:
57 api_cls.return_value = Mock()
58 await provider.handle_async_init()
59 # forget the init-time calls so tests can assert zero backend/mass activity
60 mass_mock.reset_mock()
61 return provider
62
63
64def _make_folders() -> list[RecommendationFolder]:
65 """Build two folders with items, shaped like the background builder's output."""
66 return [
67 RecommendationFolder(
68 item_id=f"{INSTANCE_ID}_similar_artists",
69 name="Discover Similar Artists",
70 translation_key="discover_similar_artists",
71 translation_params=["5"],
72 provider=INSTANCE_ID,
73 items=UniqueList(
74 [
75 ItemMapping(
76 media_type=MediaType.ARTIST,
77 item_id="a1",
78 provider=INSTANCE_ID,
79 name="Artist One",
80 )
81 ]
82 ),
83 subtitle="Based on your top 5 artists",
84 icon="mdi-account-music-outline",
85 ),
86 RecommendationFolder(
87 item_id=f"{INSTANCE_ID}_chart_top_tracks",
88 name="Global Top Tracks",
89 translation_key="global_top_tracks",
90 provider=INSTANCE_ID,
91 items=UniqueList(
92 [
93 ItemMapping(
94 media_type=MediaType.TRACK,
95 item_id="t1",
96 provider=INSTANCE_ID,
97 name="Track One",
98 )
99 ]
100 ),
101 subtitle="Most popular tracks worldwide",
102 icon="mdi-chart-box",
103 ),
104 ]
105
106
107async def test_get_recommendations_returns_rows_without_items(
108 provider: LastFMRecommendationsProvider, mass_mock: Mock
109) -> None:
110 """Rows mirror the in-memory folders' identity, stripped of items, with zero backend I/O."""
111 folders = _make_folders()
112 provider._recommendation_folders = folders
113
114 rows = await provider.get_recommendations()
115
116 assert [row.item_id for row in rows] == [
117 f"{INSTANCE_ID}_similar_artists",
118 f"{INSTANCE_ID}_chart_top_tracks",
119 ]
120 similar_artists = rows[0]
121 assert len(similar_artists.items) == 0
122 assert similar_artists.provider == INSTANCE_ID
123 assert similar_artists.name == "Discover Similar Artists"
124 assert similar_artists.translation_key == "discover_similar_artists"
125 assert similar_artists.translation_params == ["5"]
126 assert similar_artists.icon == "mdi-account-music-outline"
127 assert similar_artists.subtitle == "Based on your top 5 artists"
128 # stripping produced fresh copies: the in-memory folders keep their items
129 assert similar_artists is not folders[0]
130 assert len(folders[0].items) == 1
131 # static rows: no backend or mass activity at all
132 assert cast("Mock", provider.api).mock_calls == []
133 assert mass_mock.mock_calls == []
134
135
136async def test_get_recommendations_empty_before_first_build(
137 provider: LastFMRecommendationsProvider,
138) -> None:
139 """Before the background builder has run, there are no rows."""
140 assert await provider.get_recommendations() == []
141
142
143async def test_get_recommendation_items_returns_row_items(
144 provider: LastFMRecommendationsProvider, mass_mock: Mock
145) -> None:
146 """One row's items come straight from the matching in-memory folder, with zero backend I/O."""
147 folders = _make_folders()
148 provider._recommendation_folders = folders
149
150 items = await provider.get_recommendation_items(f"{INSTANCE_ID}_chart_top_tracks")
151
152 assert items == folders[1].items
153 assert [item.item_id for item in items] == ["t1"]
154 assert cast("Mock", provider.api).mock_calls == []
155 assert mass_mock.mock_calls == []
156
157
158async def test_get_recommendation_items_unknown_id_returns_empty(
159 provider: LastFMRecommendationsProvider,
160) -> None:
161 """An unknown row item_id yields an empty UniqueList."""
162 provider._recommendation_folders = _make_folders()
163
164 result = await provider.get_recommendation_items("bogus_row")
165
166 assert isinstance(result, UniqueList)
167 assert len(result) == 0
168
169
170async def test_refresh_recommendations_serves_previous_generation_during_rebuild(
171 provider: LastFMRecommendationsProvider,
172) -> None:
173 """Old rows keep being served mid-rebuild; only the final swap replaces them."""
174 old_folders = _make_folders()
175 provider._recommendation_folders = old_folders
176
177 new_folder = RecommendationFolder(
178 item_id=f"{INSTANCE_ID}_new_row",
179 name="New Row",
180 translation_key="new_row",
181 provider=INSTANCE_ID,
182 items=UniqueList(),
183 )
184
185 builder_started = asyncio.Event()
186 resume_builder = asyncio.Event()
187
188 async def gated_builder() -> AsyncIterator[RecommendationFolder]:
189 builder_started.set()
190 await resume_builder.wait()
191 yield new_folder
192
193 provider.recommendations_manager.build_recommendation_folders = gated_builder # type: ignore[method-assign]
194
195 refresh_task = asyncio.create_task(provider._refresh_recommendations())
196 await builder_started.wait()
197
198 # Mid-rebuild: the previous generation's rows and items are still served.
199 mid_rows = await provider.get_recommendations()
200 assert [row.item_id for row in mid_rows] == [
201 f"{INSTANCE_ID}_similar_artists",
202 f"{INSTANCE_ID}_chart_top_tracks",
203 ]
204 mid_items = await provider.get_recommendation_items(f"{INSTANCE_ID}_chart_top_tracks")
205 assert [item.item_id for item in mid_items] == ["t1"]
206
207 resume_builder.set()
208 await refresh_task
209
210 # After the swap: only the newly built generation is served.
211 new_rows = await provider.get_recommendations()
212 assert [row.item_id for row in new_rows] == [f"{INSTANCE_ID}_new_row"]
213