/
/
/
1"""
2Tests for what a queue stop tears down.
3
4Stopping only the device leaves the queue's session open, so its item buffers keep
5producing and a provider serving a live session stays tethered to Music Assistant.
6That teardown has to happen even when the device could not be told to stop at all.
7"""
8
9from __future__ import annotations
10
11import contextlib
12from collections.abc import AsyncIterator
13from typing import TYPE_CHECKING, Any, cast
14from unittest.mock import AsyncMock, MagicMock
15
16import pytest
17from music_assistant_models.enums import PlaybackState
18from music_assistant_models.errors import PlayerUnavailableError
19
20from music_assistant.controllers.player_queues import PlayerQueuesController
21from music_assistant.controllers.player_queues.state import PlayerQueueData
22
23if TYPE_CHECKING:
24 from music_assistant_models.player_queue import PlayerQueue
25
26
27def _fake_controller() -> MagicMock:
28 """Build a MagicMock standing in for the controller, owning one playing queue."""
29 queue = MagicMock(
30 queue_id="q",
31 active=True,
32 state=PlaybackState.PLAYING,
33 corrected_elapsed_time=42.0,
34 )
35 fake = MagicMock()
36 data = PlayerQueueData(queue=cast("PlayerQueue", queue))
37 data.session_id = "sess-1"
38 fake._queue_data = {"q": data}
39 fake.get = MagicMock(
40 side_effect=lambda qid: d.queue if (d := fake._queue_data.get(qid)) else None
41 )
42 fake.mass.players._handle_cmd_stop = AsyncMock()
43
44 def _close_coro(target: Any, **_kwargs: Any) -> None:
45 # the cleanup is handed to create_task as a coroutine; nothing awaits it here
46 if hasattr(target, "close"):
47 target.close()
48
49 fake.mass.create_task = MagicMock(side_effect=_close_coro)
50
51 @contextlib.asynccontextmanager
52 async def _no_lock(*_args: Any, **_kwargs: Any) -> AsyncIterator[None]:
53 """Stand in for the playback lock the stop is wrapped in."""
54 yield
55
56 fake.mass.players.get_player_lock = _no_lock
57 # the play-action wrapper flags the queue while the stop runs
58 queue.extra_attributes = {}
59 return fake
60
61
62async def _stop(fake: MagicMock) -> None:
63 """Run a stop against the fake controller."""
64 await PlayerQueuesController._handle_stop(cast("PlayerQueuesController", fake), "q")
65
66
67@pytest.mark.asyncio
68async def test_stop_ends_the_session_and_clears_the_buffers() -> None:
69 """A stop closes the playback session and hands the audio data to the cleanup."""
70 fake = _fake_controller()
71
72 await _stop(fake)
73
74 fake.mass.players._handle_cmd_stop.assert_awaited_once_with("q")
75 assert fake._queue_data["q"].session_id is None
76 fake.mass.streams.audio_processing.clear.assert_called_once_with("q", "sess-1")
77 fake._cleanup_queue_audio_data.assert_called_once_with("q")
78
79
80@pytest.mark.asyncio
81async def test_a_player_that_cannot_be_stopped_still_loses_its_session() -> None:
82 """
83 A device gone unavailable must not keep the queue tethered to its provider.
84
85 Its power was switched off outside MA and it dropped off the network before the
86 stop arrived - exactly the case where the session has to be released.
87 """
88 fake = _fake_controller()
89 fake.mass.players._handle_cmd_stop.side_effect = PlayerUnavailableError("gone")
90
91 with pytest.raises(PlayerUnavailableError):
92 await _stop(fake)
93
94 assert fake._queue_data["q"].session_id is None
95 fake.mass.streams.audio_processing.clear.assert_called_once_with("q", "sess-1")
96 fake._cleanup_queue_audio_data.assert_called_once_with("q")
97
98
99@pytest.mark.asyncio
100async def test_a_stop_that_lost_the_race_to_a_new_session_leaves_it_alone() -> None:
101 """Playback that restarted while the stop ran keeps its own session."""
102 fake = _fake_controller()
103
104 async def _restart_the_session(_queue_id: str) -> None:
105 fake._queue_data["q"].session_id = "sess-2"
106
107 fake.mass.players._handle_cmd_stop.side_effect = _restart_the_session
108
109 await _stop(fake)
110
111 assert fake._queue_data["q"].session_id == "sess-2"
112