/
/
/
1"""Tests for the HEOS player."""
2
3from __future__ import annotations
4
5from datetime import UTC, datetime
6from typing import cast
7from unittest.mock import AsyncMock, MagicMock
8
9from music_assistant_models.enums import MediaType
10from pyheos import PlayState as HeosPlayState
11from pyheos import const as heos_const
12
13from music_assistant.models.player import PlayerMedia
14from music_assistant.providers.heos.player import HeosPlayer
15
16
17def _url_stream_now_playing() -> MagicMock:
18 """HEOS now-playing for a generic URL stream it cannot parse (MA's own source)."""
19 now_playing = MagicMock()
20 now_playing.source_id = heos_const.MUSIC_SOURCE_LOCAL_MUSIC
21 now_playing.type = "song"
22 now_playing.song = "Url Stream"
23 now_playing.album = "Url Stream"
24 now_playing.artist = "Url Stream"
25 now_playing.image_url = ""
26 now_playing.media_id = "1"
27 now_playing.album_id = "1"
28 now_playing.current_position = None
29 now_playing.current_position_updated = None
30 now_playing.duration = None
31 return now_playing
32
33
34def _external_now_playing(current_position: int | None, duration: int | None = 245000) -> MagicMock:
35 """HEOS now-playing for an external source, with millisecond progress values."""
36 now_playing = _url_stream_now_playing()
37 now_playing.source_id = 9999
38 now_playing.song = "External Track"
39 now_playing.current_position = current_position
40 now_playing.current_position_updated = datetime(2026, 1, 1, tzinfo=UTC)
41 now_playing.duration = duration
42 return now_playing
43
44
45def _make_player(now_playing: MagicMock) -> HeosPlayer:
46 """Build a HeosPlayer backed by a mocked device/controller."""
47 provider = MagicMock()
48 provider._heos_queue = MagicMock()
49 device = MagicMock()
50 device.player_id = "1"
51 device.name = "Kitchen"
52 device.heos = MagicMock()
53 device.state = HeosPlayState.PLAY
54 device.now_playing_media = now_playing
55 return HeosPlayer(provider, device)
56
57
58def test_url_stream_now_playing_preserves_ma_media_while_ma_controls() -> None:
59 """
60 Preserve MA's current_media against a HEOS "Url Stream" report.
61
62 HEOS cannot parse metadata from the generic URL stream MA serves, so it
63 reports ``Url Stream``. When MA controls playback that report must be
64 ignored (even if ``active_source`` is momentarily stale from the
65 ``play_url`` race) so MA's own, correct metadata is preserved.
66
67 See https://github.com/music-assistant/support/issues/5614
68 """
69 player = _make_player(_url_stream_now_playing())
70 correct_media = PlayerMedia(
71 uri="http://ma/stream/foo",
72 media_type=MediaType.TRACK,
73 title="Real Track",
74 artist="Real Artist",
75 album="Real Album",
76 )
77 player._attr_current_media = correct_media
78 player._ma_controls_playback = True
79 # Stale: a previous external source is still the recorded active source.
80 player._attr_active_source = "external_radio"
81
82 player._update_player_current_media()
83
84 # MA's correct media is preserved; the bogus "Url Stream" did not clobber it.
85 assert player._attr_current_media is correct_media
86 assert player._attr_current_media.title == "Real Track"
87 # Playback control was not latched away to the phantom external source.
88 assert player._ma_controls_playback is True
89
90
91def test_external_source_now_playing_still_updates_media() -> None:
92 """A genuinely external (non-local) source must still update current_media."""
93 now_playing = _url_stream_now_playing()
94 now_playing.source_id = 9999 # an external source, not MA's local stream
95 now_playing.song = "External Track"
96 now_playing.artist = "External Artist"
97 player = _make_player(now_playing)
98 player._attr_current_media = PlayerMedia(
99 uri="http://ma/stream/foo", media_type=MediaType.TRACK, title="MA Track"
100 )
101 player._ma_controls_playback = True
102 player._attr_active_source = player.player_id
103
104 player._update_player_current_media()
105
106 assert player._attr_current_media.title == "External Track"
107
108
109def test_external_source_is_ignored_during_ma_playback_transition() -> None:
110 """Suppress external-source media updates while MA playback is starting."""
111 player = _make_player(_external_now_playing(None))
112 player._ma_playback_starting = True
113 player._ma_controls_playback = True
114 current_media = PlayerMedia(
115 uri="http://ma/stream/foo", media_type=MediaType.TRACK, title="MA Track"
116 )
117 player._attr_current_media = current_media
118
119 player._update_player_current_media()
120
121 assert player._ma_playback_starting is True
122 assert player._ma_controls_playback is True
123 assert player._attr_current_media.title == "MA Track"
124
125 player._finish_ma_playback_transition()
126
127 assert player._attr_current_media.title == "External Track"
128 assert player._ma_controls_playback is False
129
130
131async def test_new_ma_playback_cancels_previous_transition() -> None:
132 """Keep a new MA playback transition protected from the previous timer."""
133 player = _make_player(_url_stream_now_playing())
134 mass = cast("MagicMock", player.mass)
135 device = cast("MagicMock", player._device)
136 transition_pending = True
137
138 def cancel_timer(task_id: str) -> None:
139 nonlocal transition_pending
140 assert task_id == f"heos_playback_transition_{player.player_id}"
141 transition_pending = False
142
143 async def play_url(_: str) -> None:
144 if transition_pending:
145 player._finish_ma_playback_transition()
146
147 mass.cancel_timer.side_effect = cancel_timer
148 mass.streams.resolve_stream_url = AsyncMock(return_value="http://ma/stream/new")
149 device.play_url = AsyncMock(side_effect=play_url)
150 player._ma_playback_starting = True
151
152 await player.play_media(
153 PlayerMedia(
154 uri="library://track/2",
155 media_type=MediaType.TRACK,
156 title="New MA Track",
157 )
158 )
159
160 assert player._ma_playback_starting is True
161 mass.cancel_timer.assert_called_once_with(f"heos_playback_transition_{player.player_id}")
162
163
164async def test_select_source_ends_ma_playback_transition() -> None:
165 """Allow an explicit source change to update HEOS media immediately."""
166 player = _make_player(_url_stream_now_playing())
167 mass = cast("MagicMock", player.mass)
168 device = cast("MagicMock", player._device)
169 player._ma_playback_starting = True
170
171 async def play_input_source(_: str) -> None:
172 assert player._ma_playback_starting is False
173
174 device.play_input_source = AsyncMock(side_effect=play_input_source)
175
176 await player.select_source("aux_in_1")
177
178 mass.cancel_timer.assert_called_once_with(f"heos_playback_transition_{player.player_id}")
179
180
181async def test_player_unload_cancels_playback_transition() -> None:
182 """Cancel the playback transition when the player unloads."""
183 player = _make_player(_url_stream_now_playing())
184 mass = cast("MagicMock", player.mass)
185 player._ma_playback_starting = True
186
187 await player.on_unload()
188
189 mass.cancel_timer.assert_any_call(f"heos_playback_transition_{player.player_id}")
190 assert player._ma_playback_starting is False
191
192
193def test_media_position_and_duration_reported_in_seconds() -> None:
194 """HEOS reports milliseconds; the media must carry seconds."""
195 player = _make_player(_external_now_playing(113000))
196 player._attr_active_source = "9999"
197
198 player._update_player_current_media()
199
200 assert player._attr_current_media is not None
201 assert player._attr_current_media.elapsed_time == 113
202 assert player._attr_current_media.duration == 245
203
204
205def test_media_and_player_position_agree() -> None:
206 """The media-level and player-level positions must use the same unit."""
207 player = _make_player(_external_now_playing(113000))
208 player._attr_active_source = "9999"
209
210 player.set_dynamic_attributes(update_media=True)
211
212 assert player._attr_current_media is not None
213 assert player._attr_current_media.elapsed_time == player._attr_elapsed_time == 113
214
215
216def test_position_zero_is_reported_not_dropped() -> None:
217 """Position 0 is a real position at the start of a track, not a missing one."""
218 player = _make_player(_external_now_playing(0))
219 player._attr_active_source = "9999"
220
221 player.set_dynamic_attributes(update_media=True)
222
223 assert player._attr_current_media is not None
224 assert player._attr_current_media.elapsed_time == 0
225 assert player._attr_elapsed_time == 0
226
227
228def test_missing_progress_stays_none() -> None:
229 """Without progress info, no position or duration is reported."""
230 player = _make_player(_external_now_playing(None, duration=None))
231 player._attr_active_source = "9999"
232
233 player.set_dynamic_attributes(update_media=True)
234
235 assert player._attr_current_media is not None
236 assert player._attr_current_media.elapsed_time is None
237 assert player._attr_current_media.duration is None
238 assert player._attr_elapsed_time is None
239