/
/
/
1"""
2Tests for YandexStationPlayer state transitions during external playback.
3
4Covers physical pause detection: when the user presses pause on the speaker
5while MA is streaming via ``radio_play``, Glagol reports ``playing=False`` with
6``aliceState="IDLE"``. The player must propagate ``PlaybackState.PAUSED`` to MA
7instead of silently holding the optimistic PLAYING state.
8"""
9
10from __future__ import annotations
11
12from music_assistant_models.enums import PlaybackState
13
14from music_assistant.providers.yandex_station.player import YandexStationPlayer
15
16
17def _make_player() -> YandexStationPlayer:
18 """Build a bare player skipping Player.__init__ (avoids MA core deps)."""
19 player = YandexStationPlayer.__new__(YandexStationPlayer)
20 player._player_id = "test_player"
21 player._external_playing = False
22 player._external_media = None
23 player._external_play_confirmed = False
24 player._needs_replay = False
25 player._prev_alice_state = ""
26 player._voice_resume_task = None
27 player._alice_spoke = False
28 player._pre_voice_volume = 0
29 player._intercept_active = False
30 player._last_intercepted_track_id = None
31 player._last_intercept_time = 0.0
32 player._last_mirrored_volume = None
33 player._last_progress = 0
34 player._last_progress_wall = 0.0
35 player._alice_active_pause_sent = False
36 player._saved_station_volume = None
37 player._station_muted_by_intercept = False
38 player._attr_playback_state = PlaybackState.IDLE
39 player._attr_powered = False
40 player._attr_volume_level = 0
41 return player
42
43
44class _StubConfig:
45 def get_value(self, key: str, default: object = None) -> object:
46 return False
47
48
49def _disable_voice_control(player: YandexStationPlayer) -> None:
50 player._config = _StubConfig() # type: ignore[assignment]
51
52
53# ââ External playback: physical pause detection ââââââââââââââââââââ
54
55
56def test_physical_pause_during_external_playback_sets_paused() -> None:
57 """Pressing pause on the speaker must flip MA state to PAUSED."""
58 player = _make_player()
59 _disable_voice_control(player)
60 player._external_playing = True
61 player._external_play_confirmed = True # stream has been observed playing
62 player._attr_playback_state = PlaybackState.PLAYING
63
64 player._update_playback_state(playing=False, alice_state="IDLE")
65
66 assert player._attr_playback_state == PlaybackState.PAUSED
67 assert player._external_playing is False
68 assert player._external_media is None
69 assert player._external_play_confirmed is False
70 assert player._needs_replay is True
71
72
73def test_external_startup_window_stays_playing_until_confirmed() -> None:
74 """
75 Before the station starts fetching the stream, playing=False is expected.
76
77 We must not misread that initial pre-stream window as a physical pause.
78 """
79 player = _make_player()
80 _disable_voice_control(player)
81 player._external_playing = True
82 player._external_play_confirmed = False # station hasn't confirmed yet
83 player._attr_playback_state = PlaybackState.PLAYING
84
85 player._update_playback_state(playing=False, alice_state="IDLE")
86
87 assert player._attr_playback_state == PlaybackState.PLAYING
88 assert player._external_playing is True
89 assert player._needs_replay is False
90
91
92def test_external_playing_true_sets_confirmed() -> None:
93 """The first playing=True update during external playback flips confirmed."""
94 player = _make_player()
95 _disable_voice_control(player)
96 player._external_playing = True
97 player._external_play_confirmed = False
98
99 player._update_playback_state(playing=True, alice_state="IDLE")
100
101 assert player._external_play_confirmed is True
102 assert player._attr_playback_state == PlaybackState.PLAYING
103 assert player._attr_powered is True
104
105
106def test_physical_pause_cancels_pending_voice_resume_task() -> None:
107 """
108 A pending auto-resume from a prior voice interaction must be cancelled.
109
110 Without this, the task could wake up after the user physically paused and
111 resume MA queue playback unexpectedly.
112 """
113 cancelled = False
114
115 class _FakeTask:
116 def cancel(self) -> None:
117 nonlocal cancelled
118 cancelled = True
119
120 player = _make_player()
121 _disable_voice_control(player)
122 player._external_playing = True
123 player._external_play_confirmed = True
124 player._voice_resume_task = _FakeTask() # type: ignore[assignment]
125
126 player._update_playback_state(playing=False, alice_state="IDLE")
127
128 assert cancelled is True
129 assert player._voice_resume_task is None
130 assert player._attr_playback_state == PlaybackState.PAUSED
131
132
133def test_physical_pause_ignored_during_voice_interaction() -> None:
134 """
135 Alice speaking/listening must not be treated as a physical pause.
136
137 Voice control is disabled here, so the fallthrough branch must keep PLAYING
138 (voice handling itself is tested elsewhere).
139 """
140 player = _make_player()
141 _disable_voice_control(player)
142 player._external_playing = True
143 player._external_play_confirmed = True
144
145 player._update_playback_state(playing=False, alice_state="LISTENING")
146
147 # With voice control disabled, non-IDLE alice during external playback
148 # falls through to the startup/pause branch. Since alice != IDLE, we skip
149 # the physical-pause branch â state stays optimistically PLAYING.
150 assert player._attr_playback_state == PlaybackState.PLAYING
151 assert player._external_playing is True
152