/
/
/
1"""Tests for the AmpliPi player provider."""
2
3from __future__ import annotations
4
5import asyncio
6from types import SimpleNamespace
7from typing import cast
8from unittest.mock import AsyncMock, MagicMock
9
10import pytest
11from music_assistant_models.errors import PlayerCommandFailed, SetupFailedError
12from pyamplipi.error import AmpliPiUnreachableError
13
14from music_assistant.providers.amplipi import setup, setup_flow
15from music_assistant.providers.amplipi.constants import CONF_HOST, MA_STREAM_NAME, MA_STREAM_TYPE
16from music_assistant.providers.amplipi.provider import AmpliPiPlayerProvider
17
18
19def _zone(zone_id: int, disabled: bool = False) -> SimpleNamespace:
20 """Build a lightweight stand-in for a pyamplipi Zone."""
21 return SimpleNamespace(
22 id=zone_id, name=f"Zone {zone_id}", source_id=-1, disabled=disabled, mute=False, vol=-30
23 )
24
25
26def _provider() -> AmpliPiPlayerProvider:
27 """Build a provider instance without running the (heavy) base __init__."""
28 prov = AmpliPiPlayerProvider.__new__(AmpliPiPlayerProvider)
29 prov.config = MagicMock()
30 prov.config.instance_id = "amplipi_test"
31 prov.logger = MagicMock()
32 prov.api = MagicMock()
33 prov.mass = MagicMock()
34 prov._players = {}
35 prov._ma_streams = {}
36 prov._streams = []
37 prov._stream_locks = {}
38 return prov
39
40
41class TestHandleAsyncInit:
42 """Test provider connection/initialisation."""
43
44 async def test_builds_endpoint_and_connects(self, monkeypatch: pytest.MonkeyPatch) -> None:
45 """A bare host should be turned into a full http://<host>/api endpoint."""
46 prov = AmpliPiPlayerProvider.__new__(AmpliPiPlayerProvider)
47 prov.config = MagicMock()
48 prov.config.get_value.return_value = "amplipi.local"
49 prov.mass = MagicMock()
50 # setup_data is unset here, so get_setup_value falls through to config.get_value
51 prov.config.values = {}
52 prov.mass.config.get.return_value = None
53 prov.mass.config.get_raw_provider_config_value.return_value = None
54 fake_api = MagicMock()
55 fake_api.get_status = AsyncMock(return_value="STATUS")
56 created: dict[str, object] = {}
57
58 def fake_ctor(**kwargs: object) -> MagicMock:
59 created.update(kwargs)
60 return fake_api
61
62 monkeypatch.setattr("music_assistant.providers.amplipi.provider.AmpliPi", fake_ctor)
63 await prov.handle_async_init()
64
65 assert created["endpoint"] == "http://amplipi.local/api"
66 assert prov._status == "STATUS"
67 assert prov._players == {}
68 assert prov._ma_streams == {}
69
70 async def test_full_url_host_passed_through(self, monkeypatch: pytest.MonkeyPatch) -> None:
71 """A host already containing a scheme is used verbatim as the endpoint."""
72 prov = AmpliPiPlayerProvider.__new__(AmpliPiPlayerProvider)
73 prov.config = MagicMock()
74 prov.config.get_value.return_value = "http://1.2.3.4/api"
75 prov.mass = MagicMock()
76 # setup_data is unset here, so get_setup_value falls through to config.get_value
77 prov.config.values = {}
78 prov.mass.config.get.return_value = None
79 prov.mass.config.get_raw_provider_config_value.return_value = None
80 fake_api = MagicMock()
81 fake_api.get_status = AsyncMock(return_value="STATUS")
82 created: dict[str, object] = {}
83 monkeypatch.setattr(
84 "music_assistant.providers.amplipi.provider.AmpliPi",
85 lambda **kwargs: (created.update(kwargs), fake_api)[1],
86 )
87 await prov.handle_async_init()
88 assert created["endpoint"] == "http://1.2.3.4/api"
89
90 async def test_schemed_host_without_path_gets_api(
91 self, monkeypatch: pytest.MonkeyPatch
92 ) -> None:
93 """A schemed host with no path (incl. https) keeps its scheme and gains /api."""
94 prov = AmpliPiPlayerProvider.__new__(AmpliPiPlayerProvider)
95 prov.config = MagicMock()
96 prov.config.get_value.return_value = "https://amplipi.local/"
97 prov.mass = MagicMock()
98 # setup_data is unset here, so get_setup_value falls through to config.get_value
99 prov.config.values = {}
100 prov.mass.config.get.return_value = None
101 prov.mass.config.get_raw_provider_config_value.return_value = None
102 fake_api = MagicMock()
103 fake_api.get_status = AsyncMock(return_value="STATUS")
104 created: dict[str, object] = {}
105 monkeypatch.setattr(
106 "music_assistant.providers.amplipi.provider.AmpliPi",
107 lambda **kwargs: (created.update(kwargs), fake_api)[1],
108 )
109 await prov.handle_async_init()
110 assert created["endpoint"] == "https://amplipi.local/api"
111
112 async def test_http_prefixed_bare_host_gets_scheme(
113 self, monkeypatch: pytest.MonkeyPatch
114 ) -> None:
115 """A bare host that merely starts with 'http' must still get a scheme and /api."""
116 prov = AmpliPiPlayerProvider.__new__(AmpliPiPlayerProvider)
117 prov.config = MagicMock()
118 prov.config.get_value.return_value = "http-livingroom.local"
119 prov.mass = MagicMock()
120 # setup_data is unset here, so get_setup_value falls through to config.get_value
121 prov.config.values = {}
122 prov.mass.config.get.return_value = None
123 prov.mass.config.get_raw_provider_config_value.return_value = None
124 fake_api = MagicMock()
125 fake_api.get_status = AsyncMock(return_value="STATUS")
126 created: dict[str, object] = {}
127 monkeypatch.setattr(
128 "music_assistant.providers.amplipi.provider.AmpliPi",
129 lambda **kwargs: (created.update(kwargs), fake_api)[1],
130 )
131 await prov.handle_async_init()
132 assert created["endpoint"] == "http://http-livingroom.local/api"
133
134 async def test_connection_failure_raises_setup_failed(
135 self, monkeypatch: pytest.MonkeyPatch
136 ) -> None:
137 """A failure to fetch the initial status must raise SetupFailedError."""
138 prov = AmpliPiPlayerProvider.__new__(AmpliPiPlayerProvider)
139 prov.config = MagicMock()
140 prov.config.get_value.return_value = "amplipi.local"
141 prov.mass = MagicMock()
142 # setup_data is unset here, so get_setup_value falls through to config.get_value
143 prov.config.values = {}
144 prov.mass.config.get.return_value = None
145 prov.mass.config.get_raw_provider_config_value.return_value = None
146 fake_api = MagicMock()
147 fake_api.get_status = AsyncMock(side_effect=AmpliPiUnreachableError("no route"))
148 monkeypatch.setattr(
149 "music_assistant.providers.amplipi.provider.AmpliPi", lambda **_kwargs: fake_api
150 )
151 with pytest.raises(SetupFailedError):
152 await prov.handle_async_init()
153
154
155class TestLifecycle:
156 """Test loaded_in_mass / unload."""
157
158 async def test_loaded_in_mass_adopts_streams(self) -> None:
159 """loaded_in_mass should fetch streams, re-adopt MA streams, discover and poll."""
160 prov = _provider()
161 prov.api.get_streams = AsyncMock(
162 return_value=[
163 SimpleNamespace(id=7, name=f"{MA_STREAM_NAME} 0", type=MA_STREAM_TYPE),
164 SimpleNamespace(id=8, name=f"{MA_STREAM_NAME} 2", type=MA_STREAM_TYPE),
165 SimpleNamespace(id=9, name="Groove Salad", type=MA_STREAM_TYPE),
166 SimpleNamespace(id=None, name=f"{MA_STREAM_NAME} 3", type=MA_STREAM_TYPE),
167 SimpleNamespace(id=10, name=f"{MA_STREAM_NAME} x", type=MA_STREAM_TYPE),
168 SimpleNamespace(id=11, name=f"{MA_STREAM_NAME} Radio", type="fileplayer"),
169 ]
170 )
171 prov.discover_players = AsyncMock() # type: ignore[method-assign]
172 task_obj = object()
173 prov.mass.create_task = MagicMock(return_value=task_obj) # type: ignore[method-assign]
174
175 await prov.loaded_in_mass()
176
177 prov.discover_players.assert_awaited_once()
178 assert prov._poll_task is task_obj
179 # only well-formed MA streams with an id are adopted, keyed by their source id
180 assert prov._ma_streams == {0: 7, 2: 8}
181
182 async def test_unload_reload_keeps_streams(self) -> None:
183 """A plain reload cancels (and awaits) polling and unregisters players, keeping streams."""
184 prov = _provider()
185
186 async def _poll() -> None:
187 await asyncio.sleep(3600)
188
189 task = asyncio.ensure_future(_poll())
190 prov._poll_task = task
191 player = MagicMock()
192 player.player_id = "amplipi_test_zone_0"
193 prov._players = {0: player}
194 prov.mass.players.unregister = AsyncMock() # type: ignore[method-assign]
195 prov.api.delete_stream = AsyncMock()
196
197 await prov.unload()
198
199 # the poll task is cancelled and awaited to completion
200 assert task.cancelled()
201 prov.mass.players.unregister.assert_awaited_once_with("amplipi_test_zone_0")
202 assert prov._players == {}
203 prov.api.delete_stream.assert_not_awaited()
204
205 async def test_unload_removed_deletes_streams(self) -> None:
206 """When the provider is removed, its MA streams are deleted from the controller."""
207 prov = _provider()
208 prov._poll_task = None
209 prov._players = {}
210 prov.api.get_streams = AsyncMock(
211 return_value=[SimpleNamespace(id=7, name=f"{MA_STREAM_NAME} 0", type=MA_STREAM_TYPE)]
212 )
213 prov.api.delete_stream = AsyncMock()
214
215 await prov.unload(is_removed=True)
216
217 prov.api.delete_stream.assert_awaited_once_with(7)
218
219
220class TestEnsureStream:
221 """Test the per-source MA internetradio stream management."""
222
223 async def test_reuses_existing_stream(self) -> None:
224 """An already-created stream should be re-pointed at the new url, not recreated."""
225 prov = _provider()
226 prov._ma_streams = {0: 42}
227 prov.api.set_stream = AsyncMock()
228 prov.api.create_stream = AsyncMock()
229
230 result = await prov.ensure_stream(0, "http://ma/new.flac")
231
232 assert result == 42
233 prov.api.set_stream.assert_awaited_once()
234 prov.api.create_stream.assert_not_awaited()
235
236 async def test_creates_stream_and_looks_up_id(self) -> None:
237 """A new stream is created, then its id resolved from the streams list."""
238 prov = _provider()
239 prov.api.create_stream = AsyncMock()
240 prov.api.get_streams = AsyncMock(
241 return_value=[
242 SimpleNamespace(id=7, name=f"{MA_STREAM_NAME} 1", type=MA_STREAM_TYPE),
243 SimpleNamespace(id=8, name="Groove Salad", type=MA_STREAM_TYPE),
244 ]
245 )
246 result = await prov.ensure_stream(1, "http://ma/x.flac")
247
248 assert result == 7
249 assert prov._ma_streams[1] == 7
250
251 async def test_create_failure_raises(self) -> None:
252 """If the new stream cannot be found after creation, raise PlayerCommandFailed."""
253 prov = _provider()
254 prov.api.create_stream = AsyncMock()
255 prov.api.get_streams = AsyncMock(return_value=[])
256 with pytest.raises(PlayerCommandFailed):
257 await prov.ensure_stream(2, "http://ma/x.flac")
258
259
260class TestDiscoverAndHelpers:
261 """Test discover_players and small helpers."""
262
263 async def test_discover_registers_enabled_zones(self, monkeypatch: pytest.MonkeyPatch) -> None:
264 """Disabled zones and zones without an id are skipped; the rest are registered."""
265 prov = _provider()
266 prov._status = SimpleNamespace(
267 zones=[
268 _zone(0),
269 _zone(1, disabled=True),
270 SimpleNamespace(id=None, disabled=False),
271 _zone(2),
272 ],
273 sources=[],
274 )
275 prov.mass.players.register_or_update = AsyncMock() # type: ignore[method-assign]
276 monkeypatch.setattr(
277 "music_assistant.providers.amplipi.provider.AmpliPiZonePlayer",
278 lambda _provider, zone_id: SimpleNamespace(
279 player_id=f"z{zone_id}", update_from_status=MagicMock()
280 ),
281 )
282 await prov.discover_players()
283
284 assert set(prov._players) == {0, 2}
285 assert prov.mass.players.register_or_update.await_count == 2
286
287 async def test_discover_skips_already_registered(self, monkeypatch: pytest.MonkeyPatch) -> None:
288 """A zone already in _players is not registered again."""
289 prov = _provider()
290 prov._players = {0: MagicMock()}
291 prov._status = SimpleNamespace(zones=[_zone(0)], sources=[])
292 prov.mass.players.register_or_update = AsyncMock() # type: ignore[method-assign]
293 monkeypatch.setattr(
294 "music_assistant.providers.amplipi.provider.AmpliPiZonePlayer",
295 lambda _provider, _zone_id: SimpleNamespace(player_id="z0"),
296 )
297 await prov.discover_players()
298 prov.mass.players.register_or_update.assert_not_awaited()
299
300 def test_status_property(self) -> None:
301 """The status property returns the last polled status."""
302 prov = _provider()
303 prov._status = "STATUS"
304 assert prov.status == "STATUS"
305
306 def test_zone_id_for(self) -> None:
307 """zone_id_for maps a player_id back to its zone id, or None when unknown."""
308 prov = _provider()
309 prov._players = {
310 0: SimpleNamespace(player_id="a"), # type: ignore[dict-item]
311 1: SimpleNamespace(player_id="b"), # type: ignore[dict-item]
312 }
313 assert prov.zone_id_for("b") == 1
314 assert prov.zone_id_for("missing") is None
315
316 def test_selectable_streams_filters(self) -> None:
317 """Native streams + RCA inputs are selectable; MA streams and fileplayer are excluded."""
318 prov = _provider()
319 prov._streams = [
320 SimpleNamespace(id=996, name="Input 1", type="rca"),
321 SimpleNamespace(id=1000, name="Groove Salad", type="internetradio"),
322 SimpleNamespace(id=1008, name="External Media", type="fileplayer"),
323 SimpleNamespace(id=1009, name=f"{MA_STREAM_NAME} 3", type="internetradio"),
324 SimpleNamespace(id=None, name="No id", type="airplay"),
325 # a user stream that merely starts with the MA name prefix stays selectable
326 SimpleNamespace(id=1010, name=f"{MA_STREAM_NAME} Radio", type="internetradio"),
327 ]
328 assert {s.id for s in prov.selectable_streams()} == {996, 1000, 1010}
329
330
331class TestPollLoop:
332 """Test the polling loop's success and error handling."""
333
334 async def test_poll_success_updates_players(self, monkeypatch: pytest.MonkeyPatch) -> None:
335 """A successful poll refreshes status, streams and every player."""
336 prov = _provider()
337 new_status = SimpleNamespace(zones=[_zone(0)], sources=[])
338 prov.api.get_status = AsyncMock(return_value=new_status)
339 prov.api.get_streams = AsyncMock(return_value=[])
340 prov.discover_players = AsyncMock() # type: ignore[method-assign]
341 player = MagicMock()
342 prov._players = {0: player}
343 monkeypatch.setattr(
344 "music_assistant.providers.amplipi.provider.asyncio.sleep",
345 AsyncMock(side_effect=[None, asyncio.CancelledError()]),
346 )
347 with pytest.raises(asyncio.CancelledError):
348 await prov._poll_loop()
349
350 assert prov._status == new_status
351 prov.discover_players.assert_awaited()
352 player.update_from_status.assert_called_with(new_status)
353
354 async def test_poll_propagates_cancellation(self, monkeypatch: pytest.MonkeyPatch) -> None:
355 """A CancelledError raised by the status fetch must propagate, not be swallowed."""
356 prov = _provider()
357 prov.api.get_status = AsyncMock(side_effect=asyncio.CancelledError())
358 prov._players = {}
359 monkeypatch.setattr(
360 "music_assistant.providers.amplipi.provider.asyncio.sleep",
361 AsyncMock(side_effect=[None]),
362 )
363 with pytest.raises(asyncio.CancelledError):
364 await prov._poll_loop()
365 cast("MagicMock", prov.logger).warning.assert_not_called()
366
367 async def test_poll_error_marks_unavailable(self, monkeypatch: pytest.MonkeyPatch) -> None:
368 """A failed poll marks every player unavailable and continues."""
369 prov = _provider()
370 prov.api.get_status = AsyncMock(side_effect=AmpliPiUnreachableError("boom"))
371 player = MagicMock()
372 prov._players = {0: player}
373 monkeypatch.setattr(
374 "music_assistant.providers.amplipi.provider.asyncio.sleep",
375 AsyncMock(side_effect=[None, asyncio.CancelledError()]),
376 )
377 with pytest.raises(asyncio.CancelledError):
378 await prov._poll_loop()
379
380 player.set_unavailable.assert_called_once()
381 cast("MagicMock", prov.logger).warning.assert_called()
382
383
384class TestRemoveMaStreams:
385 """Test cleanup of MA-created streams."""
386
387 async def test_removes_only_ma_streams(self) -> None:
388 """Only streams named with the MA prefix, of MA_STREAM_TYPE, with an id, are deleted."""
389 prov = _provider()
390 prov._ma_streams = {0: 7}
391 prov.api.get_streams = AsyncMock(
392 return_value=[
393 SimpleNamespace(id=7, name=f"{MA_STREAM_NAME} 1", type=MA_STREAM_TYPE),
394 SimpleNamespace(id=8, name="Groove Salad", type=MA_STREAM_TYPE),
395 SimpleNamespace(id=None, name=f"{MA_STREAM_NAME} 2", type=MA_STREAM_TYPE),
396 # a user stream that merely starts with the same name prefix must survive
397 SimpleNamespace(id=9, name=f"{MA_STREAM_NAME} Radio", type="fileplayer"),
398 ]
399 )
400 prov.api.delete_stream = AsyncMock()
401
402 await prov._remove_ma_streams()
403
404 prov.api.delete_stream.assert_awaited_once_with(7)
405 assert prov._ma_streams == {}
406
407
408class TestModuleEntryPoints:
409 """Test the provider module setup / config entry hooks."""
410
411 async def test_setup_returns_provider(self, monkeypatch: pytest.MonkeyPatch) -> None:
412 """setup() should construct and return an AmpliPiPlayerProvider."""
413 monkeypatch.setattr(
414 "music_assistant.providers.amplipi.AmpliPiPlayerProvider",
415 lambda *_args: "PROVIDER",
416 )
417 result = await setup(MagicMock(), MagicMock(), MagicMock())
418 assert result == "PROVIDER" # type: ignore[comparison-overlap]
419
420 async def test_get_config_entries_has_no_setup_entries(self) -> None:
421 """The host moved to the setup flow, so the options entries no longer expose it."""
422 entries = await _provider().get_config_entries()
423 assert all(e.key != CONF_HOST for e in entries)
424
425 async def test_setup_flow_exposes_host(self) -> None:
426 """The setup flow must collect a required Host entry."""
427 host = next(e for e in setup_flow._ENTRIES if e.key == CONF_HOST)
428 assert host.required is True
429