/
/
1"""Regression tests for delayed next-track enqueueing in the player queue stream feeder."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncIterator
7from contextlib import asynccontextmanager
8from types import SimpleNamespace
9from unittest.mock import AsyncMock, MagicMock
10
11import pytest
12from music_assistant_models.enums import PlaybackState
13from music_assistant_models.player_queue import PlayerQueue
14from music_assistant_models.queue_item import QueueItem
15
16from music_assistant.controllers.player_queues import PlayerQueuesController
17from music_assistant.controllers.player_queues.state import PlayerQueueData
18
19
20@pytest.mark.parametrize(
21 "index_in_buffer",
22 [0, 1],
23 ids=["aligned-buffer-index", "dynamic-queue-reindexed"],
24)
25async def test_enqueue_next_item_waits_for_playing_player_update(index_in_buffer: int) -> None:
26 """Enqueue the expected next item after an update, even if its buffered index is stale."""
27 controller = PlayerQueuesController.__new__(PlayerQueuesController)
28 controller.logger = MagicMock()
29
30 wait_entered = asyncio.Event()
31 release_wait = asyncio.Event()
32
33 @asynccontextmanager
34 async def wait_for_player_update(*_args: object, **_kwargs: object) -> AsyncIterator[None]:
35 wait_entered.set()
36 await release_wait.wait()
37 yield
38
39 player_state = SimpleNamespace(
40 playback_state=PlaybackState.IDLE,
41 active_source="q1",
42 )
43 player = SimpleNamespace(state=player_state)
44
45 mass = MagicMock()
46 mass.players = MagicMock()
47 mass.players.wait_for_player_update = MagicMock(side_effect=wait_for_player_update)
48 mass.players.get_player = MagicMock(return_value=player)
49 mass.players.enqueue_next_media = AsyncMock()
50 controller.mass = mass
51
52 current_item = _make_queue_item("q1", "nerin")
53 next_item = _make_queue_item("q1", "another-love")
54 future_item = _make_queue_item("q1", "future-track")
55 queue_items = [current_item, next_item, future_item]
56 queue = PlayerQueue(
57 queue_id="q1",
58 active=True,
59 display_name="Q1",
60 available=True,
61 items=len(queue_items),
62 state=PlaybackState.IDLE,
63 current_index=0,
64 index_in_buffer=index_in_buffer,
65 current_item=current_item,
66 )
67 controller._queue_data = {
68 "q1": PlayerQueueData(
69 queue=queue,
70 items=queue_items,
71 session_id="session-1",
72 )
73 }
74
75 controller._enqueue_next_item("q1", next_item)
76 enqueue_callback = mass.call_later.call_args.args[1]
77 enqueue_task = asyncio.create_task(enqueue_callback(next_item))
78 await asyncio.sleep(0)
79
80 assert wait_entered.is_set()
81 mass.players.enqueue_next_media.assert_not_awaited()
82
83 player_state.playback_state = PlaybackState.PLAYING
84 release_wait.set()
85 await enqueue_task
86
87 mass.players.wait_for_player_update.assert_called_once_with(
88 "q1",
89 attribute_name="playback_state",
90 attribute_value=PlaybackState.PLAYING,
91 )
92 mass.players.enqueue_next_media.assert_awaited_once()
93 assert mass.players.enqueue_next_media.await_args.kwargs["player_id"] == "q1"
94 assert (
95 mass.players.enqueue_next_media.await_args.kwargs["media"].queue_item_id
96 == next_item.queue_item_id
97 )
98 assert controller._queue_data["q1"].next_item_id_enqueued == next_item.queue_item_id
99
100
101def _make_queue_item(queue_id: str, item_id: str) -> QueueItem:
102 """Build a minimal playable queue item."""
103 return QueueItem(
104 queue_id=queue_id,
105 queue_item_id=item_id,
106 name=item_id,
107 duration=60,
108 )
109