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
8from music_assistant_models.enums import MediaType
9
10from music_assistant.constants import DB_TABLE_PLAYLOG
11from music_assistant.controllers.music.recommendations.library import (
12    library_items,
13    library_rows,
14)
15from music_assistant.mass import MusicAssistant
16
17if TYPE_CHECKING:
18    import pytest
19    from music_assistant_models.media_items import ItemMapping
20
21EXPECTED_DEFAULT_ORDER = [
22    "in_progress",
23    "recently_played",
24    "recently_added_tracks",
25    "recently_added_albums",
26    "random_artists",
27    "random_albums",
28    "recent_favorite_tracks",
29    "favorite_playlists",
30    "favorite_radio",
31    "recent_artists",
32    "recent_tracks",
33]
34
35
36async def test_default_recommendations_order(mass: MusicAssistant) -> None:
37    """The default library rows appear in their canonical order."""
38    folders = await mass.music.recommendations.get_recommendations()
39    defaults = [f.item_id for f in folders if f.item_id in EXPECTED_DEFAULT_ORDER]
40    assert defaults == EXPECTED_DEFAULT_ORDER
41
42
43async def test_recommendations_rows_have_no_items(mass: MusicAssistant) -> None:
44    """The rows listing returns descriptors only; no row carries items."""
45    await _add_playlog_row(
46        mass, item_id="album-1", media_type=MediaType.ALBUM, timestamp=2000, user_initiated=True
47    )
48    folders = await mass.music.recommendations.get_recommendations()
49    assert folders
50    assert all(folder.items == [] for folder in folders)
51
52
53async def test_library_rows_have_descriptor_fields() -> None:
54    """Library rows carry their identity fields, correct defaults, and no items."""
55    rows = library_rows()
56    in_progress = next(f for f in rows if f.item_id == "in_progress")
57    assert in_progress.provider == "library"
58    assert in_progress.name == "In progress"
59    assert in_progress.translation_key == "in_progress_items"
60    assert in_progress.icon == "mdi-motion-play"
61    assert in_progress.enabled_by_default is True
62    random_artists = next(f for f in rows if f.item_id == "random_artists")
63    assert random_artists.enabled_by_default is False
64    assert all(folder.items == [] for folder in rows)
65
66
67async def test_recently_played_rolls_up_to_container(mass: MusicAssistant) -> None:
68    """Playing an album shows the album, not its individual tracks."""
69    await _add_playlog_row(
70        mass, item_id="album-1", media_type=MediaType.ALBUM, timestamp=2000, user_initiated=True
71    )
72    await _add_playlog_row(
73        mass, item_id="track-1", media_type=MediaType.TRACK, timestamp=1999, user_initiated=False
74    )
75    await _add_playlog_row(
76        mass,
77        item_id="track-direct",
78        media_type=MediaType.TRACK,
79        timestamp=2001,
80        user_initiated=True,
81    )
82    items = await mass.music.recommendations.get_recommendation_items("library", "recently_played")
83    item_ids = {item.item_id for item in items}
84    assert "album-1" in item_ids
85    assert "track-1" not in item_ids
86    assert "track-direct" in item_ids
87
88
89async def test_recent_artists_and_tracks_rows_present(mass: MusicAssistant) -> None:
90    """Recent Artists shows played artists; Recent Tracks shows played tracks."""
91    await _add_playlog_row(
92        mass, item_id="artist-1", media_type=MediaType.ARTIST, timestamp=3000, user_initiated=True
93    )
94    await _add_playlog_row(
95        mass, item_id="track-9", media_type=MediaType.TRACK, timestamp=2999, user_initiated=False
96    )
97
98    artist_items = await mass.music.recommendations.get_recommendation_items(
99        "library", "recent_artists"
100    )
101    track_items = await mass.music.recommendations.get_recommendation_items(
102        "library", "recent_tracks"
103    )
104
105    assert "artist-1" in {item.item_id for item in artist_items}
106    assert "artist-1" not in {item.item_id for item in track_items}
107    assert "track-9" in {item.item_id for item in track_items}
108    assert "track-9" not in {item.item_id for item in artist_items}
109
110
111async def test_recently_played_includes_podcast_and_audiobook_containers(
112    mass: MusicAssistant,
113) -> None:
114    """Recently Played includes podcast/audiobook containers but excludes episodes and non-user-initiated tracks."""
115    await _add_playlog_row(
116        mass, item_id="album-x", media_type=MediaType.ALBUM, timestamp=3000, user_initiated=True
117    )
118    await _add_playlog_row(
119        mass,
120        item_id="podcast-x",
121        media_type=MediaType.PODCAST,
122        timestamp=3001,
123        user_initiated=False,
124    )
125    await _add_playlog_row(
126        mass,
127        item_id="audiobook-x",
128        media_type=MediaType.AUDIOBOOK,
129        timestamp=3002,
130        user_initiated=False,
131    )
132    await _add_playlog_row(
133        mass,
134        item_id="episode-x",
135        media_type=MediaType.PODCAST_EPISODE,
136        timestamp=3003,
137        user_initiated=False,
138    )
139    await _add_playlog_row(
140        mass,
141        item_id="loose-track",
142        media_type=MediaType.TRACK,
143        timestamp=2999,
144        user_initiated=False,
145    )
146    items = await mass.music.recommendations.get_recommendation_items("library", "recently_played")
147    item_ids = {item.item_id for item in items}
148    assert "album-x" in item_ids, "album (user-initiated) should appear"
149    assert "podcast-x" in item_ids, "podcast show should always appear"
150    assert "audiobook-x" in item_ids, "audiobook should always appear"
151    assert "episode-x" not in item_ids, "podcast episode should not appear"
152    assert "loose-track" not in item_ids, "non-user-initiated track should be filtered out"
153
154
155async def test_recently_played_always_include_media_types_query(mass: MusicAssistant) -> None:
156    """always_include_media_types OR-s in those types regardless of user_initiated_only."""
157    await _add_playlog_row(
158        mass,
159        item_id="podcast-q",
160        media_type=MediaType.PODCAST,
161        timestamp=5000,
162        user_initiated=False,
163    )
164    await _add_playlog_row(
165        mass,
166        item_id="track-q",
167        media_type=MediaType.TRACK,
168        timestamp=4999,
169        user_initiated=False,
170    )
171    results = await mass.music.recently_played(
172        media_types=[MediaType.TRACK],
173        user_initiated_only=True,
174        always_include_media_types=[MediaType.PODCAST],
175    )
176    result_ids = {item.item_id for item in results}
177    assert "podcast-q" in result_ids, "podcast should be returned via always_include_media_types"
178    assert "track-q" not in result_ids, "non-user-initiated track should be excluded"
179
180
181async def test_every_library_row_dispatches_a_query() -> None:
182    """
183    Every id listed by library_rows() reaches a real query branch in library_items().
184
185    The rows listing and the items dispatch live in two separate functions; this
186    pins that no listed row silently falls through to the empty default arm.
187    """
188    mass = AsyncMock()
189    for folder in library_rows():
190        mass.reset_mock()
191        await library_items(mass, folder.item_id)
192        assert mass.method_calls, f"row {folder.item_id!r} did not dispatch a library query"
193
194
195async def test_library_rows_listed_by_controller(mass: MusicAssistant) -> None:
196    """Every library row appears in the controller's rows listing."""
197    folders = await mass.music.recommendations.get_recommendations()
198    listed = {f.item_id for f in folders if f.provider == "library"}
199    assert {f.item_id for f in library_rows()} <= listed
200
201
202async def test_unknown_library_row_returns_empty(mass: MusicAssistant) -> None:
203    """Requesting items for an unknown builtin row returns an empty list."""
204    items = await mass.music.recommendations.get_recommendation_items("library", "no_such_row")
205    assert items == []
206
207
208async def test_failing_library_row_items_isolated(
209    mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
210) -> None:
211    """A library row whose items query raises returns an empty list, not an error."""
212
213    async def _boom(**_kwargs: object) -> list[ItemMapping]:
214        raise RuntimeError("row boom")
215
216    monkeypatch.setattr(mass.music, "in_progress_items", _boom)
217    items = await mass.music.recommendations.get_recommendation_items("library", "in_progress")
218    assert items == []
219
220
221async def _add_playlog_row(
222    mass: MusicAssistant,
223    *,
224    item_id: str,
225    media_type: MediaType,
226    timestamp: int,
227    user_initiated: bool,
228    userid: str = "user-a",
229) -> None:
230    await mass.music.database.insert(
231        DB_TABLE_PLAYLOG,
232        {
233            "item_id": item_id,
234            "provider": "library",
235            "media_type": media_type.value,
236            "name": f"{media_type.value} {item_id}",
237            "timestamp": timestamp,
238            "fully_played": True,
239            "seconds_played": 180,
240            "userid": userid,
241            "user_initiated": user_initiated,
242        },
243    )
244