/
/
/
1"""Tests for telling Music Assistant playback apart from external playback."""
2
3from __future__ import annotations
4
5import logging
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock
8
9from music_assistant_models.enums import PlaybackState
10from music_assistant_models.player import PlayerMedia
11
12from music_assistant.constants import ATTR_ANNOUNCEMENT_IN_PROGRESS
13from music_assistant.providers.hass_players.player import HomeAssistantPlayer
14
15BASE_URL = "http://10.0.0.5:8097"
16PLAYER_ID = "media_player.bedroom"
17
18
19def _make_player() -> HomeAssistantPlayer:
20 """Create a HomeAssistantPlayer with mocked dependencies."""
21 player = HomeAssistantPlayer.__new__(HomeAssistantPlayer)
22 mass = MagicMock()
23 mass.closing = False
24 mass.streams.base_url = BASE_URL
25 mass.streams.resolve_stream_url = AsyncMock(return_value=f"{BASE_URL}/flow/s1/q1/i1/x.mp3")
26 player.mass = mass
27 player.logger = logging.getLogger("test.hass_players.player")
28 player._player_id = PLAYER_ID
29 player._provider = MagicMock()
30 player._provider.mass = mass
31 player._extra_data = {}
32 player.hass = MagicMock()
33 player.hass.call_service = AsyncMock()
34 player._hass_attributes = {}
35 player._attr_supported_features = set()
36 player._attr_source_list = []
37 player._attr_group_members = []
38 player._attr_playback_state = PlaybackState.IDLE
39 player._attr_active_source = None
40 player._attr_current_media = None
41 player._ma_playback_active = False
42 player._ma_playback_started = False
43 player._reports_stream_url = False
44 player.update_state = MagicMock() # type: ignore[misc, method-assign]
45 return player
46
47
48def _external_attributes(content_id: str = "spotify:track:42") -> dict[str, Any]:
49 """Return HA state attributes for content MA did not hand the entity."""
50 return {
51 "media_content_id": content_id,
52 "media_title": "Some Song",
53 "media_artist": "Some Artist",
54 }
55
56
57async def test_ma_playback_detected_when_entity_hides_the_stream_url() -> None:
58 """An entity that reports its own metadata instead of our URL is still MA playback."""
59 player = _make_player()
60 media = PlayerMedia(uri="library://track/1", title="MA Track")
61 player._ma_playback_active = True
62 player._attr_current_media = media
63 player._attr_playback_state = PlaybackState.PLAYING
64
65 player._update_attributes(_external_attributes())
66
67 assert player._attr_active_source is None
68 # the queue controller provides the media, so the entity may not overwrite it
69 assert player._attr_current_media is media
70
71
72async def test_external_playback_detected_without_ma_session() -> None:
73 """Playback that MA did not start is reported as an external source."""
74 player = _make_player()
75 player._attr_playback_state = PlaybackState.PLAYING
76
77 player._update_attributes(_external_attributes())
78
79 assert player._attr_active_source == "External"
80 assert player._attr_current_media is not None
81 assert player._attr_current_media.title == "Some Song"
82
83
84async def test_entity_echoing_stream_url_still_detects_takeover() -> None:
85 """An entity that reports our stream URL is judged by it, also mid-session."""
86 player = _make_player()
87 player._ma_playback_active = True
88 player._attr_playback_state = PlaybackState.PLAYING
89
90 player._update_attributes(
91 {"media_content_id": f"{BASE_URL}/flow/s1/q1/i1/x.mp3", "media_title": "MA Track"}
92 )
93 assert player._attr_active_source is None
94 assert player._reports_stream_url is True
95
96 # another app takes the speaker over while MA still considers its session open
97 player._update_attributes(_external_attributes())
98
99 assert player._attr_active_source == "External"
100
101
102async def test_idle_entity_ends_the_ma_session() -> None:
103 """Playback that starts after the entity went idle is external again."""
104 player = _make_player()
105
106 await player.play_media(PlayerMedia(uri="library://track/1", title="MA Track"))
107 player.update_from_compressed_state({"s": "playing"})
108 player.update_from_compressed_state({"s": "idle"})
109 assert player._ma_playback_active is False
110
111 player.update_from_compressed_state({"s": "playing", "a": _external_attributes()})
112
113 assert player._attr_active_source == "External"
114
115
116async def test_play_media_starts_the_ma_session() -> None:
117 """A play command from MA marks the session as ours."""
118 player = _make_player()
119
120 await player.play_media(PlayerMedia(uri="library://track/1", title="MA Track"))
121
122 assert player._ma_playback_active is True
123
124
125async def test_play_media_takes_over_from_an_external_source() -> None:
126 """Starting MA playback drops the external source without waiting for the entity."""
127 player = _make_player()
128 player._attr_playback_state = PlaybackState.PLAYING
129 player._update_attributes(_external_attributes())
130 assert player._attr_active_source == "External"
131
132 # the entity may never report an attribute change to announce the handover
133 await player.play_media(PlayerMedia(uri="library://track/1", title="MA Track"))
134
135 assert player._attr_active_source is None
136
137
138async def test_late_idle_of_previous_session_is_ignored() -> None:
139 """An entity reporting the stop of the previous track late keeps our session ours."""
140 player = _make_player()
141 player._ma_playback_active = True
142 player._ma_playback_started = True
143 player._attr_playback_state = PlaybackState.PLAYING
144
145 # switching tracks stops the entity first, so its idle can arrive after our play
146 await player.play_media(PlayerMedia(uri="library://track/2", title="Next Track"))
147 player.update_from_compressed_state({"s": "idle"})
148 player.update_from_compressed_state({"s": "playing", "a": _external_attributes()})
149
150 assert player._ma_playback_active is True
151 assert player._attr_active_source is None
152
153
154async def test_power_on_state_does_not_end_the_ma_session() -> None:
155 """An entity that reports powering on before it plays keeps our session ours."""
156 player = _make_player()
157
158 await player.play_media(PlayerMedia(uri="library://track/1", title="MA Track"))
159 player.update_from_compressed_state({"s": "on"})
160 player.update_from_compressed_state({"s": "playing", "a": _external_attributes()})
161
162 assert player._attr_active_source is None
163
164
165async def test_announcement_does_not_disturb_source_detection() -> None:
166 """An announcement neither ends our session nor proves the entity reports our URL."""
167 player = _make_player()
168 player._attr_playback_state = PlaybackState.PLAYING
169 player._update_attributes(_external_attributes())
170 assert player._attr_active_source == "External"
171
172 player.extra_data[ATTR_ANNOUNCEMENT_IN_PROGRESS] = True
173 player.update_from_compressed_state(
174 {
175 "s": "playing",
176 "a": {"media_content_id": f"{BASE_URL}/announcement/{PLAYER_ID}.mp3"},
177 }
178 )
179 player.update_from_compressed_state({"s": "idle"})
180
181 assert player._reports_stream_url is False
182 assert player._attr_active_source == "External"
183
184 # an unrelated update after the announcement may not pick the announcement up either
185 player.extra_data[ATTR_ANNOUNCEMENT_IN_PROGRESS] = False
186 player.update_from_compressed_state({"s": "playing", "a": {"volume_level": 0.5}})
187
188 assert player._reports_stream_url is False
189
190
191async def test_stop_ends_the_ma_session() -> None:
192 """A stop command from MA releases the session."""
193 player = _make_player()
194 player._ma_playback_active = True
195
196 await player.stop()
197
198 assert player._ma_playback_active is False
199