/
/
1"""Tests for giving up on an external source that has been paused for too long."""
2
3from __future__ import annotations
4
5import time
6from unittest.mock import MagicMock
7
8import pytest
9from music_assistant_models.enums import PlaybackState
10
11from music_assistant.constants import EXTERNAL_PAUSE_IDLE_TIMEOUT
12from tests.common import MockPlayer, MockProvider
13
14EXTERNAL_SOURCE = "spotify"
15
16
17@pytest.fixture
18def mock_mass() -> MagicMock:
19 """Create a mock MusicAssistant instance."""
20 mass = MagicMock()
21 mass.closing = False
22 mass.config.get_raw_player_config_value = MagicMock(
23 side_effect=lambda _player_id, _key, default=None: default
24 )
25 mass.player_queues.get = MagicMock(return_value=None)
26 mass.players.scale_volume_from_device = MagicMock(side_effect=lambda _player_id, volume: volume)
27 return mass
28
29
30@pytest.fixture
31def player(mock_mass: MagicMock) -> MockPlayer:
32 """Create a player reporting a paused external source, with the check opted in."""
33 provider = MockProvider("test_provider", mass=mock_mass)
34 player = MockPlayer(provider, "player_1", "Player 1")
35 player._attr_external_pause_idle_timeout = EXTERNAL_PAUSE_IDLE_TIMEOUT
36 player._attr_playback_state = PlaybackState.PAUSED
37 player._attr_active_source = EXTERNAL_SOURCE
38 player.set_current_media(uri="spotify:track:1", title="Shout")
39 player.update_state()
40 return player
41
42
43def _backdate_pause(player: MockPlayer, seconds: float) -> None:
44 """Pretend the source has already been sitting paused for the given time."""
45 player._Player__external_pause_since = time.time() - seconds # type: ignore[attr-defined]
46
47
48def test_a_source_that_just_paused_stays_resumable(
49 player: MockPlayer, mock_mass: MagicMock
50) -> None:
51 """Test a real pause is still handed to the source, so pausing and resuming works."""
52 assert player.state.playback_state is PlaybackState.PAUSED
53 assert player.state.active_source == EXTERNAL_SOURCE
54 # the device reports nothing when the session goes stale, so we come back to it ourselves
55 armed_checks = [
56 call
57 for call in mock_mass.call_later.call_args_list
58 if call.kwargs.get("task_id") == "external_pause_player_1"
59 ]
60 assert len(armed_checks) == 1
61 assert armed_checks[0].args[0] > EXTERNAL_PAUSE_IDLE_TIMEOUT
62
63
64def test_a_source_paused_within_the_grace_period_stays_resumable(player: MockPlayer) -> None:
65 """Test a genuine pause survives for as long as the grace period lasts."""
66 _backdate_pause(player, EXTERNAL_PAUSE_IDLE_TIMEOUT - 5)
67
68 player.update_state()
69
70 assert player.state.playback_state is PlaybackState.PAUSED
71 assert player.state.active_source == EXTERNAL_SOURCE
72 assert player.state.current_media is not None
73
74
75def test_a_source_paused_for_too_long_is_reported_as_idle(player: MockPlayer) -> None:
76 """Test an abandoned session stops looking resumable, so play starts our own queue."""
77 _backdate_pause(player, EXTERNAL_PAUSE_IDLE_TIMEOUT + 1)
78
79 player.update_state()
80
81 assert player.state.playback_state is PlaybackState.IDLE
82 assert player._attr_current_media is None
83 # the player's own queue is reachable again, which is what makes the play button work
84 assert player.state.active_source == player.player_id
85
86
87def test_an_ended_source_is_not_picked_back_up_from_what_the_device_reports(
88 player: MockPlayer,
89) -> None:
90 """Test the next update does not resurrect the source the device still reports as paused."""
91 player.mark_external_source_ended()
92 player.update_state()
93 assert player.state.playback_state is PlaybackState.IDLE
94
95 # the device keeps reporting the very same paused source on every update that follows
96 player._attr_playback_state = PlaybackState.PAUSED
97 player._attr_active_source = EXTERNAL_SOURCE
98 player.update_state()
99
100 assert player.state.playback_state is PlaybackState.IDLE
101 assert player.state.active_source == player.player_id
102
103
104def test_a_source_that_starts_playing_again_is_given_a_fresh_start(player: MockPlayer) -> None:
105 """Test a source we gave up on is trusted again once the device really plays it."""
106 player.mark_external_source_ended()
107 player.update_state()
108
109 player._attr_playback_state = PlaybackState.PLAYING
110 player._attr_active_source = EXTERNAL_SOURCE
111 player.update_state()
112 player._attr_playback_state = PlaybackState.PAUSED
113 player.update_state()
114
115 assert player.state.playback_state is PlaybackState.PAUSED
116 assert player.state.active_source == EXTERNAL_SOURCE
117
118
119def test_another_source_is_not_tarred_with_the_same_brush(player: MockPlayer) -> None:
120 """Test giving up on one source does not shorten the grace period of the next."""
121 player.mark_external_source_ended()
122 player.update_state()
123
124 player._attr_playback_state = PlaybackState.PAUSED
125 player._attr_active_source = "tidal"
126 player.update_state()
127
128 assert player.state.playback_state is PlaybackState.PAUSED
129 assert player.state.active_source == "tidal"
130
131
132def test_resuming_the_source_gives_a_later_pause_the_full_grace_period(
133 player: MockPlayer,
134) -> None:
135 """Test the clock only runs while the player reports paused."""
136 _backdate_pause(player, EXTERNAL_PAUSE_IDLE_TIMEOUT - 5)
137 player._attr_playback_state = PlaybackState.PLAYING
138 player.update_state()
139
140 player._attr_playback_state = PlaybackState.PAUSED
141 player.update_state()
142
143 assert player.state.playback_state is PlaybackState.PAUSED
144 assert player.state.active_source == EXTERNAL_SOURCE
145
146
147def test_our_own_paused_playback_is_never_expired(player: MockPlayer) -> None:
148 """Test pausing the Music Assistant queue is left alone, however long it lasts."""
149 player._attr_active_source = player.player_id
150 _backdate_pause(player, EXTERNAL_PAUSE_IDLE_TIMEOUT + 1)
151
152 player.update_state()
153
154 assert player.state.playback_state is PlaybackState.PAUSED
155
156
157def test_a_paused_queue_source_is_never_expired(player: MockPlayer, mock_mass: MagicMock) -> None:
158 """Test a source that resolves to a Music Assistant queue is not treated as external."""
159 mock_mass.player_queues.get = MagicMock(return_value=MagicMock())
160 _backdate_pause(player, EXTERNAL_PAUSE_IDLE_TIMEOUT + 1)
161
162 player.update_state()
163
164 assert player.state.playback_state is PlaybackState.PAUSED
165
166
167def test_a_source_rendered_by_an_output_protocol_is_never_expired(player: MockPlayer) -> None:
168 """Test playback that Music Assistant renders itself is left to the protocol player."""
169 player.set_active_output_protocol("airplay_player_1")
170 _backdate_pause(player, EXTERNAL_PAUSE_IDLE_TIMEOUT + 1)
171
172 player.update_state()
173
174 assert player._attr_playback_state is PlaybackState.PAUSED
175 assert player._attr_active_source == EXTERNAL_SOURCE
176
177
178def test_a_player_that_does_not_opt_in_keeps_its_paused_source(
179 mock_mass: MagicMock,
180) -> None:
181 """Test devices that report an abandoned source as stopped themselves are untouched."""
182 provider = MockProvider("test_provider", mass=mock_mass)
183 player = MockPlayer(provider, "player_2", "Player 2")
184 player._attr_playback_state = PlaybackState.PAUSED
185 player._attr_active_source = EXTERNAL_SOURCE
186 player.update_state()
187 _backdate_pause(player, EXTERNAL_PAUSE_IDLE_TIMEOUT + 1)
188
189 player.update_state()
190
191 assert player.state.playback_state is PlaybackState.PAUSED
192 assert player.state.active_source == EXTERNAL_SOURCE
193
194
195@pytest.mark.asyncio
196async def test_unloading_leaves_no_pending_check_behind(
197 player: MockPlayer, mock_mass: MagicMock
198) -> None:
199 """Test unloading a player cancels the check that would re-evaluate its paused source."""
200 await player.on_unload()
201
202 mock_mass.cancel_timer.assert_any_call("external_pause_player_1")
203
204
205@pytest.mark.asyncio
206async def test_unloading_a_player_that_does_not_opt_in_cancels_nothing(
207 mock_mass: MagicMock,
208) -> None:
209 """Test the check is not cleaned up for players that never armed it."""
210 provider = MockProvider("test_provider", mass=mock_mass)
211 player = MockPlayer(provider, "player_2", "Player 2")
212
213 await player.on_unload()
214
215 mock_mass.cancel_timer.assert_not_called()
216