music-assistant-server

1"""Tests for the recommendations subcontroller (rows + builtin items)."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import MediaType
8
9from music_assistant.constants import DB_TABLE_PLAYLOG
10from music_assistant.mass import MusicAssistant
11from music_assistant.providers.recommendations import LibraryRecommendationsProvider, LibraryRowID
12
13if TYPE_CHECKING:
14    import pytest
15    from music_assistant_models.media_items import ItemMapping
16
17EXPECTED_DEFAULT_ORDER = [
18    "in_progress",
19    "recently_played",
20    "recently_added_tracks",
21    "recently_added_albums",
22    "random_artists",
23    "random_albums",
24    "recent_favorite_tracks",
25    "favorite_playlists",
26    "favorite_radio",
27    "recent_artists",
28    "recent_tracks",
29    "forgotten_tracks",
30    "forgotten_albums",
31    "forgotten_artists",
32    "most_played_tracks",
33    "never_played_tracks",
34]
35
36
37async def test_default_recommendations_order(mass: MusicAssistant) -> None:
38    """The default library rows appear in their canonical order."""
39    folders = await mass.music.recommendations.get_recommendations()
40    defaults = [f.item_id for f in folders if f.item_id in EXPECTED_DEFAULT_ORDER]
41    assert defaults == EXPECTED_DEFAULT_ORDER
42
43
44async def test_recommendations_rows_have_no_items(mass: MusicAssistant) -> None:
45    """The rows listing returns descriptors only; no row carries items."""
46    await _add_playlog_row(
47        mass, item_id="album-1", media_type=MediaType.ALBUM, timestamp=2000, user_initiated=True
48    )
49    folders = await mass.music.recommendations.get_recommendations()
50    assert folders
51    assert all(folder.items == [] for folder in folders)
52
53
54async def test_library_rows_have_descriptor_fields(mass: MusicAssistant) -> None:
55    """Library rows carry their identity fields, correct defaults, and no items."""
56    provider = mass.get_provider("recommendations")
57    assert provider is not None, "recommendations provider should be loaded as builtin"
58    assert isinstance(provider, LibraryRecommendationsProvider)
59    rows = await provider.get_recommendations()
60    in_progress = next(f for f in rows if f.item_id == "in_progress")
61    assert in_progress.provider == "recommendations"
62    assert in_progress.name == "In progress"
63    assert in_progress.translation_key == "in_progress_items"
64    assert in_progress.icon == "mdi-motion-play"
65    assert in_progress.enabled_by_default is True
66    random_artists = next(f for f in rows if f.item_id == "random_artists")
67    assert random_artists.enabled_by_default is False
68    assert all(folder.items == [] for folder in rows)
69
70
71async def test_recently_played_rolls_up_to_container(mass: MusicAssistant) -> None:
72    """Playing an album shows the album, not its individual tracks."""
73    await _add_playlog_row(
74        mass, item_id="album-1", media_type=MediaType.ALBUM, timestamp=2000, user_initiated=True
75    )
76    await _add_playlog_row(
77        mass, item_id="track-1", media_type=MediaType.TRACK, timestamp=1999, user_initiated=False
78    )
79    await _add_playlog_row(
80        mass,
81        item_id="track-direct",
82        media_type=MediaType.TRACK,
83        timestamp=2001,
84        user_initiated=True,
85    )
86    items = await mass.music.recommendations.get_recommendation_items(
87        "recommendations", "recently_played"
88    )
89    item_ids = {item.item_id for item in items}
90    assert "album-1" in item_ids
91    assert "track-1" not in item_ids
92    assert "track-direct" in item_ids
93
94
95async def test_recent_artists_and_tracks_rows_present(mass: MusicAssistant) -> None:
96    """Recent Artists shows played artists; Recent Tracks shows played tracks."""
97    await _add_playlog_row(
98        mass, item_id="artist-1", media_type=MediaType.ARTIST, timestamp=3000, user_initiated=True
99    )
100    await _add_playlog_row(
101        mass, item_id="track-9", media_type=MediaType.TRACK, timestamp=2999, user_initiated=False
102    )
103
104    artist_items = await mass.music.recommendations.get_recommendation_items(
105        "recommendations", "recent_artists"
106    )
107    track_items = await mass.music.recommendations.get_recommendation_items(
108        "recommendations", "recent_tracks"
109    )
110
111    assert "artist-1" in {item.item_id for item in artist_items}
112    assert "artist-1" not in {item.item_id for item in track_items}
113    assert "track-9" in {item.item_id for item in track_items}
114    assert "track-9" not in {item.item_id for item in artist_items}
115
116
117async def test_recently_played_includes_podcast_and_audiobook_containers(
118    mass: MusicAssistant,
119) -> None:
120    """Recently Played includes podcast/audiobook containers but excludes episodes and non-user-initiated tracks."""
121    await _add_playlog_row(
122        mass, item_id="album-x", media_type=MediaType.ALBUM, timestamp=3000, user_initiated=True
123    )
124    await _add_playlog_row(
125        mass,
126        item_id="podcast-x",
127        media_type=MediaType.PODCAST,
128        timestamp=3001,
129        user_initiated=False,
130    )
131    await _add_playlog_row(
132        mass,
133        item_id="audiobook-x",
134        media_type=MediaType.AUDIOBOOK,
135        timestamp=3002,
136        user_initiated=False,
137    )
138    await _add_playlog_row(
139        mass,
140        item_id="episode-x",
141        media_type=MediaType.PODCAST_EPISODE,
142        timestamp=3003,
143        user_initiated=False,
144    )
145    await _add_playlog_row(
146        mass,
147        item_id="loose-track",
148        media_type=MediaType.TRACK,
149        timestamp=2999,
150        user_initiated=False,
151    )
152    items = await mass.music.recommendations.get_recommendation_items(
153        "recommendations", "recently_played"
154    )
155    item_ids = {item.item_id for item in items}
156    assert "album-x" in item_ids, "album (user-initiated) should appear"
157    assert "podcast-x" in item_ids, "podcast show should always appear"
158    assert "audiobook-x" in item_ids, "audiobook should always appear"
159    assert "episode-x" not in item_ids, "podcast episode should not appear"
160    assert "loose-track" not in item_ids, "non-user-initiated track should be filtered out"
161
162
163async def test_recently_played_always_include_media_types_query(mass: MusicAssistant) -> None:
164    """always_include_media_types OR-s in those types regardless of user_initiated_only."""
165    await _add_playlog_row(
166        mass,
167        item_id="podcast-q",
168        media_type=MediaType.PODCAST,
169        timestamp=5000,
170        user_initiated=False,
171    )
172    await _add_playlog_row(
173        mass,
174        item_id="track-q",
175        media_type=MediaType.TRACK,
176        timestamp=4999,
177        user_initiated=False,
178    )
179    results = await mass.music.recently_played(
180        media_types=[MediaType.TRACK],
181        user_initiated_only=True,
182        always_include_media_types=[MediaType.PODCAST],
183    )
184    result_ids = {item.item_id for item in results}
185    assert "podcast-q" in result_ids, "podcast should be returned via always_include_media_types"
186    assert "track-q" not in result_ids, "non-user-initiated track should be excluded"
187
188
189async def test_every_library_row_dispatches_a_query(mass: MusicAssistant) -> None:
190    """
191    Every id listed by get_recommendations() reaches a real query branch in get_recommendation_items().
192
193    The rows listing and the items dispatch live in two separate functions; this
194    pins that no listed row silently falls through to the empty default arm.
195    """
196    provider = mass.get_provider("recommendations")
197    assert provider is not None
198    assert isinstance(provider, LibraryRecommendationsProvider)
199
200    # Verify every enum value has a corresponding match case by checking that all folder IDs
201    # from get_recommendations() are valid LibraryRowID enum members
202    valid_ids = {row_id.value for row_id in LibraryRowID}
203    for folder in await provider.get_recommendations():
204        assert folder.item_id in valid_ids, (
205            f"row {folder.item_id!r} not in LibraryRowID enum - likely missing match case"
206        )
207
208
209async def test_library_rows_listed_by_controller(mass: MusicAssistant) -> None:
210    """Every library row appears in the controller's rows listing."""
211    folders = await mass.music.recommendations.get_recommendations()
212    listed = {f.item_id for f in folders if f.provider == "recommendations"}
213    provider = mass.get_provider("recommendations")
214    assert provider is not None
215    assert isinstance(provider, LibraryRecommendationsProvider)
216    expected_rows = {f.item_id for f in await provider.get_recommendations()}
217    assert expected_rows <= listed
218
219
220async def test_unknown_library_row_returns_empty(mass: MusicAssistant) -> None:
221    """Requesting items for an unknown builtin row returns an empty list."""
222    items = await mass.music.recommendations.get_recommendation_items(
223        "recommendations", "no_such_row"
224    )
225    assert items == []
226
227
228async def test_failing_library_row_items_isolated(
229    mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
230) -> None:
231    """A library row whose items query raises returns an empty list, not an error."""
232
233    async def _boom(**_kwargs: object) -> list[ItemMapping]:
234        raise RuntimeError("row boom")
235
236    monkeypatch.setattr(mass.music, "in_progress_items", _boom)
237    items = await mass.music.recommendations.get_recommendation_items(
238        "recommendations", "in_progress"
239    )
240    assert items == []
241
242
243async def test_forgotten_tracks_row_callable(mass: MusicAssistant) -> None:
244    """Forgotten Tracks row can be called and uses played_only parameter."""
245    items = await mass.music.recommendations.get_recommendation_items(
246        "recommendations", "forgotten_tracks"
247    )
248    assert isinstance(items, list)
249
250
251async def test_forgotten_albums_row_callable(mass: MusicAssistant) -> None:
252    """Forgotten Albums row can be called and uses played_only parameter."""
253    items = await mass.music.recommendations.get_recommendation_items(
254        "recommendations", "forgotten_albums"
255    )
256    assert isinstance(items, list)
257
258
259async def test_forgotten_artists_row_callable(mass: MusicAssistant) -> None:
260    """Forgotten Artists row can be called and uses played_only parameter."""
261    items = await mass.music.recommendations.get_recommendation_items(
262        "recommendations", "forgotten_artists"
263    )
264    assert isinstance(items, list)
265
266
267async def test_most_played_tracks_row_callable(mass: MusicAssistant) -> None:
268    """Most Played Tracks row can be called and uses play_count_desc ordering."""
269    items = await mass.music.recommendations.get_recommendation_items(
270        "recommendations", "most_played_tracks"
271    )
272    assert isinstance(items, list)
273
274
275async def test_never_played_tracks_row_callable(mass: MusicAssistant) -> None:
276    """Never / Rarely Played row can be called and uses play_count ordering."""
277    items = await mass.music.recommendations.get_recommendation_items(
278        "recommendations", "never_played_tracks"
279    )
280    assert isinstance(items, list)
281
282
283async def _add_playlog_row(
284    mass: MusicAssistant,
285    *,
286    item_id: str,
287    media_type: MediaType,
288    timestamp: int,
289    user_initiated: bool,
290    userid: str = "user-a",
291) -> None:
292    await mass.music.database.insert(
293        DB_TABLE_PLAYLOG,
294        {
295            "item_id": item_id,
296            "provider": "library",
297            "media_type": media_type.value,
298            "name": f"{media_type.value} {item_id}",
299            "timestamp": timestamp,
300            "fully_played": True,
301            "seconds_played": 180,
302            "userid": userid,
303            "user_initiated": user_initiated,
304        },
305    )
306