/
/
/
1"""Tests for how a fully served flow stream is closed off."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock, patch
6
7import pytest
8
9from music_assistant.controllers.streams.constants import FLOW_STREAM_LEAD_OUT_SECONDS
10from music_assistant.controllers.streams.controller import StreamsController
11from music_assistant.helpers.webserver import DEFAULT_SHUTDOWN_TIMEOUT
12
13SESSION_ID = "session-1"
14QUEUE_ID = "queue-1"
15
16
17def _controller(*, exhausted: bool) -> StreamsController:
18 """Build a streams controller whose queue reports the given end-of-queue state."""
19 mass = MagicMock()
20 mass.config.get_raw_core_config_value.return_value = "GLOBAL"
21 mass.player_queues.flow_queue_exhausted = MagicMock(
22 side_effect=lambda qid, sid: exhausted and (qid, sid) == (QUEUE_ID, SESSION_ID)
23 )
24 return StreamsController(mass)
25
26
27async def _finish(
28 controller: StreamsController, session_id: str = SESSION_ID
29) -> tuple[AsyncMock, MagicMock]:
30 """Finish a flow stream against a stub response, returning the sleep and the response."""
31 resp = MagicMock()
32 with patch(
33 "music_assistant.controllers.streams.controller.asyncio.sleep", new=AsyncMock()
34 ) as sleep:
35 await controller._finish_flow_stream(resp, QUEUE_ID, session_id)
36 return sleep, resp
37
38
39@pytest.mark.asyncio
40async def test_holds_connection_open_after_the_last_queue_item() -> None:
41 """A flow stream that played the queue to its end lets the player drain first."""
42 sleep, _ = await _finish(_controller(exhausted=True))
43 sleep.assert_awaited_once_with(FLOW_STREAM_LEAD_OUT_SECONDS)
44
45
46@pytest.mark.asyncio
47async def test_closes_the_connection_after_the_lead_out() -> None:
48 """The player only learns the stream ended once the connection is really closed."""
49 _, resp = await _finish(_controller(exhausted=True))
50 resp.force_close.assert_called_once()
51
52
53@pytest.mark.asyncio
54async def test_closes_without_delay_when_the_flow_restarts() -> None:
55 """A flow that ended early must free the player at once so the next stream can start."""
56 sleep, resp = await _finish(_controller(exhausted=False))
57 sleep.assert_not_awaited()
58 resp.force_close.assert_called_once()
59
60
61@pytest.mark.asyncio
62async def test_closes_without_delay_when_superseded() -> None:
63 """A newer stream session owns playback, so the stale response must not linger."""
64 sleep, resp = await _finish(_controller(exhausted=True), session_id="session-2")
65 sleep.assert_not_awaited()
66 resp.force_close.assert_called_once()
67
68
69def test_lead_out_fits_within_the_webserver_shutdown_budget() -> None:
70 """A lead-out in flight must not outlive the shutdown grace period."""
71 assert FLOW_STREAM_LEAD_OUT_SECONDS < DEFAULT_SHUTDOWN_TIMEOUT
72