/
/
/
1"""
2End-to-end tests for a queue entering "dynamic mode" through the real queue controller.
3
4Boots a hermetic MusicAssistant (fake `test` provider + demo players) and exercises the
5source-recording and finite->dynamic transition: playing a finite source records it as a source
6even while the queue stays linear, and introducing a dynamic source rebuilds the upcoming tail
7into a bounded, smart-shuffled managed pool over ALL the queue's sources. Shuffle and repeat are
8locked (implicit) while the queue is dynamic.
9"""
10
11from __future__ import annotations
12
13from typing import TYPE_CHECKING, cast
14
15import pytest
16from music_assistant_models.enums import QueueOption, RepeatMode
17from music_assistant_models.errors import InvalidCommand
18from music_assistant_models.media_items import Playlist
19
20from music_assistant.controllers.player_queues import managed_pool
21from music_assistant.controllers.player_queues.constants import MANAGED_POOL_MAX
22from music_assistant.mass import MusicAssistant
23from music_assistant.models.music_provider import MusicProvider
24from music_assistant.providers.radio_playlist import radio_playlist_uri
25
26from .conftest import demo_players, wait_for
27
28if TYPE_CHECKING:
29 from music_assistant_models.media_items import Album, ItemMapping, MediaItemType
30
31
32async def _play_finite_album(mass: MusicAssistant, queue_id: str, album_id: str = "0_0") -> Album:
33 """Play a finite album linearly and wait until it is the active (non-dynamic) queue."""
34 test_prov = cast("MusicProvider", mass.get_provider("test"))
35 assert test_prov is not None
36 album = await test_prov.get_album(album_id) # 20 tracks: <album_id>_0 .. <album_id>_19
37 await mass.player_queues.play_media(queue_id, cast("MediaItemType | ItemMapping | str", album))
38 assert await wait_for(
39 lambda: (q := mass.player_queues.get(queue_id)) is not None and q.current_index is not None
40 ), "album never started playing"
41 return album
42
43
44async def _add_dynamic_playlist(mass: MusicAssistant, queue_id: str, seed_id: str = "4_4_0") -> str:
45 """Add a dynamic radio playlist (seeded from a track) to the queue and return its uri."""
46 test_prov = cast("MusicProvider", mass.get_provider("test"))
47 assert test_prov is not None
48 seed = await test_prov.get_track(seed_id)
49 uri = radio_playlist_uri(seed)
50 await mass.player_queues.play_media(queue_id, uri, option=QueueOption.ADD)
51 return uri
52
53
54@pytest.mark.asyncio
55async def test_finite_source_recorded_even_while_linear(e2e_mass: MusicAssistant) -> None:
56 """Playing a finite album records it as a source, but the queue stays linear (not dynamic)."""
57 queue_id = demo_players(e2e_mass)[0].player_id
58 album = await _play_finite_album(e2e_mass, queue_id)
59
60 queue = e2e_mass.player_queues.get(queue_id)
61 assert queue is not None
62 # a finite-only queue is NOT dynamic even though its source is recorded ...
63 assert queue.is_dynamic is False
64 # ... and the whole album is enqueued linearly (no bounded pool yet)
65 assert len(e2e_mass.player_queues.items(queue_id)) == 20
66 # the finite source is recorded server-side and projected onto the wire `sources`
67 data = e2e_mass.player_queues._queue_data[queue_id]
68 assert [item.uri for item in data.source_items] == [album.uri]
69 assert [mapping.uri for mapping in queue.sources] == [album.uri]
70
71
72@pytest.mark.asyncio
73async def test_source_items_records_finite_and_dynamic(e2e_mass: MusicAssistant) -> None:
74 """Adding a dynamic playlist to a finite queue records BOTH the finite and dynamic sources."""
75 queue_id = demo_players(e2e_mass)[0].player_id
76 album = await _play_finite_album(e2e_mass, queue_id)
77 await _add_dynamic_playlist(e2e_mass, queue_id)
78
79 data = e2e_mass.player_queues._queue_data[queue_id]
80 source_uris = {item.uri for item in data.source_items}
81 # the finite album is still a source, now mixed with the dynamic playlist
82 assert album.uri in source_uris
83 assert any(isinstance(item, Playlist) and item.is_dynamic for item in data.source_items)
84
85
86@pytest.mark.asyncio
87async def test_finite_then_dynamic_rebuilds_tail_into_bounded_pool(
88 e2e_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
89) -> None:
90 """Adding a dynamic source collapses the finite tail into a bounded mix of all sources."""
91 # shrink the pool target so the collapse is visible against a 20-track album
92 monkeypatch.setattr(managed_pool, "MANAGED_POOL_TARGET", 10)
93 queue_id = demo_players(e2e_mass)[0].player_id
94 await _play_finite_album(e2e_mass, queue_id)
95 queue = e2e_mass.player_queues.get(queue_id)
96 assert queue is not None
97 assert queue.current_item is not None
98 playing_id = queue.current_item.media_item.item_id if queue.current_item.media_item else None
99 assert playing_id == "0_0_0"
100
101 await _add_dynamic_playlist(e2e_mass, queue_id)
102
103 queue = e2e_mass.player_queues.get(queue_id)
104 assert queue is not None
105 assert queue.current_index is not None
106 # the queue is now a managed pool ...
107 assert queue.is_dynamic is True
108 items = e2e_mass.player_queues.items(queue_id)
109 tail_ids = [
110 item.media_item.item_id for item in items[queue.current_index + 1 :] if item.media_item
111 ]
112 # ... whose upcoming tail is bounded, not the whole 20-track album left in front of the pool
113 assert 0 < len(tail_ids) <= MANAGED_POOL_MAX
114 all_ids = [item.media_item.item_id for item in items if item.media_item]
115 assert len([tid for tid in all_ids if tid.startswith("0_0_")]) < 20
116 # the current track is kept playing across the rebuild
117 assert queue.current_item is not None
118 assert queue.current_item.media_item is not None
119 assert queue.current_item.media_item.item_id == "0_0_0"
120 # the bounded mix draws from BOTH the finite album (0_0_*) and the dynamic playlist (others)
121 assert any(tid.startswith("0_0_") for tid in tail_ids)
122 assert any(not tid.startswith("0_0_") for tid in tail_ids)
123
124
125@pytest.mark.asyncio
126async def test_entering_dynamic_mode_enables_smart_shuffle(e2e_mass: MusicAssistant) -> None:
127 """Transitioning into dynamic mode auto-enables (implicit) shuffle + smart shuffle."""
128 queue_id = demo_players(e2e_mass)[0].player_id
129 await _play_finite_album(e2e_mass, queue_id)
130 queue = e2e_mass.player_queues.get(queue_id)
131 assert queue is not None
132 # a plain linear album starts with shuffle off and smart shuffle inactive
133 assert queue.shuffle_enabled is False
134 assert queue.smart_shuffle_active is False
135
136 await _add_dynamic_playlist(e2e_mass, queue_id)
137
138 queue = e2e_mass.player_queues.get(queue_id)
139 assert queue is not None
140 assert queue.is_dynamic is True
141 # dynamic mode is an always-on smart mix; the state reflects that
142 assert queue.shuffle_enabled is True
143 assert queue.smart_shuffle_active is True
144
145
146@pytest.mark.asyncio
147async def test_replace_next_dynamic_keeps_current_track_playing(e2e_mass: MusicAssistant) -> None:
148 """REPLACE_NEXT with a dynamic source rebuilds the tail without interrupting the current track."""
149 queue_id = demo_players(e2e_mass)[0].player_id
150 await _play_finite_album(e2e_mass, queue_id)
151 queue = e2e_mass.player_queues.get(queue_id)
152 assert queue is not None
153 assert queue.current_item is not None
154 assert queue.current_item.media_item is not None
155 playing_before = queue.current_item.media_item.item_id
156
157 test_prov = cast("MusicProvider", e2e_mass.get_provider("test"))
158 assert test_prov is not None
159 seed = await test_prov.get_track("4_4_0")
160 await e2e_mass.player_queues.play_media(
161 queue_id, radio_playlist_uri(seed), option=QueueOption.REPLACE_NEXT
162 )
163
164 queue = e2e_mass.player_queues.get(queue_id)
165 assert queue is not None
166 assert queue.is_dynamic is True
167 # the current track keeps playing across the tail rebuild (REPLACE_NEXT does not jump playback)
168 assert queue.current_item is not None
169 assert queue.current_item.media_item is not None
170 assert queue.current_item.media_item.item_id == playing_before
171 # the upcoming tail was replaced by the managed pool
172 assert queue.current_index is not None
173 assert len(e2e_mass.player_queues.items(queue_id)) > queue.current_index + 1
174
175
176@pytest.mark.asyncio
177async def test_shuffle_and_repeat_locked_while_dynamic(e2e_mass: MusicAssistant) -> None:
178 """Shuffle and repeat toggles are rejected while the queue is in dynamic mode."""
179 queue_id = demo_players(e2e_mass)[0].player_id
180 await _play_finite_album(e2e_mass, queue_id)
181 await _add_dynamic_playlist(e2e_mass, queue_id)
182 queue = e2e_mass.player_queues.get(queue_id)
183 assert queue is not None
184 assert queue.is_dynamic is True
185
186 with pytest.raises(InvalidCommand):
187 await e2e_mass.player_queues.set_shuffle(queue_id, False)
188 with pytest.raises(InvalidCommand):
189 await e2e_mass.player_queues.set_repeat(queue_id, RepeatMode.ALL)
190