music-assistant-server

1"""Tests for the album-loudness decision taken while loading a queue item."""
2
3from __future__ import annotations
4
5from typing import cast
6from unittest.mock import AsyncMock, MagicMock
7
8from music_assistant_models.enums import MediaType, RepeatMode
9from music_assistant_models.media_items import (
10    Album,
11    ItemMapping,
12    Playlist,
13    ProviderMapping,
14    Track,
15)
16from music_assistant_models.player_queue import PlayerQueue
17from music_assistant_models.queue_item import QueueItem
18
19from music_assistant.controllers.player_queues.controller import PlayerQueuesController
20from music_assistant.controllers.player_queues.state import PlayerQueueData
21
22QUEUE_ID = "queue-1"
23
24PROVIDER_ALBUM = ItemMapping(
25    media_type=MediaType.ALBUM,
26    item_id="album-prov-1",
27    provider="spotify--abc",
28    name="Kind of Blue",
29)
30OTHER_PROVIDER_ALBUM = ItemMapping(
31    media_type=MediaType.ALBUM,
32    item_id="album-prov-2",
33    provider="spotify--abc",
34    name="Sketches of Spain",
35)
36LIBRARY_ALBUM = Album(
37    item_id="7",
38    provider="library",
39    name="Kind of Blue",
40    provider_mappings={
41        ProviderMapping(
42            item_id="album-prov-1",
43            provider_domain="spotify",
44            provider_instance="spotify--abc",
45        )
46    },
47)
48OTHER_LIBRARY_ALBUM = Album(
49    item_id="8",
50    provider="library",
51    name="Sketches of Spain",
52    provider_mappings={
53        ProviderMapping(
54            item_id="album-prov-2",
55            provider_domain="spotify",
56            provider_instance="spotify--abc",
57        )
58    },
59)
60PLAYLIST = Playlist(
61    item_id="playlist-1",
62    provider="spotify--abc",
63    name="Jazz essentials",
64    provider_mappings={
65        ProviderMapping(
66            item_id="playlist-1",
67            provider_domain="spotify",
68            provider_instance="spotify--abc",
69        )
70    },
71)
72
73
74def _queue_item(item_id: str, album: Album | ItemMapping | None) -> QueueItem:
75    """Build a queue item holding a track on the given album."""
76    return QueueItem(
77        queue_id=QUEUE_ID,
78        queue_item_id=item_id,
79        name=item_id,
80        duration=300,
81        media_item=Track(
82            item_id=item_id,
83            provider="spotify--abc",
84            name=item_id,
85            duration=300,
86            provider_mappings={
87                ProviderMapping(
88                    item_id=item_id,
89                    provider_domain="spotify",
90                    provider_instance="spotify--abc",
91                )
92            },
93            album=album,
94        ),
95    )
96
97
98def _controller(
99    items: list[QueueItem],
100    enqueued: list[Album | Playlist | Track] | None = None,
101    library_album: Album | None = None,
102) -> PlayerQueuesController:
103    """
104    Build a bare controller whose queue holds the given items and enqueued parents.
105
106    :param items: The items the queue holds.
107    :param enqueued: The parent media items the user enqueued on it.
108    :param library_album: The library album the items' own album resolves to while loading.
109    """
110    controller = PlayerQueuesController.__new__(PlayerQueuesController)
111    controller.logger = MagicMock()
112    controller._queue_data = {
113        QUEUE_ID: PlayerQueueData(
114            queue=PlayerQueue(
115                queue_id=QUEUE_ID,
116                active=True,
117                display_name="Test queue",
118                available=True,
119                items=len(items),
120            ),
121            items=items,
122            enqueued_media_items=list(enqueued or []),
123        )
124    }
125    tracks_by_uri = {item.uri: item.media_item for item in items}
126    mass = MagicMock()
127    mass.music.get_library_item_by_prov_id = AsyncMock(
128        side_effect=lambda media_type, *_: library_album if media_type == MediaType.ALBUM else None
129    )
130    mass.music.get_item_by_uri = AsyncMock(side_effect=lambda uri: tracks_by_uri[uri])
131    mass.streams.audio.get_stream_details = AsyncMock(return_value=MagicMock(duration=None))
132    controller.mass = mass
133    return controller
134
135
136async def _prefer_album_loudness(
137    items: list[QueueItem],
138    index: int,
139    enqueued: list[Album | Playlist | Track] | None = None,
140    repeat_mode: RepeatMode = RepeatMode.OFF,
141) -> bool:
142    """Load the item at the given index and read the loudness decision taken for it."""
143    controller = _controller(items, enqueued)
144    controller._queue_data[QUEUE_ID].queue.repeat_mode = repeat_mode
145    await controller._load_item(items[index])
146    get_stream_details = cast("AsyncMock", controller.mass.streams.audio.get_stream_details)
147    return cast("bool", get_stream_details.call_args.kwargs["prefer_album_loudness"])
148
149
150async def test_track_of_an_enqueued_album_uses_album_loudness() -> None:
151    """The tracks of an album the user pressed play on are normalized on the album loudness."""
152    items = [
153        _queue_item("track-1", PROVIDER_ALBUM),
154        _queue_item("track-2", PROVIDER_ALBUM),
155    ]
156    assert await _prefer_album_loudness(items, 0, enqueued=[LIBRARY_ALBUM])
157
158
159async def test_shuffled_album_still_uses_album_loudness() -> None:
160    """An album played on shuffle is still that album, however its tracks end up ordered."""
161    items = [
162        _queue_item("track-1", PROVIDER_ALBUM),
163        _queue_item("track-2", OTHER_PROVIDER_ALBUM),
164        _queue_item("track-3", PROVIDER_ALBUM),
165    ]
166    assert await _prefer_album_loudness(items, 0, enqueued=[LIBRARY_ALBUM])
167
168
169async def test_adjacent_playlist_tracks_of_one_album_use_track_loudness() -> None:
170    """A playlist that happens to place two tracks of one album together is no album play."""
171    items = [
172        _queue_item("track-1", PROVIDER_ALBUM),
173        _queue_item("track-2", PROVIDER_ALBUM),
174    ]
175    assert not await _prefer_album_loudness(items, 0, enqueued=[PLAYLIST])
176
177
178async def test_track_added_beside_an_enqueued_album_uses_track_loudness() -> None:
179    """On a mixed queue only the enqueued album's own tracks play as part of an album."""
180    items = [
181        _queue_item("track-1", PROVIDER_ALBUM),
182        _queue_item("track-2", OTHER_PROVIDER_ALBUM),
183    ]
184    enqueued: list[Album | Playlist | Track] = [
185        LIBRARY_ALBUM,
186        cast("Track", items[1].media_item),
187    ]
188    assert await _prefer_album_loudness(items, 0, enqueued=enqueued)
189    assert not await _prefer_album_loudness(items, 1, enqueued=enqueued)
190
191
192async def test_a_different_enqueued_album_does_not_apply() -> None:
193    """Only the album a track actually belongs to counts, not any album on the queue."""
194    items = [_queue_item("track-1", PROVIDER_ALBUM)]
195    assert not await _prefer_album_loudness(items, 0, enqueued=[OTHER_LIBRARY_ALBUM])
196
197
198async def test_queue_without_an_enqueued_album_uses_track_loudness() -> None:
199    """A queue that records no album parent (a browsed folder, a restored queue) is no album play."""
200    items = [
201        _queue_item("track-1", PROVIDER_ALBUM),
202        _queue_item("track-2", PROVIDER_ALBUM),
203    ]
204    assert not await _prefer_album_loudness(items, 0)
205
206
207async def test_library_album_enqueued_matches_the_provider_album_on_the_item() -> None:
208    """The enqueued album and the queue's tracks may hold different shapes of the same album."""
209    items = [_queue_item("track-1", PROVIDER_ALBUM)]
210    assert await _prefer_album_loudness(items, 0, enqueued=[LIBRARY_ALBUM])
211
212
213async def test_provider_album_enqueued_matches_the_library_album_on_the_item() -> None:
214    """The same album seen from both representations is recognised the other way around too."""
215    items = [_queue_item("track-1", LIBRARY_ALBUM)]
216    provider_album = Album(
217        item_id="album-prov-1",
218        provider="spotify--abc",
219        name="Kind of Blue",
220        provider_mappings={
221            ProviderMapping(
222                item_id="album-prov-1",
223                provider_domain="spotify",
224                provider_instance="spotify--abc",
225            )
226        },
227    )
228    assert await _prefer_album_loudness(items, 0, enqueued=[provider_album])
229
230
231async def test_item_without_an_album_uses_track_loudness() -> None:
232    """An item that carries no album at all has no album loudness to prefer."""
233    items = [_queue_item("track-1", None)]
234    assert not await _prefer_album_loudness(items, 0, enqueued=[LIBRARY_ALBUM])
235
236
237async def test_repeat_single_ignores_the_enqueued_album() -> None:
238    """A track repeating on its own is not played as part of the album it was enqueued with."""
239    items = [
240        _queue_item("track-1", PROVIDER_ALBUM),
241        _queue_item("track-2", PROVIDER_ALBUM),
242    ]
243    assert not await _prefer_album_loudness(
244        items, 0, enqueued=[LIBRARY_ALBUM], repeat_mode=RepeatMode.ONE
245    )
246
247
248async def test_next_item_is_loaded_with_the_album_decision() -> None:
249    """The preload of the item that plays next takes the same decision as the current one."""
250    items = [
251        _queue_item("track-1", PROVIDER_ALBUM),
252        _queue_item("track-2", PROVIDER_ALBUM),
253    ]
254    controller = _controller(items, enqueued=[LIBRARY_ALBUM])
255    await controller.load_next_queue_item(QUEUE_ID, items[0].queue_item_id)
256    get_stream_details = cast("AsyncMock", controller.mass.streams.audio.get_stream_details)
257    assert get_stream_details.call_args.kwargs["prefer_album_loudness"]
258
259
260async def test_enqueued_album_survives_a_restart() -> None:
261    """
262    An album queue restored from cache still plays as an album.
263
264    The decision reads the enqueued parent, which is only recognised as an album while it
265    round-trips as one; a mapping-shaped restore would silently drop to track loudness.
266    """
267    items = [_queue_item("track-1", PROVIDER_ALBUM)]
268    controller = _controller(items, enqueued=[LIBRARY_ALBUM])
269    queue_data = controller._queue_data[QUEUE_ID]
270    restored = PlayerQueueData.from_cache(queue_data.to_cache(), queue_data.items_to_cache())
271
272    controller._queue_data[QUEUE_ID] = restored
273    await controller._load_item(restored.items[0])
274
275    get_stream_details = cast("AsyncMock", controller.mass.streams.audio.get_stream_details)
276    assert get_stream_details.call_args.kwargs["prefer_album_loudness"]
277
278
279async def test_enqueued_provider_album_matches_an_items_slim_library_album() -> None:
280    """
281    A queue item may hold only a slim mapping of its album until it is loaded.
282
283    That mapping carries no provider ids of its own, so it can only be matched against the
284    album the user enqueued once loading resolved it to the full library album.
285    """
286    library_album_mapping = ItemMapping(
287        media_type=MediaType.ALBUM,
288        item_id="7",
289        provider="library",
290        name="Kind of Blue",
291    )
292    provider_album = Album(
293        item_id="album-prov-1",
294        provider="spotify--abc",
295        name="Kind of Blue",
296        provider_mappings={
297            ProviderMapping(
298                item_id="album-prov-1",
299                provider_domain="spotify",
300                provider_instance="spotify--abc",
301            )
302        },
303    )
304    items = [_queue_item("track-1", library_album_mapping)]
305    # loading resolves that slim mapping to the full library album, which does carry them
306    controller = _controller(items, enqueued=[provider_album], library_album=LIBRARY_ALBUM)
307
308    await controller._load_item(items[0])
309
310    get_stream_details = cast("AsyncMock", controller.mass.streams.audio.get_stream_details)
311    assert get_stream_details.call_args.kwargs["prefer_album_loudness"]
312