/
/
/
1"""Unit tests for the AI Radio sticky queue DJ state."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from collections.abc import Callable, Coroutine
8from pathlib import Path
9from types import SimpleNamespace
10from typing import Any, cast
11
12import pytest
13from music_assistant_models.enums import EventType, MediaType
14from music_assistant_models.errors import InvalidDataError, MusicAssistantError
15
16from music_assistant.providers.ai_radio.constants import (
17 ATTR_GAP_NEXT_ID,
18 ATTR_HOST_ID,
19 ATTR_QUEUE_DJ,
20 ATTR_SESSION_ID,
21)
22from music_assistant.providers.ai_radio.models import PlannedSection, SessionState
23from music_assistant.providers.ai_radio.queue_dj import AIRadioQueueDJMixin
24from music_assistant.providers.ai_radio.runtime import AIRadioRuntimeMixin
25from music_assistant.providers.ai_radio.storage import AIRadioStorageMixin
26
27
28class FakeQueue:
29 """Minimal PlayerQueue stand-in."""
30
31 def __init__(
32 self, queue_id: str, current_index: int | None, index_in_buffer: int | None
33 ) -> None:
34 """Initialize the fake queue with its playback pointers."""
35 self.queue_id = queue_id
36 self.current_index = current_index
37 self.index_in_buffer = index_in_buffer
38
39
40class FakeQueueItem:
41 """Minimal QueueItem stand-in."""
42
43 _counter = 0
44
45 def __init__(
46 self, name: str, duration: int | None = 200, extra: dict[str, Any] | None = None
47 ) -> None:
48 """Initialize the fake queue item with a unique id."""
49 FakeQueueItem._counter += 1
50 self.queue_item_id = f"qi{FakeQueueItem._counter}"
51 # mirrors the insert order a real queue restores when it is un-shuffled
52 self.sort_index = FakeQueueItem._counter
53 self.name = name
54 self.duration = duration
55 self.media_item = None
56 self.extra_attributes: dict[str, Any] = dict(extra or {})
57
58
59class FakePlayerQueues:
60 """Minimal PlayerQueues controller stand-in."""
61
62 def __init__(self, queue: FakeQueue, items: list[FakeQueueItem]) -> None:
63 """Initialize with one queue and its items."""
64 self._queue = queue
65 self._items = items
66 # what each update_items call added, as (items, index), and what it dropped. the real
67 # controller only ever replaces the whole list, so both are derived from the diff
68 self.loads: list[tuple[list[Any], int]] = []
69 self.deleted: list[str] = []
70 self.update_calls = 0
71 # stands in for the QUEUE_ITEMS_UPDATED event a real update_items emits
72 self.on_items_updated: Callable[[], None] | None = None
73
74 def get(self, queue_id: str) -> FakeQueue | None:
75 """Return the queue when the id matches."""
76 return self._queue if queue_id == self._queue.queue_id else None
77
78 def items(self, queue_id: str, limit: int = 500, offset: int = 0) -> list[Any]:
79 """Return one page of queue items."""
80 return self._items[offset : offset + limit]
81
82 def update_items(self, queue_id: str, queue_items: list[Any]) -> None:
83 """Replace the queue items with the given list."""
84 self.update_calls += 1
85 previous_ids = {item.queue_item_id for item in self._items}
86 current_ids = {item.queue_item_id for item in queue_items}
87 self.deleted.extend(
88 item.queue_item_id for item in self._items if item.queue_item_id not in current_ids
89 )
90 self.loads.extend(
91 ([item], index)
92 for index, item in enumerate(queue_items)
93 if item.queue_item_id not in previous_ids
94 )
95 self._items = queue_items
96 if self.on_items_updated is not None:
97 self.on_items_updated()
98
99
100class FakeMass:
101 """Minimal MusicAssistant stand-in for the queue DJ mixin."""
102
103 def __init__(self, player_queues: FakePlayerQueues) -> None:
104 """Initialize with the fake player queues controller."""
105 self.player_queues = player_queues
106 self.tasks: list[asyncio.Task[Any]] = []
107 self._tasks_by_id: dict[str, asyncio.Task[Any]] = {}
108
109 def create_task(
110 self, target: Coroutine[Any, Any, Any], task_id: str | None = None
111 ) -> asyncio.Task[Any]:
112 """Run the given coroutine as a task, deduplicating on task id like mass does."""
113 if task_id and (existing := self._tasks_by_id.get(task_id)) and not existing.done():
114 target.close()
115 return existing
116 task = asyncio.ensure_future(target)
117 self.tasks.append(task)
118 if task_id:
119 self._tasks_by_id[task_id] = task
120 return task
121
122 def subscribe(self, *args: Any, **kwargs: Any) -> Any:
123 """Return a no-op unsubscribe callback."""
124 return lambda: None
125
126
127class StubConfig:
128 """Minimal ProviderConfig stand-in exposing get_value."""
129
130 def get_value(self, key: str, default: Any = None) -> Any:
131 """Return the default for every config key."""
132 return default
133
134
135class DummyQueueDJ(AIRadioQueueDJMixin, AIRadioStorageMixin):
136 """Minimal harness for queue DJ state tests."""
137
138 instance_id = "ai_radio_test"
139
140 def __init__(self, tmp_path: Path) -> None:
141 """Initialize dummy mixin state."""
142 self.logger = logging.getLogger(__name__)
143 self._hosts: dict[str, dict[str, Any]] = {
144 "rick": {"id": "rick", "name": "Rick", "instructions": "x", "tts_engine": ""},
145 }
146 self._dj_queues: dict[str, Any] = {}
147 self._dj_file = tmp_path / "queue_dj.json"
148 self._dj_lock = asyncio.Lock()
149 # the disable path reaches the real clip cleanup, which needs a queue layer;
150 # this one holds no queue at all, so cleanup finds nothing to do
151 # the mixin declares `mass: MusicAssistant`; a fake stands in for tests, so its own
152 # attributes (not MusicAssistant's) are what callers here actually see
153 self.mass: FakeMass = FakeMass( # type: ignore[assignment]
154 FakePlayerQueues(FakeQueue("other-queue", None, None), [])
155 )
156
157 def _schedule_replan(self, queue_id: str) -> None:
158 """Record replan requests instead of running them."""
159 self.replanned = getattr(self, "replanned", [])
160 self.replanned.append(queue_id)
161
162
163class ReplanQueueDJ(AIRadioRuntimeMixin, AIRadioQueueDJMixin, AIRadioStorageMixin):
164 """Harness combining the queue DJ mixin with the real planner and clip builder."""
165
166 instance_id = "ai_radio_test"
167 domain = "ai_radio"
168
169 def __init__(
170 self,
171 tmp_path: Path,
172 queue: FakeQueue,
173 items: list[FakeQueueItem],
174 host: dict[str, Any],
175 ) -> None:
176 """Initialize the harness around one fake queue."""
177 self.logger = logging.getLogger(__name__)
178 self.config = cast("Any", StubConfig())
179 self._sections = {_transition_section()["id"]: _transition_section()}
180 self._sessions: dict[str, SessionState] = {}
181 self._hosts: dict[str, dict[str, Any]] = {host["id"]: host}
182 self._dj_queues: dict[str, Any] = {}
183 self._dj_file = tmp_path / "queue_dj.json"
184 self._dj_lock = asyncio.Lock()
185 self._unloading = False
186 self.player_queues = FakePlayerQueues(queue, items)
187 # see DummyQueueDJ.__init__ for why this needs its own annotation
188 self.mass: FakeMass = FakeMass(self.player_queues) # type: ignore[assignment]
189
190
191def _transition_section() -> dict[str, Any]:
192 """Return the single shared section used by the queue DJ test hosts."""
193 return {
194 "id": "Song_Transition",
195 "name": "Song Transition",
196 "type": "ai_text",
197 "web_search": "disabled",
198 "prompt": "From <prev_songinfo> to <next_songinfo>",
199 "constraints": {"max_chars": 200},
200 }
201
202
203def _must_host() -> dict[str, Any]:
204 """Return a host that always plans a section between songs."""
205 return {
206 "id": "rick",
207 "name": "Rick",
208 "instructions": "keep it short",
209 "tts_engine": "",
210 "section_ids": ["Song_Transition"],
211 "section_order": [{"when": "between_songs", "flow": [{"MUST": "Song_Transition"}]}],
212 "merge_section_id": "",
213 }
214
215
216def _optional_host(min_gap_songs: int) -> dict[str, Any]:
217 """Return a host whose section is certain to fire but guarded by a song gap."""
218 host = _must_host()
219 host["section_order"] = [
220 {
221 "when": "between_songs",
222 "flow": [
223 {
224 "OPTIONAL": {
225 "section": "Song_Transition",
226 "chance": 1.0,
227 "guards": {"min_gap_songs": min_gap_songs},
228 }
229 }
230 ],
231 }
232 ]
233 return host
234
235
236def _hourly_sections() -> list[dict[str, Any]]:
237 """Return two once-per-hour sections plus the ai_meta section that merges them."""
238 return [
239 {
240 "id": "Weather",
241 "name": "Weather",
242 "type": "ai_text",
243 "web_search": "disabled",
244 "prompt": "Give the forecast",
245 "constraints": {"max_chars": 200},
246 },
247 {
248 "id": "News",
249 "name": "News",
250 "type": "ai_text",
251 "web_search": "disabled",
252 "prompt": "Give the headlines",
253 "constraints": {"max_chars": 200},
254 },
255 {
256 "id": "Smoother",
257 "name": "Between Songs Mix",
258 "type": "ai_meta",
259 "prompt": "Combine these: <section_drafts>",
260 },
261 ]
262
263
264def _hourly_host() -> dict[str, Any]:
265 """Return a host whose two hourly sections merge into a single clip per gap."""
266 host = _must_host()
267 host["section_ids"] = ["Weather", "News", "Smoother"]
268 host["section_order"] = [
269 {
270 "when": "between_songs",
271 "flow": [
272 {
273 "OPTIONAL": {
274 "section": "Weather",
275 "chance": 1.0,
276 "guards": {"max_per_60min": 1},
277 }
278 },
279 {
280 "OPTIONAL": {"section": "News", "chance": 1.0, "guards": {"max_per_60min": 1}},
281 },
282 ],
283 }
284 ]
285 host["merge_section_id"] = "Smoother"
286 return host
287
288
289def _track(index: int) -> FakeQueueItem:
290 """Return a fake music queue item."""
291 return FakeQueueItem(f"Artist {index} - Song {index}")
292
293
294def _dj_clip(gap_next_id: str, session_id: str) -> FakeQueueItem:
295 """Return a fake DJ clip queue item announcing the given track."""
296 return FakeQueueItem(
297 "Song Transition",
298 duration=30,
299 extra={
300 ATTR_QUEUE_DJ: True,
301 ATTR_GAP_NEXT_ID: gap_next_id,
302 ATTR_SESSION_ID: session_id,
303 },
304 )
305
306
307def _make_replan_dj(
308 tmp_path: Path,
309 items: list[FakeQueueItem],
310 current_index: int | None = 0,
311 index_in_buffer: int | None = 0,
312 host: dict[str, Any] | None = None,
313) -> ReplanQueueDJ:
314 """Build an armed replan harness around the given queue items."""
315 queue = FakeQueue("queue-1", current_index, index_in_buffer)
316 dummy = ReplanQueueDJ(tmp_path, queue, items, host or _must_host())
317 # set_queue_dj marks a state plannable once its clip cleanup ran, and these harnesses
318 # start from a queue whose switch already settled
319 dummy._arm_dj_state("queue-1", "rick").ready = True
320 return dummy
321
322
323async def test_set_queue_dj_enables_and_persists(tmp_path: Path) -> None:
324 """Arm a queue DJ, persist it, and reload it into a fresh instance."""
325 dummy = DummyQueueDJ(tmp_path)
326 mapping = await dummy.set_queue_dj("queue-1", "rick")
327 assert mapping == {"queue-1": "rick"}
328 assert dummy._dj_queues["queue-1"].host_id == "rick"
329 assert dummy._dj_queues["queue-1"].dj_session_id
330 assert dummy._dj_queues["queue-1"].ready is True
331 assert dummy.replanned == ["queue-1"]
332 assert dummy._dj_file.exists()
333
334 fresh = DummyQueueDJ(tmp_path)
335 await fresh._load_queue_dj()
336 assert fresh._dj_queues["queue-1"].host_id == "rick"
337 # nothing has to be cleaned up before a boot arm, so it may plan on the first event
338 assert fresh._dj_queues["queue-1"].ready is True
339
340
341async def test_set_queue_dj_rejects_unknown_host(tmp_path: Path) -> None:
342 """Reject arming a queue DJ with an unknown host id."""
343 dummy = DummyQueueDJ(tmp_path)
344 with pytest.raises(InvalidDataError):
345 await dummy.set_queue_dj("queue-1", "nobody")
346
347
348async def test_set_queue_dj_none_disables(tmp_path: Path) -> None:
349 """Disable an armed queue DJ by passing host_id=None."""
350 dummy = DummyQueueDJ(tmp_path)
351 await dummy.set_queue_dj("queue-1", "rick")
352 mapping = await dummy.set_queue_dj("queue-1", None)
353 assert mapping == {}
354 assert dummy._dj_queues == {}
355
356
357async def test_status_returns_mapping(tmp_path: Path) -> None:
358 """Return the queue-to-host mapping for an armed queue DJ."""
359 dummy = DummyQueueDJ(tmp_path)
360 await dummy.set_queue_dj("queue-1", "rick")
361 assert await dummy.get_queue_dj_status() == {"queue-1": "rick"}
362
363
364async def test_replan_inserts_clip_between_upcoming_tracks(tmp_path: Path) -> None:
365 """Inject one DJ clip into every plannable gap ahead of the playback guards."""
366 tracks = [_track(index) for index in range(4)]
367 dummy = _make_replan_dj(tmp_path, list(tracks))
368
369 await dummy._replan_queue("queue-1")
370
371 queues = dummy.player_queues
372 assert len(queues.loads) == 2
373 for queue_items, insert_at_index in queues.loads:
374 assert insert_at_index > 1
375 clip = queue_items[0]
376 assert clip.extra_attributes[ATTR_QUEUE_DJ] is True
377 assert clip.extra_attributes[ATTR_HOST_ID] == "rick"
378 announced = {items[0].extra_attributes[ATTR_GAP_NEXT_ID] for items, _ in queues.loads}
379 assert announced == {tracks[2].queue_item_id, tracks[3].queue_item_id}
380
381 final_items = queues.items("queue-1")
382 for index, item in enumerate(final_items):
383 if item.extra_attributes.get(ATTR_QUEUE_DJ):
384 successor = final_items[index + 1]
385 assert successor.queue_item_id == item.extra_attributes[ATTR_GAP_NEXT_ID]
386 state = dummy._dj_queues["queue-1"]
387 assert state.clip_counter == 2
388 assert state.decided_gap_ids == {tracks[2].queue_item_id, tracks[3].queue_item_id}
389 assert state.songs_before_window == 1
390
391
392async def test_replan_skips_gaps_that_already_have_a_clip(tmp_path: Path) -> None:
393 """Leave a gap alone when it already holds a DJ clip for the following track."""
394 tracks = [_track(index) for index in range(4)]
395 dummy = _make_replan_dj(tmp_path, list(tracks))
396 state = dummy._dj_queues["queue-1"]
397 existing = _dj_clip(tracks[2].queue_item_id, state.dj_session_id)
398 dummy.player_queues._items = [tracks[0], tracks[1], existing, tracks[2], tracks[3]]
399
400 await dummy._replan_queue("queue-1")
401
402 queues = dummy.player_queues
403 assert queues.deleted == []
404 assert len(queues.loads) == 1
405 assert queues.loads[0][0][0].extra_attributes[ATTR_GAP_NEXT_ID] == tracks[3].queue_item_id
406
407
408async def test_replan_repairs_stale_clip_after_reorder(tmp_path: Path) -> None:
409 """Delete a DJ clip that no longer sits in front of the track it announces."""
410 tracks = [_track(index) for index in range(4)]
411 dummy = _make_replan_dj(tmp_path, list(tracks))
412 state = dummy._dj_queues["queue-1"]
413 stale = _dj_clip("vanished-item", state.dj_session_id)
414 dummy.player_queues._items = [tracks[0], tracks[1], stale, tracks[2], tracks[3]]
415
416 await dummy._replan_queue("queue-1")
417
418 queues = dummy.player_queues
419 assert queues.deleted == [stale.queue_item_id]
420 # the freed gap is plannable again in the same pass
421 announced = {items[0].extra_attributes[ATTR_GAP_NEXT_ID] for items, _ in queues.loads}
422 assert announced == {tracks[2].queue_item_id, tracks[3].queue_item_id}
423
424
425async def test_replan_respects_buffer_guard(tmp_path: Path) -> None:
426 """Never insert into the gap right after the item already loaded into the buffer."""
427 tracks = [_track(index) for index in range(4)]
428 dummy = _make_replan_dj(tmp_path, list(tracks), current_index=0, index_in_buffer=1)
429
430 await dummy._replan_queue("queue-1")
431
432 queues = dummy.player_queues
433 assert len(queues.loads) == 1
434 queue_items, insert_at_index = queues.loads[0]
435 assert insert_at_index == 3
436 assert queue_items[0].extra_attributes[ATTR_GAP_NEXT_ID] == tracks[3].queue_item_id
437
438
439async def test_disable_removes_all_pending_clips(tmp_path: Path) -> None:
440 """Disabling drops every upcoming DJ clip whatever session it came from, played ones stay."""
441 tracks = [_track(index) for index in range(3)]
442 dummy = _make_replan_dj(tmp_path, [], current_index=2, index_in_buffer=2)
443 state = dummy._dj_queues["queue-1"]
444 played_clip = _dj_clip(tracks[1].queue_item_id, state.dj_session_id)
445 pending_clip = _dj_clip(tracks[2].queue_item_id, state.dj_session_id)
446 foreign_clip = _dj_clip(tracks[2].queue_item_id, "other-session")
447 dummy.player_queues._items = [
448 tracks[0],
449 played_clip,
450 tracks[1],
451 pending_clip,
452 tracks[2],
453 foreign_clip,
454 ]
455
456 await dummy.set_queue_dj("queue-1", None)
457
458 assert dummy._dj_queues == {}
459 assert dummy.player_queues.deleted == [pending_clip.queue_item_id, foreign_clip.queue_item_id]
460
461
462async def test_switch_removes_clips_of_a_session_no_state_ever_knew(tmp_path: Path) -> None:
463 """Clear a clip left behind by an earlier provider run, whose session id nothing remembers."""
464 tracks = [_track(index) for index in range(4)]
465 dummy = _make_replan_dj(tmp_path, [])
466 stale = _dj_clip(tracks[2].queue_item_id, "dj_from_a_previous_run")
467 dummy.player_queues._items = [tracks[0], tracks[1], stale, tracks[2], tracks[3]]
468
469 daisy = _must_host()
470 daisy["id"] = "daisy"
471 dummy._hosts["daisy"] = daisy
472 await dummy.set_queue_dj("queue-1", "daisy")
473
474 assert stale.queue_item_id in dummy.player_queues.deleted
475 await asyncio.gather(*dummy.mass.tasks)
476
477
478async def test_cleanup_keeps_the_freshly_armed_sessions_clips(tmp_path: Path) -> None:
479 """A clip the newly armed session already injected survives the previous host's cleanup."""
480 tracks = [_track(index) for index in range(4)]
481 dummy = _make_replan_dj(tmp_path, [])
482 old_state = dummy._dj_queues["queue-1"]
483 old_clip = _dj_clip(tracks[2].queue_item_id, old_state.dj_session_id)
484 dummy.player_queues._items = [tracks[0], tracks[1], old_clip, tracks[2], tracks[3]]
485 write_queue_dj = dummy._write_queue_dj
486
487 async def _write_and_inject_racing_clip() -> None:
488 # stands in for a replan of the freshly armed session landing a clip in the window
489 # before the cleanup of the previous host's clips got its turn
490 await write_queue_dj()
491 session_id = dummy._dj_queues["queue-1"].dj_session_id
492 dummy.player_queues._items.insert(4, _dj_clip(tracks[3].queue_item_id, session_id))
493
494 dummy._write_queue_dj = _write_and_inject_racing_clip # type: ignore[method-assign]
495 daisy = _must_host()
496 daisy["id"] = "daisy"
497 dummy._hosts["daisy"] = daisy
498
499 await dummy.set_queue_dj("queue-1", "daisy")
500
501 assert dummy.player_queues.deleted == [old_clip.queue_item_id]
502 await asyncio.gather(*dummy.mass.tasks)
503
504
505async def test_switching_host_removes_old_hosts_pending_clips(tmp_path: Path) -> None:
506 """Switching the sticky DJ to a new host clears the old host's unplayed clips."""
507 tracks = [_track(index) for index in range(4)]
508 dummy = _make_replan_dj(tmp_path, list(tracks))
509 old_state = dummy._dj_queues["queue-1"]
510 await dummy._replan_queue("queue-1")
511 queues = dummy.player_queues
512 old_clip_ids = {
513 item.queue_item_id
514 for item in queues.items("queue-1")
515 if item.extra_attributes.get(ATTR_QUEUE_DJ)
516 }
517 assert old_clip_ids # sanity: rick's clips actually landed
518
519 daisy = _must_host()
520 daisy["id"] = "daisy"
521 daisy["name"] = "Daisy"
522 dummy._hosts["daisy"] = daisy
523
524 await dummy.set_queue_dj("queue-1", "daisy")
525 await asyncio.gather(*dummy.mass.tasks)
526
527 # rick's pending clips were removed rather than left to render under his persona
528 assert old_clip_ids <= set(queues.deleted)
529 remaining_dj_clips = [
530 item for item in queues.items("queue-1") if item.extra_attributes.get(ATTR_QUEUE_DJ)
531 ]
532 assert remaining_dj_clips
533 for clip in remaining_dj_clips:
534 assert clip.extra_attributes[ATTR_HOST_ID] == "daisy"
535 assert clip.extra_attributes[ATTR_SESSION_ID] != old_state.dj_session_id
536
537
538async def test_min_gap_songs_guard_holds_across_passes(tmp_path: Path) -> None:
539 """Carry the section history across passes so a min_gap_songs guard keeps holding."""
540 tracks = [_track(index) for index in range(4)]
541 dummy = _make_replan_dj(tmp_path, list(tracks), host=_optional_host(3))
542
543 await dummy._replan_queue("queue-1")
544
545 queues = dummy.player_queues
546 assert len(queues.loads) == 1
547 assert queues.loads[0][0][0].extra_attributes[ATTR_GAP_NEXT_ID] == tracks[2].queue_item_id
548
549 extra = [_track(4), _track(5)]
550 queues._items.extend(extra)
551 await dummy._replan_queue("queue-1")
552
553 # the clip announced track 2, so the next one may land no earlier than track 5
554 assert len(queues.loads) == 2
555 assert queues.loads[1][0][0].extra_attributes[ATTR_GAP_NEXT_ID] == extra[1].queue_item_id
556
557
558async def test_planning_axis_stays_anchored_to_real_queue_positions(tmp_path: Path) -> None:
559 """Growing the queue tail must not rewind the song axis the guards count on."""
560 tracks = [_track(index) for index in range(4)]
561 dummy = _make_replan_dj(tmp_path, list(tracks))
562
563 await dummy._replan_queue("queue-1")
564 state = dummy._dj_queues["queue-1"]
565 # the window opened on track 1, so track 0 is the only song behind it
566 assert state.songs_before_window == 1
567 assert [song for song, _minute in state.history["Song_Transition"]] == [2, 3]
568
569 dummy.player_queues._items.extend([_track(4), _track(5)])
570 await dummy._replan_queue("queue-1")
571
572 # the window still opens on track 1, so the appended tracks keep their absolute positions
573 assert state.songs_before_window == 1
574 assert [song for song, _minute in state.history["Song_Transition"]] == [2, 3, 4, 5]
575
576
577async def test_scheduled_replan_serves_requests_landing_during_a_pass(tmp_path: Path) -> None:
578 """A replan request raised by the pass's own inserts is served and then converges."""
579 tracks = [_track(index) for index in range(4)]
580 dummy = _make_replan_dj(tmp_path, list(tracks))
581 queues = dummy.player_queues
582 queues.on_items_updated = lambda: dummy._schedule_replan("queue-1")
583
584 dummy._schedule_replan("queue-1")
585 await asyncio.gather(*dummy.mass.tasks)
586
587 assert len(queues.loads) == 2
588 assert dummy._dj_queues["queue-1"].replan_pending is False
589
590
591async def test_failing_pass_leaves_the_dj_schedulable(tmp_path: Path) -> None:
592 """A raising pass clears its request, does not retry, and recovers on a later event."""
593 tracks = [_track(index) for index in range(4)]
594 dummy = _make_replan_dj(tmp_path, list(tracks))
595 working_build_program = dummy._build_program
596 attempts: list[str] = []
597
598 def _failing_build_program(_station: dict[str, Any], host: dict[str, Any]) -> dict[str, Any]:
599 attempts.append(host["id"])
600 # an event landing mid-pass re-requests a replan behind the still running task
601 dummy._schedule_replan("queue-1")
602 raise MusicAssistantError("misconfigured host")
603
604 dummy._build_program = _failing_build_program # type: ignore[method-assign, assignment]
605 dummy._schedule_replan("queue-1")
606 await asyncio.gather(*dummy.mass.tasks)
607
608 assert attempts == ["rick"]
609 assert dummy.player_queues.loads == []
610 assert dummy._dj_queues["queue-1"].replan_pending is False
611
612 dummy._build_program = working_build_program # type: ignore[method-assign]
613 dummy._schedule_replan("queue-1")
614 await asyncio.gather(*dummy.mass.tasks)
615
616 assert len(dummy.player_queues.loads) == 2
617
618
619async def test_failing_pass_clears_the_state_the_queue_was_rearmed_with(tmp_path: Path) -> None:
620 """A re-arm during a failing pass must leave the new state schedulable, not the old one."""
621 tracks = [_track(index) for index in range(4)]
622 dummy = _make_replan_dj(tmp_path, list(tracks))
623 old_state = dummy._dj_queues["queue-1"]
624 working_build_program = dummy._build_program
625
626 def _failing_build_program(_station: dict[str, Any], host: dict[str, Any]) -> dict[str, Any]:
627 # set_queue_dj re-arms the queue mid-pass, marks the fresh state plannable once its
628 # clip cleanup ran, and latches its request onto the still running task
629 dummy._arm_dj_state("queue-1", host["id"]).ready = True
630 dummy._schedule_replan("queue-1")
631 raise MusicAssistantError("misconfigured host")
632
633 dummy._build_program = _failing_build_program # type: ignore[method-assign, assignment]
634 dummy._schedule_replan("queue-1")
635 await asyncio.gather(*dummy.mass.tasks)
636
637 new_state = dummy._dj_queues["queue-1"]
638 assert new_state is not old_state
639 assert new_state.replan_pending is False
640
641 dummy._build_program = working_build_program # type: ignore[method-assign]
642 dummy._schedule_replan("queue-1")
643 await asyncio.gather(*dummy.mass.tasks)
644
645 assert len(dummy.player_queues.loads) == 2
646
647
648async def test_unloading_provider_schedules_no_replans(tmp_path: Path) -> None:
649 """An unloading provider starts no new replan work."""
650 dummy = _make_replan_dj(tmp_path, [_track(index) for index in range(4)])
651 dummy._unloading = True
652
653 dummy._schedule_replan("queue-1")
654
655 assert dummy.mass.tasks == []
656 state = dummy._dj_queues["queue-1"]
657 assert state.replan_pending is False
658 assert state.task is None
659
660
661async def test_injection_rereads_a_guard_that_moved_during_the_pass(tmp_path: Path) -> None:
662 """A guard that advanced while the pass awaited is honoured by the injections."""
663 tracks = [_track(index) for index in range(4)]
664 dummy = _make_replan_dj(tmp_path, list(tracks), current_index=0, index_in_buffer=0)
665 prepare_runtime_tokens = dummy._prepare_runtime_tokens
666
667 async def _slow_prepare(program: dict[str, Any]) -> dict[str, str]:
668 # stands in for a slow token source: the player buffers ahead while we wait
669 dummy.player_queues._queue.index_in_buffer = 1
670 return await prepare_runtime_tokens(program)
671
672 dummy._prepare_runtime_tokens = _slow_prepare # type: ignore[method-assign]
673 await dummy._replan_queue("queue-1")
674
675 queues = dummy.player_queues
676 assert len(queues.loads) == 1
677 assert queues.loads[0][0][0].extra_attributes[ATTR_GAP_NEXT_ID] == tracks[3].queue_item_id
678
679
680async def test_stale_pass_inserts_nothing_after_a_mid_pass_dj_switch(tmp_path: Path) -> None:
681 """A pass that resumes after set_queue_dj replaced its state must not insert clips."""
682 tracks = [_track(index) for index in range(4)]
683 dummy = _make_replan_dj(tmp_path, list(tracks))
684 old_state = dummy._dj_queues["queue-1"]
685 prepare_runtime_tokens = dummy._prepare_runtime_tokens
686
687 daisy = _must_host()
688 daisy["id"] = "daisy"
689 dummy._hosts["daisy"] = daisy
690
691 async def _switch_mid_pass(program: dict[str, Any]) -> dict[str, str]:
692 # stands in for set_queue_dj swapping this queue to another host while the
693 # in-flight pass is still awaiting a slow runtime token fetch
694 dummy._arm_dj_state("queue-1", "daisy")
695 return await prepare_runtime_tokens(program)
696
697 dummy._prepare_runtime_tokens = _switch_mid_pass # type: ignore[method-assign]
698 await dummy._replan_queue("queue-1")
699
700 assert dummy.player_queues.loads == []
701 new_state = dummy._dj_queues["queue-1"]
702 assert new_state is not old_state
703 assert new_state.decided_gap_ids == set()
704
705
706async def test_replan_holds_off_until_the_switch_cleanup_ran(tmp_path: Path) -> None:
707 """A pass entering between arming a host and clearing the old clips must not plan."""
708 tracks = [_track(index) for index in range(4)]
709 dummy = _make_replan_dj(tmp_path, list(tracks))
710 # stands in for set_queue_dj arming the new host while a drain task is already awake
711 state = dummy._arm_dj_state("queue-1", "rick")
712 state.replan_pending = True
713
714 await dummy._replan_queue("queue-1")
715
716 assert dummy.player_queues.loads == []
717 assert state.decided_gap_ids == set()
718 # the request is dropped, not kept, or the drain loop would spin on the bail
719 assert state.replan_pending is False
720
721
722async def test_switch_refills_the_gaps_its_own_cleanup_frees(tmp_path: Path) -> None:
723 """A replan racing a host switch must not leave the gaps its cleanup frees unplanned."""
724 tracks = [_track(index) for index in range(4)]
725 dummy = _make_replan_dj(tmp_path, list(tracks))
726 await dummy._replan_queue("queue-1")
727 queues = dummy.player_queues
728 old_clip_ids = {
729 item.queue_item_id
730 for item in queues.items("queue-1")
731 if item.extra_attributes.get(ATTR_QUEUE_DJ)
732 }
733 assert len(old_clip_ids) == 2 # sanity: rick filled both gaps
734
735 daisy = _must_host()
736 daisy["id"] = "daisy"
737 dummy._hosts["daisy"] = daisy
738 write_queue_dj = dummy._write_queue_dj
739
740 async def _write_and_replan() -> None:
741 # a queue event wakes the drain task while the switch sits between arming the new
742 # state and clearing the previous host's clips
743 await write_queue_dj()
744 await dummy._replan_queue("queue-1")
745
746 dummy._write_queue_dj = _write_and_replan # type: ignore[method-assign]
747 await dummy.set_queue_dj("queue-1", "daisy")
748 await asyncio.gather(*dummy.mass.tasks)
749
750 assert old_clip_ids <= set(queues.deleted)
751 remaining = [
752 item for item in queues.items("queue-1") if item.extra_attributes.get(ATTR_QUEUE_DJ)
753 ]
754 assert len(remaining) == 2
755 for clip in remaining:
756 assert clip.extra_attributes[ATTR_HOST_ID] == "daisy"
757
758
759async def test_a_failing_switch_cleanup_leaves_no_armed_state(tmp_path: Path) -> None:
760 """A switch whose cleanup blows up must not leave behind a state that can never plan."""
761 dummy = _make_replan_dj(tmp_path, [_track(index) for index in range(3)])
762 daisy = _must_host()
763 daisy["id"] = "daisy"
764 dummy._hosts["daisy"] = daisy
765
766 def _failing_cleanup(queue_id: str) -> None: # noqa: ARG001
767 raise MusicAssistantError("queue layer is unhappy")
768
769 dummy._remove_pending_dj_clips = _failing_cleanup # type: ignore[method-assign]
770
771 with pytest.raises(MusicAssistantError):
772 await dummy.set_queue_dj("queue-1", "daisy")
773
774 assert dummy._dj_queues == {}
775 assert dummy.mass.tasks == []
776
777
778async def test_a_rejected_clip_keeps_no_guard_history(tmp_path: Path) -> None:
779 """A merged clip the splice rejects must not block its own sections from airing later."""
780 tracks = [_track(index) for index in range(4)]
781 dummy = _make_replan_dj(tmp_path, list(tracks), host=_hourly_host())
782 for section in _hourly_sections():
783 dummy._sections[section["id"]] = section
784 prepare_runtime_tokens = dummy._prepare_runtime_tokens
785
786 async def _slow_prepare(program: dict[str, Any]) -> dict[str, str]:
787 # the player buffers ahead while the pass awaits, so the planned gap is spoken for
788 dummy.player_queues._queue.index_in_buffer = 1
789 return await prepare_runtime_tokens(program)
790
791 dummy._prepare_runtime_tokens = _slow_prepare # type: ignore[method-assign]
792 await dummy._replan_queue("queue-1")
793
794 state = dummy._dj_queues["queue-1"]
795 assert dummy.player_queues.loads == []
796 assert state.history["Weather"] == []
797 assert state.history["News"] == []
798
799 # the queue grows, so both sections get a fresh gap to air in
800 dummy._prepare_runtime_tokens = prepare_runtime_tokens # type: ignore[method-assign]
801 extra = [_track(4), _track(5)]
802 dummy.player_queues._items.extend(extra)
803 await dummy._replan_queue("queue-1")
804
805 assert len(dummy.player_queues.loads) == 1
806 clip = dummy.player_queues.loads[0][0][0]
807 assert clip.name == "Weather + News"
808 assert clip.extra_attributes[ATTR_GAP_NEXT_ID] == extra[0].queue_item_id
809
810
811async def test_a_pass_applies_all_its_clips_in_one_queue_update(tmp_path: Path) -> None:
812 """A whole window of clips reaches the clients as one update, not one per clip."""
813 tracks = [_track(index) for index in range(6)]
814 dummy = _make_replan_dj(tmp_path, list(tracks))
815
816 await dummy._replan_queue("queue-1")
817
818 queues = dummy.player_queues
819 assert len(queues.loads) == 4 # one clip per plannable gap
820 assert queues.update_calls == 1
821 for clip_items, _index in queues.loads:
822 clip = clip_items[0]
823 target = next(
824 item
825 for item in queues.items("queue-1")
826 if item.queue_item_id == clip.extra_attributes[ATTR_GAP_NEXT_ID]
827 )
828 # sharing the target's sort index keeps the clip next to it when un-shuffling
829 assert clip.sort_index == target.sort_index
830
831
832async def test_a_switch_clears_and_refills_in_one_update_per_phase(tmp_path: Path) -> None:
833 """The cleanup and the refill of a switch each replace the queue exactly once."""
834 tracks = [_track(index) for index in range(6)]
835 dummy = _make_replan_dj(tmp_path, list(tracks))
836 await dummy._replan_queue("queue-1")
837 queues = dummy.player_queues
838 assert queues.update_calls == 1
839 daisy = _must_host()
840 daisy["id"] = "daisy"
841 dummy._hosts["daisy"] = daisy
842
843 await dummy.set_queue_dj("queue-1", "daisy")
844
845 assert queues.update_calls == 2
846 assert len(queues.deleted) == 4
847
848 await asyncio.gather(*dummy.mass.tasks)
849
850 assert queues.update_calls == 3
851 assert len(queues.loads) == 8
852
853
854async def test_a_vanished_target_leaves_its_gap_open(tmp_path: Path) -> None:
855 """A gap whose target left the queue mid-pass stays open, so its return is planned."""
856 tracks = [_track(index) for index in range(6)]
857 dummy = _make_replan_dj(tmp_path, list(tracks))
858 prepare_runtime_tokens = dummy._prepare_runtime_tokens
859
860 async def _slow_prepare(program: dict[str, Any]) -> dict[str, str]:
861 # stands in for a slow token source: the planned tracks are already fixed when the
862 # user removes one of them from the queue
863 dummy.player_queues.update_items(
864 "queue-1",
865 [item for item in dummy.player_queues.items("queue-1") if item is not tracks[3]],
866 )
867 return await prepare_runtime_tokens(program)
868
869 dummy._prepare_runtime_tokens = _slow_prepare # type: ignore[method-assign]
870 await dummy._replan_queue("queue-1")
871
872 queues = dummy.player_queues
873 announced = {items[0].extra_attributes[ATTR_GAP_NEXT_ID] for items, _ in queues.loads}
874 assert announced == {
875 tracks[2].queue_item_id,
876 tracks[4].queue_item_id,
877 tracks[5].queue_item_id,
878 }
879 state = dummy._dj_queues["queue-1"]
880 assert state.decided_gap_ids == announced
881
882 # the track returns, as a reorder that rebuilds the queue does
883 dummy._prepare_runtime_tokens = prepare_runtime_tokens # type: ignore[method-assign]
884 queues._items.append(tracks[3])
885 await dummy._replan_queue("queue-1")
886
887 assert queues.loads[-1][0][0].extra_attributes[ATTR_GAP_NEXT_ID] == tracks[3].queue_item_id
888 assert tracks[3].queue_item_id in state.decided_gap_ids
889
890
891async def test_reordering_the_queue_keeps_the_moved_gaps_plannable(tmp_path: Path) -> None:
892 """Shuffling new tracks into the upcoming items must not hide the moved gaps for good."""
893 tracks = [_track(index) for index in range(4)]
894 dummy = _make_replan_dj(tmp_path, list(tracks))
895 await dummy._replan_queue("queue-1")
896 queues = dummy.player_queues
897 assert len(queues.loads) == 2
898 clips = {
899 item.extra_attributes[ATTR_GAP_NEXT_ID]: item
900 for item in queues.items("queue-1")
901 if item.extra_attributes.get(ATTR_QUEUE_DJ)
902 }
903
904 # shuffle on add: the new tracks land in front of the old ones, moving the previously
905 # last planned track to the very back. the clips travelled along with their own track
906 extra = [_track(4), _track(5)]
907 queues._items = [
908 tracks[0],
909 extra[0],
910 tracks[1],
911 extra[1],
912 clips[tracks[2].queue_item_id],
913 tracks[2],
914 clips[tracks[3].queue_item_id],
915 tracks[3],
916 ]
917 await dummy._replan_queue("queue-1")
918
919 assert queues.deleted == []
920 announced = {items[0].extra_attributes[ATTR_GAP_NEXT_ID] for items, _ in queues.loads[2:]}
921 assert announced == {tracks[1].queue_item_id, extra[1].queue_item_id}
922
923
924async def test_appending_a_batch_plans_only_the_new_gaps(tmp_path: Path) -> None:
925 """A batch appended to a progressive queue is planned without re-rolling settled gaps."""
926 tracks = [_track(index) for index in range(4)]
927 dummy = _make_replan_dj(tmp_path, list(tracks))
928 await dummy._replan_queue("queue-1")
929 queues = dummy.player_queues
930 state = dummy._dj_queues["queue-1"]
931
932 extra = [_track(4), _track(5)]
933 queues._items.extend(extra)
934 await dummy._replan_queue("queue-1")
935
936 assert len(queues.loads) == 4
937 announced = {items[0].extra_attributes[ATTR_GAP_NEXT_ID] for items, _ in queues.loads[2:]}
938 assert announced == {extra[0].queue_item_id, extra[1].queue_item_id}
939 # the settled gaps never reached the planner, so their events are registered once
940 assert [song for song, _minute in state.history["Song_Transition"]] == [2, 3, 4, 5]
941
942
943async def test_a_repaired_clip_reopens_its_gap(tmp_path: Path) -> None:
944 """A clip the repair drops leaves its gap open, so a later pass fills it again."""
945 tracks = [_track(index) for index in range(4)]
946 dummy = _make_replan_dj(tmp_path, list(tracks))
947 await dummy._replan_queue("queue-1")
948 queues = dummy.player_queues
949 stale = next(
950 item
951 for item in queues.items("queue-1")
952 if item.extra_attributes.get(ATTR_GAP_NEXT_ID) == tracks[3].queue_item_id
953 )
954
955 # a queue edit moved the clip away from the track it announces
956 queues._items = [item for item in queues._items if item is not stale] + [stale]
957 await dummy._replan_queue("queue-1")
958
959 assert queues.deleted == [stale.queue_item_id]
960 assert queues.loads[-1][0][0].extra_attributes[ATTR_GAP_NEXT_ID] == tracks[3].queue_item_id
961
962
963async def test_replan_yields_the_queue_to_a_running_show(tmp_path: Path) -> None:
964 """A show owning the queue plans its own breaks, so the sticky DJ stays out of it."""
965 dummy = _make_replan_dj(tmp_path, [_track(index) for index in range(4)])
966 dummy._sessions["s1"] = SessionState(
967 session_id="s1", station_id="station_a", queue_id="queue-1"
968 )
969
970 await dummy._replan_queue("queue-1")
971
972 assert dummy.player_queues.loads == []
973 assert dummy._dj_queues["queue-1"].replan_pending is False
974
975
976async def test_replan_ignores_a_show_running_on_another_queue(tmp_path: Path) -> None:
977 """A show elsewhere leaves this queue's DJ working."""
978 dummy = _make_replan_dj(tmp_path, [_track(index) for index in range(4)])
979 dummy._sessions["s1"] = SessionState(
980 session_id="s1", station_id="station_a", queue_id="queue-2"
981 )
982 dummy._sessions["s2"] = SessionState(
983 session_id="s2", station_id="station_b", queue_id="queue-1", status="completed"
984 )
985
986 await dummy._replan_queue("queue-1")
987
988 assert len(dummy.player_queues.loads) == 2
989
990
991async def test_replan_keeps_state_when_the_queue_is_not_registered_yet(tmp_path: Path) -> None:
992 """A queue that has not registered yet is not treated as a vanished queue."""
993 dummy = _make_replan_dj(tmp_path, [_track(index) for index in range(4)])
994 # the boot-time replan runs before on_player_register created the queue
995 dummy.player_queues._queue = FakeQueue("some-other-queue", None, None)
996 dummy._dj_file.write_text("untouched")
997
998 await dummy._replan_queue("queue-1")
999
1000 assert "queue-1" in dummy._dj_queues
1001 assert dummy._dj_file.read_text() == "untouched"
1002
1003
1004async def test_queue_added_event_replans_an_armed_queue(tmp_path: Path) -> None:
1005 """A queue registering after the provider loaded resumes injection on its own."""
1006 tracks = [_track(index) for index in range(4)]
1007 dummy = _make_replan_dj(tmp_path, list(tracks))
1008 dummy.player_queues._queue = FakeQueue("some-other-queue", None, None)
1009 await dummy._replan_queue("queue-1")
1010 assert dummy.player_queues.loads == []
1011
1012 dummy.player_queues._queue = FakeQueue("queue-1", 0, 0)
1013 await dummy._on_dj_queue_event(
1014 cast("Any", SimpleNamespace(event=EventType.QUEUE_ADDED, object_id="queue-1"))
1015 )
1016 await asyncio.gather(*dummy.mass.tasks)
1017
1018 assert len(dummy.player_queues.loads) == 2
1019
1020
1021async def test_dj_clip_is_sound_effect_media_type(tmp_path: Path) -> None:
1022 """A DJ clip's media item is a SoundEffect, the type the dynamic pool exclusions key on."""
1023 dummy = _make_replan_dj(tmp_path, [])
1024 section = PlannedSection(
1025 order=0,
1026 clip_id="dj_clip_1",
1027 section_id="Song_Transition",
1028 section_name="Song Transition",
1029 when="between_songs",
1030 insert_at_index=0,
1031 prompt="",
1032 max_chars=200,
1033 web_search_mode="disabled",
1034 )
1035 clip = dummy._section_to_clip_item("queue-1", "sess", {"id": "", "host_id": "rick"}, section)
1036 assert clip.media_item is not None
1037 assert clip.media_item.media_type == MediaType.SOUND_EFFECT
1038