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