music-assistant-server

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