/
/
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 Album, ItemMapping, ProviderMapping, Track
10from music_assistant_models.player_queue import PlayerQueue
11from music_assistant_models.queue_item import QueueItem
12
13from music_assistant.controllers.player_queues.controller import PlayerQueuesController
14from music_assistant.controllers.player_queues.state import PlayerQueueData
15
16QUEUE_ID = "queue-1"
17
18PROVIDER_ALBUM = ItemMapping(
19 media_type=MediaType.ALBUM,
20 item_id="album-prov-1",
21 provider="spotify--abc",
22 name="Kind of Blue",
23)
24OTHER_PROVIDER_ALBUM = ItemMapping(
25 media_type=MediaType.ALBUM,
26 item_id="album-prov-2",
27 provider="spotify--abc",
28 name="Sketches of Spain",
29)
30LIBRARY_ALBUM = Album(
31 item_id="7",
32 provider="library",
33 name="Kind of Blue",
34 provider_mappings={
35 ProviderMapping(
36 item_id="album-prov-1",
37 provider_domain="spotify",
38 provider_instance="spotify--abc",
39 )
40 },
41)
42
43
44def _queue_item(item_id: str, album: Album | ItemMapping) -> QueueItem:
45 """Build a queue item holding a track on the given album."""
46 return QueueItem(
47 queue_id=QUEUE_ID,
48 queue_item_id=item_id,
49 name=item_id,
50 duration=300,
51 media_item=Track(
52 item_id=item_id,
53 provider="spotify--abc",
54 name=item_id,
55 duration=300,
56 provider_mappings={
57 ProviderMapping(
58 item_id=item_id,
59 provider_domain="spotify",
60 provider_instance="spotify--abc",
61 )
62 },
63 album=album,
64 ),
65 )
66
67
68def _controller(items: list[QueueItem]) -> PlayerQueuesController:
69 """Build a bare controller whose queue holds the given items."""
70 controller = PlayerQueuesController.__new__(PlayerQueuesController)
71 controller.logger = MagicMock()
72 controller._queue_data = {
73 QUEUE_ID: PlayerQueueData(
74 queue=PlayerQueue(
75 queue_id=QUEUE_ID,
76 active=True,
77 display_name="Test queue",
78 available=True,
79 items=len(items),
80 ),
81 items=items,
82 )
83 }
84 tracks_by_uri = {item.uri: item.media_item for item in items}
85 mass = MagicMock()
86 mass.music.get_library_item_by_prov_id = AsyncMock(return_value=None)
87 mass.music.get_item_by_uri = AsyncMock(side_effect=lambda uri: tracks_by_uri[uri])
88 mass.streams.audio.get_stream_details = AsyncMock(return_value=MagicMock(duration=None))
89 controller.mass = mass
90 return controller
91
92
93async def _prefer_album_loudness(
94 items: list[QueueItem], current_index: int, repeat_mode: RepeatMode = RepeatMode.OFF
95) -> bool:
96 """Load the item after the given index the way a player asks for it, and read the decision."""
97 controller = _controller(items)
98 controller._queue_data[QUEUE_ID].queue.repeat_mode = repeat_mode
99 await controller.load_next_queue_item(QUEUE_ID, items[current_index].queue_item_id)
100 get_stream_details = cast("AsyncMock", controller.mass.streams.audio.get_stream_details)
101 return cast("bool", get_stream_details.call_args.kwargs["prefer_album_loudness"])
102
103
104async def test_standalone_track_uses_track_loudness() -> None:
105 """A track surrounded by other albums is normalized on its own loudness."""
106 items = [
107 _queue_item("track-1", OTHER_PROVIDER_ALBUM),
108 _queue_item("track-2", PROVIDER_ALBUM),
109 ]
110 assert not await _prefer_album_loudness(items, current_index=0)
111
112
113async def test_second_track_of_an_album_uses_album_loudness() -> None:
114 """The track right after the first one of an album is still part of that album."""
115 items = [
116 _queue_item("track-1", PROVIDER_ALBUM),
117 _queue_item("track-2", PROVIDER_ALBUM),
118 ]
119 assert await _prefer_album_loudness(items, current_index=0)
120
121
122async def test_library_album_matches_provider_album_of_previous_item() -> None:
123 """
124 A loaded previous item carries the library album while the item being loaded has the provider one.
125
126 Both describe the same album, so the album loudness applies.
127 """
128 items = [
129 _queue_item("track-1", LIBRARY_ALBUM),
130 _queue_item("track-2", PROVIDER_ALBUM),
131 ]
132 assert await _prefer_album_loudness(items, current_index=0)
133
134
135async def test_library_album_matches_provider_album_of_next_item() -> None:
136 """The same album seen from both representations is recognised on the next-item side too."""
137 items = [
138 _queue_item("track-1", OTHER_PROVIDER_ALBUM),
139 _queue_item("track-2", PROVIDER_ALBUM),
140 _queue_item("track-3", LIBRARY_ALBUM),
141 ]
142 assert await _prefer_album_loudness(items, current_index=0)
143
144
145async def test_different_albums_use_track_loudness() -> None:
146 """Neighbouring tracks from genuinely different albums do not form an album."""
147 items = [
148 _queue_item("track-1", PROVIDER_ALBUM),
149 _queue_item("track-2", OTHER_PROVIDER_ALBUM),
150 _queue_item("track-3", PROVIDER_ALBUM),
151 ]
152 assert not await _prefer_album_loudness(items, current_index=0)
153
154
155async def test_first_item_of_the_queue_has_no_previous_item() -> None:
156 """Starting at the first item, nothing plays before it, least of all the end of the queue."""
157 items = [
158 _queue_item("track-1", OTHER_PROVIDER_ALBUM),
159 _queue_item("track-2", PROVIDER_ALBUM),
160 _queue_item("track-3", OTHER_PROVIDER_ALBUM),
161 ]
162 controller = _controller(items)
163 await controller._load_item(items[0], next_index=1)
164 get_stream_details = cast("AsyncMock", controller.mass.streams.audio.get_stream_details)
165 assert not get_stream_details.call_args.kwargs["prefer_album_loudness"]
166
167
168async def test_track_on_repeat_single_is_not_an_album() -> None:
169 """A track repeating on its own is not played as part of its album."""
170 items = [_queue_item("track-1", PROVIDER_ALBUM)]
171 assert not await _prefer_album_loudness(items, current_index=0, repeat_mode=RepeatMode.ONE)
172
173
174async def test_repeat_single_ignores_the_album_around_it() -> None:
175 """Repeating one track of an album does not play the rest of that album."""
176 items = [
177 _queue_item("track-1", PROVIDER_ALBUM),
178 _queue_item("track-2", PROVIDER_ALBUM),
179 _queue_item("track-3", PROVIDER_ALBUM),
180 ]
181 assert not await _prefer_album_loudness(items, current_index=1, repeat_mode=RepeatMode.ONE)
182
183
184async def test_single_track_on_repeat_all_is_not_an_album() -> None:
185 """Repeating a one-item queue wraps onto the item itself, which is no album."""
186 items = [_queue_item("track-1", PROVIDER_ALBUM)]
187 assert not await _prefer_album_loudness(items, current_index=0, repeat_mode=RepeatMode.ALL)
188