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