music-assistant-server

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